mosaik.scenario — Classes related to the scenario creation

This module provides the interface for users to create simulation scenarios for mosaik.

The World holds all necessary data for the simulation and allows the user to start simulators. It provides a ModelFactory (and a ModelMock) via which the user can instantiate model instances (entities). The method World.run finally starts the simulation.

mosaik.scenario.SimConfig

alias of Dict[str, PythonModel | ConnectModel | CmdModel]

class mosaik.scenario.World(sim_config, mosaik_config=None, time_resolution=1.0, debug=False, cache=True, max_loop_iterations=100, skip_greetings=False, configure_logging=True, asyncio_loop=None)

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 (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])

start(sim_name, sim_id=None, **sim_params)

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

Parameters:

• sim_name (str)

• sim_id (str | None)

• sim_params (Any)

Return type:

ModelFactory

connect(src, dest, *attr_pairs, async_requests=False, time_shifted=False, initial_data={}, weak=False)

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 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)

set_initial_event(sid, time=0)

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

Parameters:

• sid (str)

• time (int)

get_data(entity_set, *attributes)

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, rt_factor=None, rt_strict=False, print_progress=True, lazy_stepping=True, *, shutdown=True)

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.

In order to perform real-time simulations, you can set rt_factor to a number > 0. A rt-factor of 1. means that 1 second in simulated time takes 1 second in real-time. An rt-factor 0f 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.

If the simulators are too slow for the rt-factor you chose, mosaik prints by default only a warning. In order to raise a RuntimeError, you can set rt_strict to True.

print_progress 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.

You can also set the lazy_stepping flag (default: True). If True a simulator can only run ahead one step of it’s successors. If False a simulator always steps as long all input is provided. This might decrease the simulation time but increase the memory consumption.

At the end of the simulation, mosaik will stop all simulators and close the connections to them. There are two exceptions to this:

• If the flag shutdown is set to False, mosaik will not close the connection. In this case, you have to call shutdown yourself.

• If the :cls:`World` is used in a with block (recommended), the connection will be closed at the end of that block, instead.

(Keeping the connection open is useful to extract final data from simulators using extra methods.)

Parameters:

• until (int)

• rt_factor (float | None)

• rt_strict (bool)

• print_progress (bool | Literal['individual'])

• lazy_stepping (bool)

• shutdown (bool)

shutdown()

Shut-down all simulators and close the server socket.

class mosaik.scenario.ModelFactory(async_model_factory, loop)

This is a facade for a simulator sim that allows the user to create new model instances (entities) within that simulator.

For every model that a simulator publicly exposes, the ModelFactory provides a ModelMock attribute that actually creates the entities.

If you access an attribute that is not a model or if the model is not marked as public, an ScenarioError is raised.

Parameters:

• async_model_factory (AsyncModelFactory)

• loop (asyncio.AbstractEventLoop)

class mosaik.scenario.ModelMock(async_model_mock, loop)

Instances of this class are exposed as attributes of ModelFactory and allow the instantiation of simulator models.

You can call an instance of this class to create exactly one entity: sim.ModelName(x=23). Alternatively, you can use the create method to create multiple entities with the same set of parameters at once: sim.ModelName.create(3, x=23).

Parameters:

• async_model_mock (AsyncModelMock)

• loop (asyncio.AbstractEventLoop)

create(num, **model_params)

Create num entities with the specified model_params and return a list with the entity dicts.

The returned list of entities is the same as returned by mosaik_api_v3.Simulator.create, but the simulator is prepended to every entity ID to make them globally unique.

Parameters:

• num (int)

• model_params (Any)

class mosaik.scenario.Entity(sid, eid, sim_name, model_mock, children, extra_info=None)

An entity represents an instance of a simulation model within mosaik.

Parameters:

• sid (str)

• eid (str)

• sim_name (str)

• model_mock (AsyncModelMock)

• children (List[Entity])

• extra_info (Any)

sid: str

The ID of the simulator this entity belongs to.

eid: str

The entity’s ID.

sim_name: str

The entity’s simulator name.

model_mock: AsyncModelMock

The entity’s type (or class).

children: List[Entity]

An entity set containing subordinate entities.

property full_id: str

Full, globally unique entity id sid.eid.