Integrating a simulation model into the mosaik ecosystem

In this section we’ll first implement a simple example simulator. We’ll then implement mosaik’s Sim-API step-by-step.

The model

We want to implement a very simple model with the following behavior:

  • val0 = init_val

  • vali = vali − 1 + delta for iN, i > 0, deltaZ

That simply means our model has a value val to which we add some delta (which is a positive or negative integer) at every simulation step. Our model has the attribute delta (with value 1 by default) which can be changed by control mechanisms to alter the behavior of the model. And it has the (output) attribute val which is its current value.

Schematic diagram of our example model.

Schematic diagram of our example model. You can change the delta and collect the val as output.

Here is a possible implementation of that simulation model in Python:

# example_model.py
"""
This module contains a simple example model.

"""


class Model:
    """Simple model that increases its value *val* with some *delta* every
    step.

    You can optionally set the initial value *init_val*. It defaults to ``0``.

    """
    def __init__(self, init_val=0):
        self.val = init_val
        self.delta = 1

    def step(self):
        """Perform a simulation step by adding *delta* to *val*."""
        self.val += self.delta

Setup for the API implementation

So lets start implementing mosaik’s Sim-API for this model. We can use the Python high-level API for this. This package eases our workload, because it already implements everything necessary for communicating with mosaik. It provides an abstract base class which we can sub-class. So we only need to implement four methods and we are done.

If you already installed mosaik and the demo, you already have this package installed in your mosaik virtualenv.

We start by creating a new simulator_mosaik.py and import the module containing the mosaik API as well as our model:

# simulator_mosaik.py
"""
Mosaik interface for the example simulator.

"""
import mosaik_api

import example_model

Simulator meta data

Next, we prepare the meta data dictionary that tells mosaik which time paradigm it follows (time-based, event-based, or hybrid), which models our simulator implements and which parameters and attributes it has. Since this data is usually constant, we define this at module level (which improves readability):

META = {
    'type': 'time-based',
    'models': {
        'ExampleModel': {
            'public': True,
            'params': ['init_val'],
            'attrs': ['delta', 'val'],
        },
    },
}

In this case we create a time-based simulator. We added our “ExampleModel” model with the parameter init_val and the attributes delta and val. At this point we don’t care if they are inputs or outputs. We just list everything we can read or write. The public flag should usually be True. You can read more about it in the Sim API docs. From this information, mosaik deduces that our model could be used in the following way:

# Model name and "params" are used for constructing instances:
model = ExampleModel(init_val=42)
# "attrs" are normal attributes:
print(model.val)
print(model.delta)

The Simulator class

The package mosaik_api defines a base class Simulator for which we now need to write a sub-class:

class ExampleSim(mosaik_api.Simulator):
    def __init__(self):
        super().__init__(META)
        self.eid_prefix = 'Model_'
        self.entities = {}  # Maps EIDs to model instances/entities

In our simulator’s __init__() method (the constructor) we need to call Simulator.__init__() and pass the meta data dictionary to it. Simulator.__init__() will add some more information to the meta data and set it as self.meta to our instance.

We also set a prefix for our entity IDs and prepare a dictionary which will hold some information about the entities that we gonna create.

We can now start to implement the four API calls init, create, step and get_data:

init()

This method will be called exactly once while the simulator is being started via World.start(). It is used for additional initialization tasks (e.g., it can handle parameters that you pass to a simulator in your scenario definition). It must return the meta data dictionary self.meta:

    def init(self, sid, time_resolution, eid_prefix=None):
        if float(time_resolution) != 1.:
            raise ValueError('ExampleSim only supports time_resolution=1., but'
                             ' %s was set.' % time_resolution)
        if eid_prefix is not None:
            self.eid_prefix = eid_prefix
        return self.meta

The first argument is the ID that mosaik gave to that simulator instance. The second argument is the time resolution of the scenario. In this example only the default value of 1. (second per integer time step) is supported. If you set another value in the scenario, the simulator would throw an error and stop.

In addition to that, you can define further (optional) parameters which you can later set in your scenario. In this case, we can optionally overwrite the eid_prefix that we defined in __init__().

create()

