`mosaik` — The Scenario API

This module provides convenient access to all classes and functions required to create scenarios and run simulations.

class World(sim_config: TypeAliasForwardRef('mosaik.async_scenario.SimConfig') | None = None, mosaik_config: MosaikConfig | None = None, time_resolution: float = 1.0, debug: bool = False, cache: bool = True, max_loop_iterations: int = 100, skip_greetings: bool = False, configure_logging: bool = True, asyncio_loop: AbstractEventLoop | None = None)[source]

The world holds all data required to specify and run the scenario.

We recommend that you use the world in a with block like so:

with mosaik.World(SIM_CONFIG) as world:
    # Scenario setup ...

    world.run(until=UNTIL)

This way, mosaik will keep the connection to the simulators alive until the end of the with block and you can still call extra methods on the to retrieve final simulation data, if needed.

However, you can also use a World outside of a with block.

The World provides a method to start a simulator process (start) and manages the simulator instances.

You have to provide a sim_config which tells the world which simulators are available and how to start them. See mosaik.simmanager.start for more details.

mosaik_config can be a dict or list of key-value pairs to set addional parameters overriding the defaults:

{
    'addr': ('127.0.0.1', 5555),
    'start_timeout': 2,  # seconds
    'stop_timeout': 2,   # seconds
}

Here, addr is the network address that mosaik will bind its socket to. start_timeout and stop_timeout specifiy a timeout (in seconds) for starting/stopping external simulator processes.

If execution_graph is set to True, an execution graph will be created during the simulation. This may be useful for debugging and testing. Note, that this increases the memory consumption and simulation time.

Using the skip_greetings and configure_logging parameters, you can configure how “wordy” mosaik will be. If you set skip_greetings to True, the big mosaik logo will no longer be shown when you create the world. If you set configure_logging to False, mosaik’s logging messages will not be enabled in loguru. You can still do this yourself by calling logger.enable("mosaik").

Parameters:

sim_config (Optional[SimConfig])
mosaik_config (Optional[MosaikConfig])
time_resolution (float)
debug (bool)
cache (bool)
max_loop_iterations (int)
skip_greetings (bool)
configure_logging (bool)
asyncio_loop (Optional[asyncio.AbstractEventLoop])

compile(*, use_cache: bool = False) → Dict[str, SimRunner][source]

Return the mapping of simulator IDs to their SimRunner instances.

This mirrors mosaik.async_scenario.AsyncWorld.compile and allows synchronous scenarios to inspect the simulators without relying on the deprecated sims accessor.

Parameters:: use_cache (bool)
Return type:: Dict[str, SimRunner]

connect(src: ~mosaik.async_scenario.Entity, dest: ~mosaik.async_scenario.Entity, *attr_pairs: str | ~typing.Tuple[str, str], async_requests: bool = False, time_shifted: bool | int = False, initial_data: ~typing.Dict[str, ~typing.Any] = {}, weak: bool = False, transform: ~typing.Callable[[~typing.Any], ~typing.Any] = <function World.<lambda>>)[source]

Warning

The keyword async_requests is deprecated and will be removed in a future release. Implement cyclic data flow using time-shifted and weak connections instead.

Connect the src entity to dest entity.

Establish a data-flow for each (src_attr, dest_attr) tuple in attr_pairs. If src_attr and dest_attr have the same name, you can optionally only pass one of them as a single string.

Raise a ScenarioError if both entities share the same simulator instance, if at least one (src. or dest.) attribute in attr_pairs does not exist, or if the connection would introduce a cycle in the data-flow (e.g., A → B → C → A).

If the dest simulator may make asynchronous requests to mosaik to query data from src (or set data to it), async_requests should be set to True so that the src simulator stays in sync with dest.

An alternative to asynchronous requests are time-shifted connections. Their data flow is always resolved after normal connections so that cycles in the data-flow can be realized without introducing deadlocks. For such a connection time_shifted should be set to True and initial_data should contain a dict with input data for the first simulation step of the receiving simulator.

