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 SimAPI step-by-step

The simulator

We want to implement a simulator for a very simple model with a discrete step size of 1. Our model will have 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 (optional) input delta which can be used by control mechanisms to alter the behavior of the model. It has the output val which is its current value.

Schematic diagram of our example model.

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

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

# simulator.py
"""
This module contains a simple example simulator.

"""


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


class Simulator(object):
    """Simulates a number of ``Model`` models and collects some data."""
    def __init__(self):
        self.models = []
        self.data = []

    def add_model(self, init_val):
        """Create an instances of ``Model`` with *init_val*."""
        model = Model(init_val)
        self.models.append(model)
        self.data.append([])  # Add list for simulation data

    def step(self, deltas=None):
        """Set new model inputs from *deltas* to the models and perform a
        simulatino step.

        *deltas* is a dictionary that maps model indices to new delta values
        for the model.

        """
        if deltas:
            # Set new deltas to model instances
            for idx, delta in deltas.items():
                self.models[idx].delta = delta

        # Step models and collect data
        for i, model in enumerate(self.models):
            model.step()
            self.data[i].append(model.val)


if __name__ == '__main__':
    # This is how the simulator could be used:
    sim = Simulator()
    for i in range(2):
        sim.add_model(init_val=0)
    sim.step()
    sim.step({0: 23, 1: 42})
    print('Simulation finished with data:')
    for i, inst in enumerate(sim.data):
        print('%d: %s' % (i, inst))

If you run this script, you’ll get the following output:

Simulation finished with data:
0: [1, 24]
1: [1, 43]

Setup for the API implementation

So lets start implementing mosaik’s Sim API for this simulator. 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 simulator:

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

"""
import mosaik_api

import simulator

Simulator meta data

Next, we prepare the meta data dictionary that tells mosaik which models our simulator implements and which parameters and attributes it has. Since this data usually constant, we defines this at module level (which improves readability):

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

We added our “Model” model with the parameter init_val and the attributes delta and val. At this point we don’t care if they are read-only or not. 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.simulator = simulator.Simulator()
        self.eid_prefix = 'Model_'
        self.entities = {}  # Maps EIDs to model indices in self.simulator

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 initialize our actual simulator class, 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 after the simulator has been started. 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, eid_prefix=None):
        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. 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):
            eid = '%s%d' % (self.eid_prefix, i)
            self.simulator.add_model(init_val)
            self.entities[eid] = i
            entities.append({'eid': eid, 'type': model})

        return entities

The first two parameters tell you how many instances of which model you should 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 returns to mosaik the time at which it wants to do its next step. It receives its current simulation time as well as a dictionary with input values from other simulators (if there are any):

    def step(self, time, inputs):
        # Get inputs
        deltas = {}
        for eid, attrs in inputs.items():
            for attr, values in attrs.items():
                model_idx = self.entities[eid]
                new_delta = sum(values.values())
                deltas[model_idx] = new_delta

        # Perform simulation step
        self.simulator.step(deltas)

        return time + 60  # Step size is 1 minute

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

{
    'Model_0': {
        'delta': {'src_eid_0': 23},
    },
    'Model_1':
        'delta': {'src_eid_1': 42},
    },
}

The inner dictionaries containing the actual values may contain multiple entries if multiple source entities provide input for another entity. 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 + 60 tells mosaik that we wish to perform the next step in one minute (in simulation time).

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):
        models = self.simulator.models
        data = {}
        for eid, attrs in outputs.items():
            model_idx = 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(models[model_idx], attr)

        return data

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

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

The expected return value would then be:

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

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.

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

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 simulator


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


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

    def init(self, sid, eid_prefix=None):
        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):
            eid = '%s%d' % (self.eid_prefix, i)
            self.simulator.add_model(init_val)
            self.entities[eid] = i
            entities.append({'eid': eid, 'type': model})

        return entities

    def step(self, time, inputs):
        # Get inputs
        deltas = {}
        for eid, attrs in inputs.items():
            for attr, values in attrs.items():
                model_idx = self.entities[eid]
                new_delta = sum(values.values())
                deltas[model_idx] = new_delta

        # Perform simulation step
        self.simulator.step(deltas)

        return time + 60  # Step size is 1 minute

    def get_data(self, outputs):
        models = self.simulator.models
        data = {}
        for eid, attrs in outputs.items():
            model_idx = 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(models[model_idx], 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 is exactly what the next section is about.