create() is called in order to initialize a number of simulation model instances (entities) within that simulator. It must return a list with some information about each entity created:

    def create(self, num, model, init_val):
        next_eid = len(self.entities)
        entities = []

        for i in range(next_eid, next_eid + num):
            model_instance = example_model.Model(init_val)
            eid = '%s%d' % (self.eid_prefix, i)
            self.entities[eid] = model_instance
            entities.append({'eid': eid, 'type': model})

        return entities

The first two parameters tell mosaik how many instances of which model you want to create. As in init(), you can specify additional parameters for your model. They must also appear in the params list in the simulator meta data or mosaik will reject them. In this case, we allow setting the initial value init_val for the model instances.

For each entity, we create a new entity ID 1 and a model instance. We also create a mapping (self.entities) from the entity ID to our model. For each entity we create we also add a dictionary containing its ID and type to the entities list which is returned to mosaik. In this example, it has num entries for the model model, but it may get more complicated if you have, e.g., hierarchical models.

1

Although entity IDs can be plain integers, it is advisable to use something more meaningful to ease debugging and analysis.

step()

The step() method tells your simulator to perform a simulation step. It receives its current simulation time, a dictionary with input values from other simulators (if there are any), and the time until the simulator can safely advance its internal time without creating a causality error. For time-based simulators (as in our example) it can be safely ignored (it is equal to the end of the simulation then). The method returns to mosaik the time at which it wants to do its next step. For event-based and hybrid simulators a next (self-)step is optional. If there is no next self-step, the return value is None/null.

    def step(self, time, inputs, max_advance):
        # Check for new delta and do step for each model instance:
        for eid, model_instance in self.entities.items():
            if eid in inputs:
                attrs = inputs[eid]
                for attr, values in attrs.items():
                    new_delta = sum(values.values())
                model_instance.delta = new_delta

            model_instance.step()

        return time + 1  # Step size is 1 second

In this example, the inputs could be something like this:

{
    'Model_0': {
        'delta': {'src_id_0': 23},
    },
    'Model_1':
        'delta': {'src_id_1': 42},‘val’ :{ 'src_id_1': 20},
    },
}

The inner dictionaries containing the actual values may contain multiple entries if multiple source entities provide input for another entity. In the case above, we have two source entities, ‘Model_0’ providing the delta value to the destination entity (object of ExampleSim) and ‘Model_1’ providing the delta and val value to the destination entity (object of ExampleSim). The source entitiy, ‘Model_0’ has the attribute ‘delta’ as the key to another nested dictionary which contains the simulator id and its corresponding ‘delta’ value. Similarly the source entity ‘Model_1’ has the attributes ‘delta’ and ‘val’ as the keys to two other nested dictionaries which contain the simulator id and its corresponding ‘delta’ and ‘val’ values.

The structure of the inputs dictionary created by mosaik is always the same as depicted above, only the number of source entities (dependent on the connections in the scenario (‘Model_0’ and ‘Model_1’ in our case)) and the number of attributes passed by the source entity varies. The first key of the nested dictionary will be the source entity (‘Model_1’), the following keys will be the attributes passed by this source entity to the destination entity (‘delta’: {‘src_id_1’: 42}, ‘val’: {‘src_id_1’: 20}).