An alternative to using async_requests to realize cyclic data-flow is given by the time_shifted kwarg. If set to True it marks the connection as cycle-closing (e.g. C → A). It must always be used with initial_data specifying a dict with the data sent to the destination simulator at the first step (e.g. {‘src_attr’: value}).

Parameters:

src (Entity)
dest (Entity)
attr_pairs (str | Tuple[str, str])
async_requests (bool)
time_shifted (bool | int)
initial_data (Dict[str, Any])
weak (bool)
transform (Callable[[Any], Any])

property entity_graph[source]: The graph of all entities. See :attr:`AsyncWorld.entity_graph <mosaik.async_scenario.AsyncWorld.entity_graph>.

get_data(entity_set: Iterable[Entity], *attributes: str) → Dict[Entity, Dict[str, Any]][source]

Get and return the values of all attributes for each entity of an entity_set.

The return value is a dict mapping the entities of entity_set to dicts containing the values of each attribute in attributes:

{
    Entity(...): {
        'attr_1': 'val_1',
        'attr_2': 'val_2',
        ...
    },
    ...
}

Parameters:

entity_set (Iterable[Entity])
attributes (str)

Return type:

Dict[Entity, Dict[str, Any]]

run(until: int, rt_factor: float | None = None, rt_strict: bool = False, print_progress: bool | Literal['individual'] = True, lazy_stepping: bool = True, *, shutdown: bool = True)[source]

Start the simulation until the simulation time until is reached. As mosaik has no way of resetting the simulators to their starting state, this method can only be called once.

Parameters:

until (int) – The end time for the simulation, exclusive (i.e. the step at time until will not be performed.)
rt_factor (float | None) – In order to perform real-time simulations, you can set rt_factor to a number > 0. A real-time factor of 1. means that 1 second in simulated time takes 1 second in real-time. An real-time factor of 0.5 will let the simulation run twice as fast as real-time. For correct behavior of the rt_factor, the time resolution of the scenario has to be set adequately, which is 1. [second] by default.
rt_strict (bool) – If the simulators are too slow for the real-time factor you chose, mosaik will emit a warning. If you want it to raise a RuntimeError, instead, you can set rt_strict to True.
print_progress (bool | Literal['individual']) – This controls whether progress bars are printed while the simulation is running. The default is to print one bar representing the global progress of the simulation. You can also set print_progress='individual' to get one bar per simulator in your simulation (in addition to the global one). print_progress=False turns off the progress bars completely. The progress bars use tqdm; see their documentation on how to write to the console without interfering with the bars.
lazy_stepping (bool) – If True a simulator can only run ahead one step of it’s successors. If False a simulator always steps as soon as all input is provided. This might decrease the simulation time but increase the memory consumption.
shutdown (bool) – If True and this World is not being used in a with block, mosaik will stop all simulators and close the connections to them at the end of the simulation run. You can set this to False if you want to keep the connections open and call shutdown yourself, later. (This is useful if you want to call extra methods on your simulator after the simulation is over; however, we recommend that you use the World in a with block.)

Raises:

RuntimeError – if this world has already been run

set_initial_event(sid: str, time: int = 0)[source]

Set an initial step for simulator sid at time time (default=0).

Parameters:

sid (str)
time (int)

shutdown()[source]: Shut-down all simulators and close the server socket.

property sims[source]: Returns the runtime simulators as a Dict with SimId as keys to SimRunner as values.

Deprecated since version 3.6: Use compile to retrieve the simulators without triggering this accessor.

start(starter: Starter | str, /, sim_id: str | None = None, **sim_params: Any) → ModelFactory[source]

Start the simulator named sim_name and return a ModelFactory for it.

Parameters:

starter (Starter | str)
sim_id (str | None)
sim_params (Any)

Return type:

ModelFactory

class AsyncWorld(sim_config: TypeAliasForwardRef('mosaik.async_scenario.SimConfig') | None = None, mosaik_config: MosaikConfig | None = None, time_resolution: float = 1.0, debug: bool = False, cache: bool = True, max_loop_iterations: int = 100, configure_logging: bool = False, skip_greetings: bool = True)[source]

The world holds all data required to specify and run the scenario.

It provides a method to start a simulator process (start) and manages the simulator instances.

You have to provide a sim_config which tells the world which simulators are available and how to start them. This will be stored as sim_config; see there for details.

mosaik_config can be a dict or list of key-value pairs to set additional parameters overriding the defaults:

{
    'addr': ('127.0.0.1', 5555),
    'start_timeout': 2,  # seconds
    'stop_timeout': 2,   # seconds
}

Here, addr is the network address that mosaik will bind its socket to. start_timeout and stop_timeout specifiy a timeout (in seconds) for starting/stopping external simulator processes.

If execution_graph is set to True, an execution graph will be created during the simulation. This may be useful for debugging and testing. Note that this increases the memory consumption and simulation time.

We recommend that you use AsyncWorld in an async with block like so:

async with AsyncWorld(SIM_CONFIG) as world:
    # call setup methods on world
    ...
    await world.run(UNTIL)

This will ensure that the connections to all remote simulators are properly closed at the end of the simulation. Alternatively, you can use the shutdown method manually.

Parameters:

sim_config (Optional[SimConfig])
mosaik_config (Optional[MosaikConfig])
time_resolution (float)
debug (bool)
cache (bool)
max_loop_iterations (int)
configure_logging (bool)
skip_greetings (bool)

cache_triggering_ancestors()[source]: Collects the ancestors of each simulator and stores them in the respective simulator object.

connect(src: ~mosaik.async_scenario.Entity, dest: ~mosaik.async_scenario.Entity, *attr_pairs: str | ~typing.Tuple[str, str], async_requests: bool = False, time_shifted: bool | int = False, initial_data: ~typing.Dict[str, ~typing.Any] = {}, weak: bool = False, transform: ~typing.Callable[[~typing.Any], ~typing.Any] = <function AsyncWorld.<lambda>>)[source]

Connect the src entity to dest entity.

Establish a data-flow for each (src_attr, dest_attr) tuple in attr_pairs. If src_attr and dest_attr have the same name, you you can optionally only pass one of them as a single string.

Raise a ScenarioError if both entities share the same simulator instance, if at least one (src. or dest.) attribute in attr_pairs does not exist, or if the connection would introduce a cycle in the data-flow (e.g., A → B → C → A).

If the dest simulator may make asynchronous requests to mosaik to query data from src (or set data to it), async_requests should be set to True so that the src simulator stays in sync with dest.

An alternative to asynchronous requests are time-shifted connections. Their data flow is always resolved after normal connections so that cycles in the data-flow can be realized without introducing deadlocks. For such a connection time_shifted should be set to True and initial_data should contain a dict with input data for the first simulation step of the receiving simulator.

An alternative to using async_requests to realize cyclic data-flow is given by the time_shifted kwarg. If set to True it marks the connection as cycle-closing (e.g. C → A). It must always be used with initial_data specifying a dict with the data sent to the destination simulator at the first step (e.g. {‘src_attr’: value}).

Parameters:

src (Entity)
dest (Entity)
attr_pairs (str | Tuple[str, str])
async_requests (bool)
time_shifted (bool | int)
initial_data (Dict[str, Any])
weak (bool)
transform (Callable[[Any], Any])

ensure_no_dataflow_cycles()[source]: Make sure that there is no cyclic dataflow with 0 total delay. Raise an exception with one such cycle otherwise.

async get_data(entity_set: Iterable[Entity], *attributes: str) → Dict[Entity, Dict[str, Any]][source]

Get and return the values of all attributes for each entity of an entity_set.

The return value is a dict mapping the entities of entity_set to dicts containing the values of each attribute in attributes:

{
    Entity(...): {
        'attr_1': 'val_1',
        'attr_2': 'val_2',
        ...
    },
    ...
}

Parameters:

entity_set (Iterable[Entity])
attributes (str)

Return type:

Dict[Entity, Dict[str, Any]]

async run(until: int, rt_factor: float | None = None, rt_strict: bool = False, print_progress: bool | Literal['individual'] = False, lazy_stepping: bool = True)[source]

Start the simulation and run it until the simulation time until is reached.

As mosaik has no way of resetting simulators to their initial state, this method can only be called once.

Unlike its counterpart run in the synchronous World, this method will not automatically close the connection to connected simulators, even outside of a with block. You need to call shutdown manually afterwards (or use a with block).

Parameters:

until (int) – The end of the simulation in mosaik time steps (exclusive).
rt_factor (float | None) – The real-time factor. If set to a number > 0, the simulation will run in real-time mode. A real-time factor of 1. means that 1 second in simulated time takes 1 second in real time. An real-time factor of 0.5 will let the simulation run twice as fast as real time. For correct behavior of the real-time factor, the time resolution of the scenario has to be set adequately (the default is 1 second).
rt_strict (bool) – If the simulators are too slow for the real-time factor you chose, mosaik will only print a warning by default. In order to raise a RuntimeError instead, you can set rt_strict to True.
print_progress (bool | Literal['individual']) –
Whether progress bars are printed while the simulation is running. The default is to print one bar representing the global progress of the simulation. You can also set the value to 'individual' to get one bar per simulator in your simulation (in addition to the global one). A value of False turns off the progress bars completely.

The progress bars use tqdm; see their documentation on how to write to the console without interfering with the bars.
lazy_stepping (bool) – Whether to prevent simulators from running ahead of their successors by more than one step. If False a simulator always steps as long all input is provided. This might decrease the simulation time but increase the memory consumption.

set_initial_event(sid: str, time: int = 0)[source]

Set an initial step for simulator sid at time time (default=0).

Parameters:

sid (str)
time (int)

async shutdown()[source]: Shut-down all simulators and close the server socket.

property sims: Dict[str, SimRunner][source]: Returns the runtime simulators as a Dict with SimId as keys to SimRunner as values.

Deprecated since version 3.6: Use compile to retrieve the simulators without triggering this accessor.

sim_config: SimConfig | None[source]

The config dictionary that tells mosaik how to start a simulator.

The sim config is a dictionary with one entry for every simulator. The entry itself tells mosaik how to start the simulator:

{
    'ExampleSimA': {
        'python': 'example_sim.mosaik:ExampleSim',
    },
    'ExampleSimB': {
        'cmd': 'example_sim %(addr)s',
        'cwd': '.',
    },
    'ExampleSimC': {
        'connect': 'host:port',
    },
}

ExampleSimA is a pure Python simulator. Mosaik will import the module example_sim.mosaik and instantiate the class ExampleSim to start the simulator.

ExampleSimB would be started by executing the command example_sim and passing the network address of mosaik das command line argument. You can optionally specify a current working directory. It defaults to ..

ExampleSimC can not be started by mosaik, so mosaik tries to connect to it.

config: MosaikConfigTotal[source]: The config dictionary for general mosaik settings.

until: int[source]: The time until which this simulation will run.

rt_factor: float | None[source]: The number of real-time seconds corresponding to one mosaik step.

time_resolution: float[source]: The number of seconds that correspond to one mosaik time step in this situation. The default value is 1.0, meaning that one integer step corresponds to one second simulated time.

max_loop_iterations: int[source]: The number of iterations allowed for same-time loops within one time step. This is checked to prevent accidental infinite loops. Increase this value if your same-time loops require many iterations to converge.

sim_progress: float[source]: The progress of the entire simulation (in percent).

entity_graph: networkx.Graph[FullId][source]: The graph of related entities. Nodes are (sid, eid) tuples. Each note has an attribute entity with an Entity.

tqdm: tqdm[NoReturn][source]: The tqdm progress bar for the total progress.

mosaik — The Scenario API

`mosaik` — The Scenario API