Base Simulation Module

Base classes and components for discrete event simulation.

This module provides the fundamental building blocks for discrete event simulation, including event management, simulation clock, and protocol handling. It defines the base classes that specific simulation implementations will extend.

Notes

Key components: - Event: Represents discrete simulation events with timing and data - SimulationClock: Manages event scheduling and time progression - BaseSimulation: Abstract base class for all simulation implementations - ProtocolEvent: Protocol-specific event data structure - SimulationEnvironment: Global simulation state container

class simulation.base.ProtocolEvent(phase_type, action, parameters=<factory>, result=None)[source]

Bases: object

Protocol-specific event data.

Represents an event that is specific to a treatment protocol, containing information about the phase, action, and associated parameters.

Parameters:

  • phase_type (PhaseType) – Type of protocol phase this event belongs to (e.g., INITIAL, MAINTENANCE)

  • action (str) – Action to be performed (e.g., ‘inject’, ‘assess’, ‘adjust_interval’)

  • parameters (Dict[str, Any], optional) – Additional parameters for the action (default: {})

  • result (Optional[Dict[str, Any]], optional) – Results from executing the action (default: None)

Examples

>>> event = ProtocolEvent(
...     phase_type=PhaseType.MAINTENANCE,
...     action='inject',
...     parameters={'drug': 'eylea', 'dose': 2.0}
... )
phase_type: PhaseType
action: str
parameters: Dict[str, Any]
result: Optional[Dict[str, Any]] = None
__init__(phase_type, action, parameters=<factory>, result=None)
Parameters:

  • phase_type (PhaseType)

  • action (str)

  • parameters (Dict[str, Any])

  • result (Dict[str, Any] | None)

Return type:

None

class simulation.base.SimulationEnvironment(start_date)[source]

Bases: object

Global simulation environment.

Maintains global state and timing information for the simulation.

Parameters:

start_date (datetime) – Start date and time for the simulation

Variables:

  • current_time (datetime) – Current simulation time

  • global_state (Dict[str, Any]) – Dictionary storing global simulation state including: - resource_availability: Dict of resource counts - queue_lengths: Dict of patient queue lengths - statistics: Aggregated simulation metrics

Notes

The environment provides shared state across all simulation components and patients. It should be used for system-wide properties rather than patient-specific state.

__init__(start_date)[source]
Parameters:

start_date (datetime)

class simulation.base.Event(time, event_type, patient_id, data=<factory>, priority=1, protocol_event=None, phase=None, protocol=None)[source]

Bases: object

Simulation event.

Represents a discrete event in the simulation with timing, type, and associated data.

Parameters:

  • time (datetime) – When the event occurs

  • event_type (str) – Type of event (e.g., ‘arrival’, ‘treatment’, ‘assessment’)

  • patient_id (str) – ID of the patient this event relates to

  • data (Dict[str, Any], optional) – Additional event data (default: {})

  • priority (int, optional) – Event priority (lower numbers = higher priority) (default: 1)

  • protocol_event (Optional[ProtocolEvent], optional) – Associated protocol-specific event data (default: None)

  • phase (Optional[ProtocolPhase], optional) – Treatment phase this event belongs to (default: None)

  • protocol (Optional[TreatmentProtocol], optional) – Treatment protocol this event belongs to (default: None)

Variables:

_hash (int) – Internal hash value for event comparison

Notes

Events are comparable and hashable based on their time, type, and patient_id. This enables proper ordering in priority queues and event lists.

Examples

>>> event = Event(
...     time=datetime(2023, 1, 1),
...     event_type='treatment',
...     patient_id='pat123',
...     data={'drug': 'eylea', 'dose': 2.0}
... )
time: datetime
event_type: str
patient_id: str
data: Dict[str, Any]
priority: int = 1
protocol_event: Optional[ProtocolEvent] = None
phase: Optional[ProtocolPhase] = None
protocol: Optional[TreatmentProtocol] = None
classmethod create_protocol_event(time, patient_id, phase_type, action, data=None, priority=1, phase=None, protocol=None)[source]

Create a protocol-specific event with phase and protocol context.

