orchestrator

context

class BaseModelState(context: T)

Bases: State[ConnectedModel], ABC

valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = ()
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = ()
class Busy(context: T)

Bases: WaitingForMessage

Base class for if a model is doing something and a response is required. While busy, one or more Command-s may come in, which need to be processed later. These are stored until the model returns and they can be processed

on_enter()
valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = (<class 'movici_simulation_core.messages.UpdateMessage'>, <class 'movici_simulation_core.services.orchestrator.context.NoUpdateMessage'>, <class 'movici_simulation_core.messages.QuitMessage'>)
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = (<class 'movici_simulation_core.messages.ErrorMessage'>,)
class ConnectedModel(name: str, timeline: ~movici_simulation_core.services.orchestrator.context.TimelineController, send: ~typing.Callable[[~movici_simulation_core.messages.Message], None], logger: ~logging.Logger = <factory>, publishes_to: list[~movici_simulation_core.services.orchestrator.context.ConnectedModel] = <factory>, subscribed_to: list[~movici_simulation_core.services.orchestrator.context.ConnectedModel] = <factory>, pub: dict | None = <factory>, sub: dict | None = <factory>, fsm_config: ~movici_simulation_core.services.orchestrator.fsm.FSMConfig[~movici_simulation_core.services.orchestrator.context.ConnectedModel] | None = None)

Bases: object

Holds connection state and other data concerning a (connected) model

busy: bool = False
failed: bool = False
fsm: FSM[ConnectedModel, Message]
fsm_config: FSMConfig[ConnectedModel] | None = None
log_invalid(message, valid_messages: Iterable[Type[Message]])
logger: Logger
name: str
next_time: int | None = None
pending_new_time: NewTimeMessage | None = None
pending_quit: QuitMessage | None = None
pending_updates: list[UpdateMessage]
pub: dict | None
publishes_to: list[ConnectedModel]
recv_event(event: Message)
send: Callable[[Message], None]
send_command(message: NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage) → None

Send a command message to the model

start()
sub: dict | None
subscribed_to: list[ConnectedModel]
timeline: TimelineController
timer: Stopwatch
class Context(models: 'ModelCollection', timeline: 'TimelineController', global_timer: 'Stopwatch' = None, phase_timer: 'Stopwatch' = None, logger: 'logging.Logger' = <factory>, orchestrator_failed: 'bool' = False)

Bases: object

property failed
finalize()
global_timer: Stopwatch = None
log_elapsed_global_time(seconds: float)
log_elapsed_phase_time(seconds: float)
log_finalize_message()
log_interconnectivity_matrix()
log_new_phase(phase: str)
log_new_time()
logger: Logger
models: ModelCollection
orchestrator_failed: bool = False
phase_timer: Stopwatch = None
queue_models_for_next_time()
recv_for_all(message: Message)

Receive a Message for all models

recv_message(model: ConnectedModel, message: Message)
timeline: TimelineController
class Done(context: T)

Bases: BaseModelState

The model is either done or failed, ignore any further messages

handle_event(event)
requires_event = True
class Finalizing(context: T)

Bases: Busy

A QuitMessage has been sent to the model, which needs to acknowledge it and shut down, ignore all commands

process_command(msg: NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage)
valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = (<class 'movici_simulation_core.messages.NewTimeMessage'>, <class 'movici_simulation_core.services.orchestrator.context.NoUpdateMessage'>, <class 'movici_simulation_core.messages.UpdateMessage'>, <class 'movici_simulation_core.messages.QuitMessage'>)
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = (<class 'movici_simulation_core.messages.AcknowledgeMessage'>, <class 'movici_simulation_core.messages.ErrorMessage'>)
class HasPendingNewTimeAndIdle(context: T)

Bases: Condition[ConnectedModel]

met() → bool
class HasPendingQuitAndIdle(context: T)

Bases: Condition[ConnectedModel]

met() → bool
class HasPendingUpdatesAndIdle(context: T)

Bases: Condition[ConnectedModel]

met() → bool
class Idle(context: T)

Bases: WaitingForMessage

Base class for if a model is awaiting further instructions

valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = (<class 'movici_simulation_core.messages.NewTimeMessage'>, <class 'movici_simulation_core.services.orchestrator.context.NoUpdateMessage'>, <class 'movici_simulation_core.messages.UpdateMessage'>, <class 'movici_simulation_core.messages.QuitMessage'>)
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = ()
class IsIdle(context: T)

Bases: Condition[ConnectedModel]

met() → bool
class ModelCollection

Bases: dict[bytes, ConnectedModel]

property busy
determine_interdependency()

calculate the subscribers for every model based on the pub/sub mask.

property failed
property next_time
reset_model_timers()
class ModelHasFailed(context: T)

Bases: Condition[ConnectedModel]

met() → bool
class ModelWaitingForDependencies(context: T)

Bases: Condition[ConnectedModel]

met() → bool
class NewTime(context: T)

Bases: Busy

A NewTime message will be sent to the model, and it needs to Acknowledge that

