simulation

common

class ActiveModuleInfo(name: str, daemon: bool)

Bases: object

daemon: bool

name: str

process: Process | None = None

class ModelFromInstanceInfo(name: str, daemon: bool, instance: movici_simulation_core.core.types.Model)

Bases: ActiveModuleInfo, ModelInfo

get_instance()

instance: Model

class ModelFromTypeInfo(name: str, daemon: bool, cls: Type[movici_simulation_core.core.types.Model], config: dict | None = None)

Bases: ActiveModuleInfo, ModelInfo

cls: Type[Model]

config: dict | None = None

get_instance()

class ModelInfo

Bases: object

get_instance() → Model

name: str

class ModelTypeInfo(identifier: str, cls: Type[movici_simulation_core.core.types.Model])

Bases: ModuleTypeInfo

cls: Type[Model]

class ModuleTypeInfo(identifier: str)

Bases: object

identifier: str

class ServiceInfo(name: str, daemon: bool, cls: Type[movici_simulation_core.core.types.Service], address: str | None = None)

Bases: ActiveModuleInfo

address: str | None = None

cls: Type[Service]

fill_service_discovery(svc_discovery: Dict[str, str])

set_port(port: int)

class ServiceTypeInfo(identifier: str, cls: Type[movici_simulation_core.core.types.Service], auto_use: bool, daemon: bool)

Bases: ModuleTypeInfo

auto_use: bool

cls: Type[Service]

daemon: bool

class SimulationRunner(modules: dict[str, ActiveModuleInfo], settings: Settings, schema: AttributeSchema, strategies: Sequence[Type])

Bases: object

distributed

class DistributedSimulationRunner(modules: dict[str, ActiveModuleInfo], settings: Settings, schema: AttributeSchema, strategies: Sequence[Type])

Bases: SimulationRunner

SimulationRunner that runs every model and service in its own separate process. The models and services then connect using TCP and zeroMQ

run() → int

starts up services from config and auto_use using ServiceRunner. Collects service addresses starts up models from config with service addresses for discovery using ModelRunner tracks models and services, terminates when necessary (question: when do we terminate everything and when does the orchestrator take over exception handling?)

class ModelRunner(model_info: ModelInfo, settings: Settings, strategies: list[Type] | None = None, schema: AttributeSchema | None = None)

Bases: ProcessRunner

Provides logic for:

• Creating a Process (daemon=False) that runs a Model. Using a wrapping function, this

  subprocess will:

  • create the model with its model adapter

  • create a (dealer) socket

  • run the model

  • catch exceptions from model, send ERROR message

  • raise exceptions when not directly from model

• Fills the ModelInfo object

By creating the process as daemon=False, models can spawn their own subprocesses

close()

entry_point()

init_data_handler: ServicedInitDataClient | None = None

socket: MessageDealerSocket | None = None

start()

update_handler: UpdateDataClient | None = None

class ProcessRunner(strategies: list[Type] | None, schema: AttributeSchema | None = None)

Bases: object

ctx = <multiprocessing.context.ForkContext object>

prepare_subprocess()

class ServiceRunner(service: ServiceInfo, settings: Settings, strategies: list[Type] | None = None, schema: AttributeSchema | None = None)

Bases: ProcessRunner

Provides logic for:

• Creating a Pipe that the Service can use to announce its port

• Creating a Process (daemon=True) that runs Service. Using a wrapping function this

  subprocess will

  • create the service

  • create a (router) socket

  • announce the port

  • run the Service

  • raise exception on failure

• Fills the ServiceInfo object

• Raising an exception if it fails to announce the port in time

By creating the process as daemon=True, services cannot spawn their own subprocesses but they can be easily terminated

TIMEOUT = 5

entry_point(conn: Connection)

start()

in_process

class InProcessOrchestratorStream

Bases: BaseStream[Tuple[str, Message]]

InProcessOrchestratorStream is a small implementation of BaseStream so that the orchestrator can run in in-process mode. Different from a regular Stream, it is not the means to a long running process, ie. its run() method returns quickly

handle_message(payload: Tuple[str, Message])

send(payload: Tuple[str, Message])

class InProcessSimulationRunner(modules: dict[str, ActiveModuleInfo], settings: Settings, schema: AttributeSchema, strategies: Sequence[Type])

Bases: SimulationRunner

A SimulationRunner that connects all models and services in a single process, as opposed to the DistributedSimulationRunner that sets up models and services that run in a separate process each and have them connect through TCP

handle_model_command(model: ModelConnector, command: Message)

run() → int

class InProcessUpdateDataClient

Bases: UpdateDataClientBase[dict]

clear()

close()

counter: Iterator[str]

get(address: str, key: str, mask: dict | None) → dict

put(data: dict) → Tuple[str, str]

reset_counter()

class NoopSerializer(*args, **kwargs)

Bases: InternalSerializationStrategy[dict]

dumps(data: dict) → dict

loads(raw_data: dict) → dict