Parameters:

  • time (datetime) – When the event occurs

  • patient_id (str) – ID of the patient

  • phase_type (PhaseType) – Type of protocol phase

  • action (str) – Action to be performed

  • data (Dict[str, Any], optional) – Additional event data

  • priority (int, optional) – Event priority

  • phase (Optional[ProtocolPhase], optional) – Treatment phase

  • protocol (Optional[TreatmentProtocol], optional) – Treatment protocol

Returns:

New protocol event

Return type:

Event

get_visit_type()[source]

Get visit type from phase if available.

Returns:

Visit type if phase is set, None otherwise

Return type:

Optional[VisitType]

get_required_actions()[source]

Get required actions from visit type.

Returns:

List of required actions for this visit type

Return type:

List[ActionType]

get_optional_actions()[source]

Get optional actions from visit type.

Returns:

List of optional actions for this visit type

Return type:

List[ActionType]

get_decisions()[source]

Get decisions from visit type.

Returns:

List of decisions required for this visit type

Return type:

List[DecisionType]

__init__(time, event_type, patient_id, data=<factory>, priority=1, protocol_event=None, phase=None, protocol=None)
Parameters:

  • time (datetime)

  • event_type (str)

  • patient_id (str)

  • data (Dict[str, Any])

  • priority (int)

  • protocol_event (ProtocolEvent | None)

  • phase (ProtocolPhase | None)

  • protocol (TreatmentProtocol | None)

Return type:

None

class simulation.base.SimulationClock(start_date)[source]

Bases: object

Manages simulation time and event scheduling.

Maintains ordered list of events and handles event scheduling with proper temporal ordering and priority handling.

Parameters:

start_date (datetime) – Start date and time for the simulation

Variables:

  • current_time (datetime) – Current simulation time

  • end_date (Optional[datetime]) – End date for the simulation

  • event_list (List) – Ordered list of pending events

  • _counter (int) – Counter for tie-breaking event ordering

__init__(start_date)[source]
Parameters:

start_date (datetime)

schedule_event(event)[source]

Schedule an event with proper ordering by time and priority.

Parameters:

event (Event) – Event to schedule

Notes

Events are ordered by: 1. Time (earlier events first) 2. Priority (lower numbers = higher priority) 3. Insertion order (tie-breaker)

get_next_event()[source]

Get next event without updating clock time.

Returns:

Next event in chronological order, or None if no events remain

Return type:

Optional[Event]

class simulation.base.BaseSimulation(start_date, environment=None)[source]

Bases: ABC

Abstract base class for simulations.

Provides core simulation functionality including event processing, protocol management, and simulation execution control.

Parameters:

  • start_date (datetime) – Start date and time for the simulation

  • environment (Optional[SimulationEnvironment], optional) – Simulation environment, creates new one if not provided

Variables:

  • clock (SimulationClock) – Manages simulation time and events

  • metrics (Dict[str, List[Any]]) – Stores simulation metrics

  • environment (SimulationEnvironment) – Global simulation environment

  • protocols (Dict[str, TreatmentProtocol]) – Registered treatment protocols

__init__(start_date, environment=None)[source]
Parameters:
abstract process_event(event)[source]

Process a simulation event.

Parameters:

event (Event) – Event to process

Notes

This method must be implemented by concrete simulation classes.

run(until)[source]

Run simulation until specified datetime.

Parameters:

until (datetime) – End date and time for the simulation

Notes

  • Processes events chronologically until end time is reached

  • Prints progress indicators (dots) for each simulated week

  • Includes safety limit to prevent infinite loops

register_protocol(protocol_type, protocol)[source]

Register a protocol for use in the simulation.

Parameters:

  • protocol_type (str) – Identifier for the protocol type

  • protocol (TreatmentProtocol) – Protocol instance to register

get_protocol(protocol_type)[source]

Get a registered protocol by type.

Parameters:

protocol_type (str) – Identifier for the protocol type

Returns:

Registered protocol or None if not found

Return type:

Optional[TreatmentProtocol]

schedule_protocol_event(time, patient_id, phase_type, action, data=None, priority=1)[source]

Schedule a protocol-specific event.

Parameters:

  • time (datetime) – When the event should occur

  • patient_id (str) – ID of the patient

  • phase_type (PhaseType) – Type of protocol phase

  • action (str) – Action to be performed

  • data (Dict[str, Any], optional) – Additional event data

  • priority (int, optional) – Event priority