on_enter()
valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = (<class 'movici_simulation_core.messages.UpdateMessage'>, <class 'movici_simulation_core.services.orchestrator.context.NoUpdateMessage'>, <class 'movici_simulation_core.messages.QuitMessage'>)
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = (<class 'movici_simulation_core.messages.AcknowledgeMessage'>, <class 'movici_simulation_core.messages.ErrorMessage'>)
class NoUpdateMessage

Bases: Message

A NoUpdateMessage is sent to a subscribed (dependent) ConnectedModel to indicate that it’s dependency has finished calculating (but not produced any data) so that the subscribed ConnectedModel can determine whether to send out any pending updates

class PendingMoreUpdates(context: T)

Bases: Idle

The model has one or more dependencies that are still calculating. We wait until a dependency returns before re-evaluating whether we can send the update to the model

valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = (<class 'movici_simulation_core.services.orchestrator.context.NoUpdateMessage'>, <class 'movici_simulation_core.messages.UpdateMessage'>, <class 'movici_simulation_core.messages.QuitMessage'>)
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = ()
class ProcessPendingQuit(context: T)

Bases: BaseModelState

While the model was Busy, a QuitMessage came in which needs to be processed, this state doesn’t wait for messages

run()
class ProcessPendingUpdates(context: T)

Bases: BaseModelState

While the model was Busy, one or more updates came in which needs to be processed, this state doesn’t wait for messages. If the model has one or more dependencies that are still calculating, we can’t send the update yet but have to wait until all dependencies are finished

process_pending_updates()
run()
class Registration(context: T)

Bases: Busy

A RegistrationMessage is expected from the model

valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = (<class 'movici_simulation_core.messages.QuitMessage'>,)
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = (<class 'movici_simulation_core.messages.RegistrationMessage'>, <class 'movici_simulation_core.messages.ErrorMessage'>)
class TimelineController(start: 'int', end: 'int', current_time: 'int | None' = None)

Bases: object

current_time: int | None = None
end: int
set_model_to_start(model: ConnectedModel)
set_next_time(model: ConnectedModel, next_time: int | None = None)
start: int
class Updating(context: T)

Bases: Busy

The model is processing an UpdateMessage and calculating a ResultMessage is expected

valid_commands: tuple[Type[NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage], ...] = (<class 'movici_simulation_core.messages.UpdateMessage'>, <class 'movici_simulation_core.services.orchestrator.context.NoUpdateMessage'>, <class 'movici_simulation_core.messages.QuitMessage'>)
valid_responses: tuple[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage], ...] = (<class 'movici_simulation_core.messages.ResultMessage'>, <class 'movici_simulation_core.messages.ErrorMessage'>)
class WaitingForMessage(context: T)

Bases: BaseModelState

handle_event(event: Message)
handle_invalid_response(msg: RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage, valid_messages: Iterable[Type[RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage]])
handle_response(msg: RegistrationMessage | AcknowledgeMessage | ResultMessage | ErrorMessage)
notify_subscribers(command: UpdateMessage | NoUpdateMessage)

Notify subscribers that this ConnectedModel’s model has returned.

process_command(msg: NewTimeMessage | UpdateMessage | NoUpdateMessage | UpdateSeriesMessage | QuitMessage)
process_new_time(msg: NewTimeMessage)
process_no_update(msg: NoUpdateMessage)
process_quit(msg: QuitMessage)
process_update(msg: UpdateMessage)
recv_message(msg: Message)
requires_event = True

fsm

class Always(context: T)

Bases: Condition

met()
class Condition(context: T)

Bases: ABC, Generic[T]

abstractmethod met() → bool
class FSM(config: FSMConfig[T], context: T = None, raise_on_done=True)

Bases: Generic[T, E]

Implementation of a Finite State Machine runner. Requires a FSMConfig as a definition of the available states and the possible transitions. The config contains an initial_state that is entered when the FSM is started by invoking the FSM.start() method. If the initial state is configured with requires_event=True, then the FSM waits until the FSM.handle_event() method is called upon which it will pass the event through to the active state by calling the state’s handle_event method, which the State must have implemented if it has requires_event set to True. If the active state does not have requires_event=True it will call the run method instead. After every state is invoked (by calling either the handle_event or run method) the configured transitions for the state (as given by the FSMConfig) are evaluated against the context. The FSM transitions to State belonging to the first matched transition, which will become the active state. If no condition is matched, then the current state remains the active state for the next iteration. The FSM continues to call the active state’s run method and transition until it encounters a state that has requires_event=True upon which it will wait for FSM.handle_event() to be called.

The FSM may terminate iff a state raises an FSMException, usually FSMDone to indicate a normal end to the FSM, or FSMError in case of an abnormal end to the FSM. If an FSMError was raised then the FSM.failure flag will be set. If no valid state ever raises an FSMException then the FSM may run indefinitely

Parameters:

  • config – An FSMConfig for the finite state machine.

  • context – An arbitrary, mutalbe object that is kept for the lifetime of the FSM. It can be used to store state that must be shared between States and will be used to evaluate any transition conditions

  • raise_on_done – A boolean to indicate whether to raise an FSMException in case the FSM terminates. The exception will be the same as the exception that terminated the FSM

