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 i ∈ N, i > 0, delta ∈ Z
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.
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 is usually constant, we define 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()
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 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.