Configuration of a Simulator

This page explains the structure of a simulator.py to enable you to create your own simulators. The simulator.py stores information about the connected models and mediates between them and mosaik (e.g. passing the order to step() as well as associated values). As there may be multiple instances of a model connected to a single simulator, you only need one simulator of a type per scenario, set within the SIM_CONFIG of scenario file. A simulator.py is always paired with a model.py.

Note

All code-blocks derive from charging_station_simulator.py as of (01/24) if not stated otherwise.

Introduction and imports

Basic explanation import of relevant packages

"""
Mosaik interface for the eELib charging station model.
Simulator for communication between orchestrator (mosaik) and charging station entities.

Author: elenia@TUBS
"""

import mosaik_api_v3
from eelib.core.devices.charging_station import charging_station_model
import eelib.utils.validation as vld
from copy import deepcopy

---

Listing of model META

State of simulator (whether it is time-discrete or event-based/hybrid)

META = {
   "type": "hybrid",

Listing of ALL attributes for each modeltype: At first, all attributes …

   "models": {
      "ChargingStation": {
         "public": True,
         "params": ["init_vals"],
         "attrs": [
            "type",
            "output_type",
            "step_size",
            "time",
            ...

… input attributes also listed in "trigger" list …

         "trigger": [
            "p_set",
            "e_bat",
            "e_bat_max",
            "p_charge_max",
            "p_discharge_max",
            "appearance",
            "appearance_end_step",
            "bev_consumption_period",
         ],  # input attributes

… output attributes also listed in ""non-persistent" list.

         "non-persistent": [
            "p_min",
            "p_max",
            "appearance_end_step",
            "p_device",
            "p",
            "discharge_efficiency",
            "charge_efficiency",
            "e_bat",
            "e_bat_max",
         ],  # output attributes

---

Initialization of simulator class

Short explanation as well as constructor __init__ for this simulator and the initialization function init()

class Sim(mosaik_api_v3.Simulator):
    """Simulator class for eELib charging station model.

    Args:
        mosaik_api_v3 (module): defines communication between mosaik and simulator

    Raises:
        ValueError: Unknown output attribute, when not described in META of simulator
    """

    def __init__(self):
        """Constructs an object of the Charging-Station:Sim class."""

        super(Sim, self).__init__(META)

        # storing of event-based output info (for same-time loop or next time step)
        self.output_cache = {}

        # initiate empty dict for model entities
        self.entities = {}

    def init(self, sid, scenario_config, time_resolution=1.0):
        """Initializes parameters for an object of the Charging-Station:Sim class.

        Args:
            sid (str): Id of the created instance of the simulator (e.g. CSSim-0)
            scenario_config (dict): scenario configuration data, like resolution or step size
            time_resolution (float): Time resolution of current scenario.

        Returns:
            meta: description of the simulator
        """

        # assign properties
        self.sid = sid
        self.scenario_config = scenario_config

        return self.meta

---

Following are the so called core functions create(), step() and get_data() that are found in every simulator.py.

Caution

All core functions of the simulators are called by mosaik and should not be deleted!

---

Creation of model entities in create()

Core functions should also be labeled in their docstring

def create(self, num, model_type, init_vals):
    """Creates entities of the eELib charging station model.
    Core function of mosaik.

    Args:
        num (int): Number of cs models to be created
        model_type (str): type of created instance (e.g. "charging_station")
        init_vals (list): list with initial values for each charging_station entity

    Returns:
        dict: created entities
    """

Creation of entities by assigning individual entity names and calling the initialization method of that model type

# generated next unused ID for entity
next_eid = len(self.entities)

# create empty list for created entities
entities_orchestrator = []

for i in range(next_eid, next_eid + num):
    # create entity by specified name and ID
    ename = "%s%s%d" % (model_type, "_", i)
    full_id = self.sid + "." + ename

    # get class of specific model and create entity with init values after validation
    entity_cls = getattr(charging_station_model, model_type)
    vld.validate_init_parameters(entity_cls, init_vals[i])
    entity = entity_cls(
        ename,
        **init_vals[i],
        step_size=self.scenario_config["step_size"],
    )

Simulator stores information about the entities

    # add info to the simulators entity-list and current entities
    self.entities[ename] = {
        "ename": ename,
        "etype": model_type,
        "model": entity,
        "full_id": full_id,
    }
    entities_orchestrator.append({"eid": ename, "type": model_type})

return entities_orchestrator

---

Stepping of models in step()

The step() method of storage_simulator.py (01/24).

First take input data (from mosaik) and set values of entities

# assign property values for each entity and attribute with entity ID
# process input signals: for the entities (eid), attr is a dict for attributes to be set
for eid, attrs in inputs.items():
    # for the attributes (attr), setter is a dict for entities with corresponding set values
    for attr, setter in attrs.items():
        # for transmitter (eid_setter), value_dict contains set values (with ids, when dict)
        setting_value_dict = deepcopy(getattr(self.entities[eid]["model"], attr))
        for eid_setter, value_dict in setter.items():
            if isinstance(value_dict, dict):
                # go by each id and search for corresponding entity id
                for getter_id, value in value_dict.items():
                    if eid in getter_id:
                        setting_value_dict[eid_setter] = value
            # value_dict is not a dict, only a single value -> write directly
            elif isinstance(value_dict, (float, int)):
                setting_value_dict[eid_setter] = value_dict
            else:
                raise TypeError("Unknown format for value_dict")
        setattr(self.entities[eid]["model"], attr, setting_value_dict)

        # check if there is more than one power set value - otherwise directly set it
        if attr == "p_set":
            if len(setting_value_dict) > 1:
                raise ValueError("There is more than one power set value for " + eid)

Then simply step each model for this time step

# call step function for each entity in the list
for ename, entity_dict in self.entities.items():
    entity_dict["model"].step(time)

Note

You might return the next timestep for when this model should be called again.

---

Handling of output data in get_data()

From a defined set of output properties, the values of the transmitting entities are read and stored into data dict

for transmitter_ename, attrs in outputs.items():
    # get name for current entity and create dict field
    entry = self.entities[transmitter_ename]
    if transmitter_ename not in self.output_cache:
        self.output_cache[transmitter_ename] = {}

    # loop over all targeted attributes and check if info is available
    for attr in attrs:
        if attr not in self.meta["models"][type(entry["model"]).__name__]["attrs"]:
            raise ValueError("Unknown output attribute: %s" % attr)

        # create empty field for cache and output data
        if attr not in self.output_cache[transmitter_ename]:
            self.output_cache[transmitter_ename][attr] = {}

        output_data_to_save = getattr(entry["model"], attr)

For each time step, the output data is continuously stored and compared to the lastly sent (output_cache) such that if nothing new is to be send out, only the time step will be send

# check if nothing is to be send out - send output 1 step later to avoid waiting for data
if not flag_output_changed:
    if self.time == self.scenario_config["n_steps"] - 1:  # is last time step?
        data["time"] = self.time + 1
    else:
        data = {"time": self.time}