catch_fsmexceptions()
ensure_not_done()
ensure_not_started()
handle_event(event: E)
run_until_event_required()
start()

Start the FSM

state: State[T]
transition()
class FSMConfig(initial_state: ~typing.Type[~movici_simulation_core.services.orchestrator.fsm.State[~movici_simulation_core.services.orchestrator.fsm.T]], states: dict[~typing.Type[~movici_simulation_core.services.orchestrator.fsm.State[~movici_simulation_core.services.orchestrator.fsm.T]], ~typing.Sequence[tuple[~typing.Type[~movici_simulation_core.services.orchestrator.fsm.Condition], ~typing.Type[~movici_simulation_core.services.orchestrator.fsm.State]]]] = <factory>, strict: bool = True)

Bases: Generic[T]

A config for setting up a FSM final state machine

Parameters:

  • initial_state – the initial state

  • states – a dictionary of all possible states as type``s and their transitions. Transitions are a sequence of ``(type[Condition], type[State]) tuples. When determining which State to transition to, the transitions are checked starting from the top entry in the transitions sequence. The FSM transitions to the state belonging to the first transition that is matched.

  • strict – a boolean whether to validate that all states mentioned in the transitions and initial state must have an entry in the states dictionary

initial_state: Type[State[T]]
states: dict[Type[State[T]], Sequence[tuple[Type[Condition], Type[State]]]]
strict: bool = True
class State(context: T)

Bases: ABC, Generic[T]

Base class for FSM States. When subclassing a State, implement one of the run or handle_event methods. If a state is a transient state and needs to run its logic and transition to a new state immediately, implement the run method. If a state needs to wait for an event to be sent, set the State.requires_event class variable to True and implement the handle_event method

handle_event(event: Any)

Entry point for a state that requires an event. This method will be called by the FSM when the class variable requires_event is set to True and it receives an event from upstream

on_enter()

Called once by the FSM when entering a state

requires_event = False
run()

Entry point for a state that does not require an event. This method will be called unconditionally by the FSM when requires_event is set to False (the default). States that do no have requires_event set are generally transient states that need to run some logic before immediately transitioning to another state

next_state(context, transitions: Sequence[tuple[Type[Condition], Type[State]]])

interconnectivity

class Publisher(*args, **kwargs)

Bases: Protocol

name: str
publishes_to: List[Publisher]
format_matrix(models: Sequence[Publisher], title='', match='X')
title     |0|1|2|3|
0|model_0 | |X|X|X|
1|model_1 | | | |X|
2|model_2 | | | |X|
3|model_10| | | | |

service

class Orchestrator

Bases: Service

The class that manages the timeline and acts as a broker between models

context: Context
fsm: FSM[Context, Tuple[str, Message]]
classmethod install(sim: Simulation)
logger: Logger
make_send(identifier: str)

create a send function that a can be used to send a message to a specific client connected to the ZMQ Router

restart_model_timer(model: str)

Allow resetting a specific model timer. This is used when running using the InProcessSimulationRunner. Orchestrator starts the timer by default when it sends out the message, but we need to actually start the timer when we start processing the message

run()
settings: Settings
setup(*, settings: Settings, stream: BaseStream, logger: Logger, **_)
start()

Start the orchestrator FSM but do not run the stream (yet)

stream: BaseStream
timeline: TimelineController

states

class AllModelsDone(context: T)

Bases: OrchestratorCondition

met() → bool
class AllModelsReady(context: T)

Bases: OrchestratorCondition

met() → bool
class EndFinalizingPhase(context: T)

Bases: OrchestratorState

run()
class Failed(context: T)

Bases: OrchestratorCondition

met() → bool
class FinalizingWaitForModels(context: T)

Bases: WaitForModels

class ModelsRegistration(context: T)

Bases: WaitForModels

class NewTime(context: T)

Bases: OrchestratorState

run()
class OrchestratorCondition(context: T)

Bases: Condition[Context], ABC

class OrchestratorState(context: T)

Bases: State[Context]

class StartFinalizingPhase(context: T)

Bases: OrchestratorState

run()
class StartInitializingPhase(context: T)

Bases: OrchestratorState

run()
class StartRunningPhase(context: T)

Bases: OrchestratorState

run()
class WaitForModels(context: T)

Bases: OrchestratorState, ABC

handle_event(event: Tuple[str, Message])
requires_event = True
class WaitForResults(context: T)

Bases: WaitForModels

stopwatch

class ReportingStopwatch(on_stop: ~typing.Callable[[float], None] | None = None, on_reset: ~typing.Callable[[float], None] | None = None, now_func: ~typing.Callable[[], float] = <built-in function monotonic>, ignore_errors=False)

Bases: Stopwatch

reset() → float
stop() → float
class Stopwatch(now_func: Callable[[], float] = None, ignore_errors=False)

Bases: object

property elapsed
reset() → float

returns Stopwatch.total_elapsed

restart() → float
restart_current()

Restart the current timer, without recording the current progress

property running
start()
stop() → float

returns Stopwatch.elapsed

property total_elapsed