The simulator receiving these inputs is responsible for aggregating them (e.g., by taking their sum, minimum or maximum. Since we are not interested in the source’s IDs, we convert that dict to a list with values.values() before we calculate the sum of all input values.

After we converted the inputs to something that our simulator can work with, we let it finally perform its next simulation step.

The return value time + 1 tells mosaik that we wish to perform the next step in one second (in simulation time), as the time_resolution is 1. (second per integer step). Instead of using a fixed (hardcoded) step size you can easily implement any other stepping behavior.

get_data()

The get_data() call allows other simulators to get the values of the delta and val attributes of our models (the attributes we listed in the simulator meta data):

    def get_data(self, outputs):
        data = {}
        for eid, attrs in outputs.items():
            model = self.entities[eid]
            data[eid] = {}
            for attr in attrs:
                if attr not in self.meta['models']['ExampleModel']['attrs']:
                    raise ValueError('Unknown output attribute: %s' % attr)

                # Get model.val or model.delta:
                data[eid][attr] = getattr(model, attr)

The outputs parameter contains the query and may in our case look like this:

{
    'Model_0': ['delta', 'value'],
    'Model_1': ['value'],
}

The Outputs dictionary may contain multiple keys if multiple destination entities ask for the output from the source entity. In this case we have two destination entities ‘Model_0’ and ‘Model_1’ which are requesting for the source attributes. ‘Model_0’ is requesting for the two source attributes ‘delta’ and ‘value’, whereas ‘Model_1’ is requesting for 1 source attribute ‘value’. The structure of the Outputs dictionary created by mosaik is always the same as depicted above, only the number of destination entities (dependent on the connections in the scenario (‘Model_0’ and ‘Model_1’ in our case)) and the number of attributes requested by the destination entity varies.

In our implementation we loop over each entity ID for which data is requested. We then loop over all requested attributes and check if they are valid. If so, we dynamically get the requested value from our model instance via getattr(obj, 'attr'). We store all values in the data dictionary and return it when we are done.

The expected return value would then be:

{
    'Model_0': {'delta': 1, 'value': 24},
    'Model_1': {'value': 3},
}

Making it executable

The last step is adding a main() method to make our simulator executable (e.g., via python -m simulator_mosaik HOST:PORT). The package mosaik_api contains the method start_simulation() which creates a socket, connects to mosaik and listens for requests from it. You just call it in your main() and pass an instance of your simulator class to it:

def main():
    return mosaik_api.start_simulation(ExampleSim())


if __name__ == '__main__':
    main()

Simulators running on different nodes than the mosaik instance are supported explicitly with the mosaik Python-API v2.4 upward via the remote flag. A simulator with the start_simulation() method in its main() can then be called e.g. via

python simulator_mosaik –r HOST:PORT

in the command line. The mosaik scenario, started independently, can then connect to the simulator via the statement connect: HOST:PORT in its “sim_config” ( Configuration). Note that it may make sense to introduce a short waiting time into your scenario to give you enough time to start both processes. Alternatively, the remote connection of simulators supports also a timeout (via the timeout flag, e.g. –t 60 in the command line call will cause your simulator to wait for 60 seconds for an initial message from mosaik).

Summary

We have now implemented the mosaik Sim-API for our simulator. The following listing combines all the bits explained above:

# simulator_mosaik.py
"""
Mosaik interface for the example simulator.

"""
import mosaik_api

import example_model


META = {
    'type': 'time-based',
    'models': {
        'ExampleModel': {
            'public': True,
            'params': ['init_val'],
            'attrs': ['delta', 'val'],
        },
    },
}


class ExampleSim(mosaik_api.Simulator):
    def __init__(self):
        super().__init__(META)
        self.eid_prefix = 'Model_'
        self.entities = {}  # Maps EIDs to model instances/entities

    def init(self, sid, time_resolution, eid_prefix=None):
        if float(time_resolution) != 1.:
            raise ValueError('ExampleSim only supports time_resolution=1., but'
                             ' %s was set.' % time_resolution)
        if eid_prefix is not None:
            self.eid_prefix = eid_prefix
        return self.meta

    def create(self, num, model, init_val):
        next_eid = len(self.entities)
        entities = []

        for i in range(next_eid, next_eid + num):
            model_instance = example_model.Model(init_val)
            eid = '%s%d' % (self.eid_prefix, i)
            self.entities[eid] = model_instance
            entities.append({'eid': eid, 'type': model})

        return entities

    def step(self, time, inputs, max_advance):
        # Check for new delta and do step for each model instance:
        for eid, model_instance in self.entities.items():
            if eid in inputs:
                attrs = inputs[eid]
                for attr, values in attrs.items():
                    new_delta = sum(values.values())
                model_instance.delta = new_delta

            model_instance.step()

        return time + 1  # Step size is 1 second

    def get_data(self, outputs):
        data = {}
        for eid, attrs in outputs.items():
            model = self.entities[eid]
            data[eid] = {}
            for attr in attrs:
                if attr not in self.meta['models']['ExampleModel']['attrs']:
                    raise ValueError('Unknown output attribute: %s' % attr)

                # Get model.val or model.delta:
                data[eid][attr] = getattr(model, attr)

        return data


def main():
    return mosaik_api.start_simulation(ExampleSim())


if __name__ == '__main__':
    main()

We can now start to write our first scenario, which we will do in the next section.