The CS-FMU-Mapper is a Python-based co-simulation framework that enables seamless integration and mapping between FMUs or OPCUA nodes and other simulation components. It provides a flexible configuration system to define component interactions, data mappings, and simulation scenarios.
Create a new Python environment and install the required dependencies:
> $ conda create -n cs-fmu-mapper python=3.10
> $ conda activate cs-fmu-mapper
> $ conda install -c conda-forge pyfmi
> $ pip3 install git+https://github.com/marvin-schl/cs-fmu-mapper.gitAlternativley you can install after cloning the repo for development purposes:
> $ conda create -n cs-fmu-mapper python=3.10
> $ conda activate cs-fmu-mapper
> $ conda install -c conda-forge pyfmi
> $ pip3 install -r requirements.txtThis repository can be installed as a package via pip. The package has not yet been uploaded to PyPI. Therefore, the package has to be installed locally. By cloning the repository and executing the following command in the root directory of the repository the package will be installed. All dependencies will be installed automatically but note that pyfmi has to be installed manually as it cannot be installed via pip.
> $ pip install -e . > $ python -m cs_fmu_mapper.main -c example/configs/config.yamlThis will run the simulation with the configuration given in example/config.yaml and will save the results in the example/results directory.
The cs-fmu-mapper can be run directly from the command line or imported as a module in your own Python code.
> $ python -m cs_fmu_mapper.main -c example/configs/config.yaml -md example/configs/scenario --debugThe arguments are the following:
-c or --config_path: Path to the configuration file.-md or --module_dir: Path to the directory containing the modular config files (only needed if modular config is used).-d or --debug: Run in debug mode.For simplicity, create a own python file and import the CSFMUMapper class from the cs_fmu_mapper module and call the run() method with the desired arguments.
import asyncio
from cs_fmu_mapper.main import CSFMUMapper
mapper = CSFMUMapper(config_path, module_dir, debug)
asyncio.run(mapper.run())Configuration is done in YAML format. The configuration file can also be formatted as JSON but YAML is recommended as it supports commenting. A detailed commented example configuration is given here.
The configuration basically defines simulation components of type plc, fmu, logger and/or scenario. Each components' section defines ouputVars and/or inputVars by specifying their component-specific access string called nodeID, e.g. the OPCUA NodeID for a plc component or the CSV column name of a scenario component. Configuration of scenario and logger components are optional. Configuration of plc component is also optional but only makes sense if there is at least a scenario or another custom component configured which interacts with the fmu.
Besides the component configuration, there is a Mapping section where the Mapping from outputVars to inputVars is configured. For a detailed explanation see the example config. The simulation will be performed in steps. If a mapping is configured as preStepMapping the mapping will be done before simulation step execution. Consequently, when a mapping is configured as postStepMapping the mapping will be done afterwards.
If there is a plc component configured the plc will be the simulation master and will trigger each simulation step. If there is no plc configured the software will simulate standalone with the configured step size timeStepPerCycle configured in the Mapping section. If every component signalizes that it is finished then every component will be notified that the simulation is finished. In standalone mode, the simulation will then finalize itself. In plc master mode the plc should react accordingly and should initiate the termination of the program.
The ConfigurationBuilder class takes the following arguments:
config_file_path: Absolute or relative path to the configuration file.module_dir: Absolute or relative path to the modular config files.pre_build_injections: (optional) Additional dictionary to inject before the build process.post_build_injections: (optional) Additional dictionary to inject after the build process.Modular configurations allow for a more flexible and scalable configuration of simulations. Instead of defining all components in a single file, you can split the configuration into multiple files and directories. This approach enhances reusability and maintainability of the configuration.
An Example for a modular configuration is given here. Modular configurations are enabled by setting the modular_config flag to true. Otherwise the configuration is treated as a full config.
The modular config is technically split into two parts by the # END_COMPONENT_DEFINITIONS tag:
Components key.The modular configs that are imported from the Components section can also be jinja2 templates. This allows for a high degree of flexibility as the components can be parametrized. For example the inputVar and outputVar can be transformed from other component's outputVar and inputVar respectively. For example:
# modular_config.yaml
...
# END_COMPONENT_DEFINITIONS
Algorithm:
inputVar:
{{ transform_vars(Model.outputVar, prefix='algo', direction='in') }}
outputVar:
{{ transform_vars(Model.inputVar, prefix='algo', direction='out') }}The transform_vars function is a function that is used to transform the variables (see ConfigurationBuilder).
# model.yaml
Model:
inputVar:
model.in.u:
nodeID: u
outputVar:
model.out.y:
nodeID: yThe full config will then look like this:
# full_config.yaml
...
Model:
inputVar:
model.in.u:
nodeID: u
outputVar:
model.out.y:
nodeID: y
Algorithm:
inputVar:
algo.in.y:
nodeID: y
outputVar:
algo.out.u:
nodeID: u
...The ListMergeMode is an enum that determines how lists should be merged. The default is EXTEND_UNIQUE which extends the list by adding unique elements and is the most useful mode. The other modes are EXTEND which extends the list by adding all elements and REPLACE which replaces the list with the new list. For more information see the OmegaConf documentation.
In order to change the ListMergeMode for a specific injection just add the list_merge_mode key to the injection dictionary. For example:
injection:
list_merge_mode: EXTEND # or EXTEND_UNIQUE or REPLACE. Defaults to EXTEND_UNIQUE (if not specified)
...Mappings are used to map the output variables of one component to the input variables of another component. The mapping is configured in the Mapping section of the configuration file. For an example see example/configs/config.yaml and the OPCUAFMUMapper class.
The mapping is done in two steps:
preStepMappings are applied before every simulation step.postStepMappings are applied after every simulation step.The mappings are configured as a dictionary where the keys are the names of the source components and the values are lists of the input variables of the destination component that shall be mapped to the output variables of the source component. For example:
Mapping:
preStepMappings:
algo.out.TIR:
- model.in.TIR
postStepMappings:
model.out.TIR:
- algo.in.TIRThe above configuration will map the TIR output variable of the algo component to the TIR input variable of the model component before every simulation step and the TIR output variable of the model component to the TIR input variable of the algo component after every simulation step.
It is also possible to map multiple input variables to the same output variable. This can be useful if the same input variable is needed in different components (e.g. for plotting). In this case, just add multiple entries to the list like this:
Mapping:
preStepMappings:
algo.out.TIR:
- model.in.TIR
- plot.in.TIRScenarios are instructions for the simulation or other components to follow. The scenario is configured in the Scenario section of the configuration file. For an example see example/configs/config.yaml and the Scenario class. The scenario component is optional.
It takes two types of inputs:
generate_schedule method. This method takes some kwargs defined in the parameters section of the scenario configuration and returns a pandas dataframe with at least a t column containing the simulation time and the new values for the scenario inputs. The columns of the dataframe must be named after the scenarios output variables. The parameters section must contain the path to the python file as key and the kwargs as values.t column containing the simulation time and the columns containing the new values for the scenario outputs.If the mappings are configured correctly the scenarios outputs will be mapped to the other components inputs.
[!NOTE] If multiple scenarios are defined the
scenariocomponent will automatically concatenate and overwrite them. The order of the scenarios in the configuration file defines the order in which they are concatenated / overwritten.Plotter
The plotter is a component that can be used to plot the output variables of the simulation. It is configured in the Plotter section of the configuration file. For an example see example/configs/config.yaml and the Plotter class.
A custom simulation component can be created by inheriting from SimulationComponent. The inherited class should define a class variable type which value determines the value of the type field in the configuration file. Also, all abstract methods from SimulationComponent have to be implemented. The constructor should take two arguments the corresponding section of the configuration as dict and a name. A minimal simulation component could look like:
from simulation_component import SimulationComponent
class CustomComponent(SimulationComponent):
type = "custom"
def __init__(self, config, name):
pass
def do_step(self, t, dt):
passTo get an overview of the methods that need to be implemented, see the SimulationComponent class.
To integrate the custom component into the simulation, it has to be added to the configuration file. The component will then automatically be initialized and the do_step() method will be executed in every simulation step. To add a custom component to the configuration file, add it under the customComponents section in the General key. For example:
General:
customComponents:
CustomComponentClass:
pathToComponent: path.to.custom.componentThe key CustomComponentClass is the class name of the custom component and the pathToComponent is the path to the custom component class.
SimulationComponent buffers the mapped input and output values in two dictionaries. One for input and the other for output values. There are getters and setters for getting and setting single values or the whole dict. The two dicts shall only be accessed by those getters and setters:
# for setting/getting the while input dict
def get_input_values(self)
def set_input_values(self, new_val)
# for setting/getting single values in the input dict
def get_input_value(self, name)
def set_input_value(self, name, new_val)
# for setting/getting the whole output dict
def get_output_values(self)
def set_output_values(self, new_val)
# for setting/getting single values in the output dict
def get_output_value(self, name)
def set_output_value(self, name, new_val)
The implementation of doStep() should read the component's output values and write them into the SimulationComponents output buffer via the above-mentioned methods. Accordingly, it should read the input buffer and write them to the component inputs. The name arguments of the above methods represent the configured unique keys of the outputVar/inputVar section of the component. If no outputVar or inputVar are configured the according getters and setters raise a NotImplementedError.
Optionally the component can implement the following methods:
def is_finished(self)
def notify_simulation_finished(self)
def finalize(self)If the component shall affect the end of a simulation its is_finished() method should return false as long as it isn't finished. The default return value if not implemented is true. As soon as every component finishes its execution the notify_simulation_finshed() callback is called. The finalize() callback is called shortly before the termination of the program. Any open connections could be closed here, or occupied memory could be freed here for example.
For logging purposes, these three methods can be used:
def log_info(self, msg)
def log_debug(self, msg)
def log_warning(self, msg)The experiment runner allows for running multiple preconfigured simulations in a row. This can be useful to compare the results of different simulations without having to manually change the configuration and run the simulation multiple times.
This is done with the ExperimentRunner class. For an example, execute the following command:
> $ python -m cs_fmu_mapper.experiment_runnerThis will run all the experiments specified in the run.yaml file, that where defined in the experiments.yaml file and save the results in the example/results directory. The base_config.yaml is used as the base configuration for all the experiments which will be extended by the parameters defined in the experiments.yaml file. There, the output_folder is the directory where the results will be saved. Each experiment will create a subdirectory in the output_folder with the name of the experiment. The pre_build_injections and post_build_injections sections of the experiment configuration will be used to overwrite the corresponding sections of the base configuration. See the ConfigurationBuilder class for more information about the injections.
The experiment runner takes the following arguments:
-bc or --base_config: Path to the base configuration file (which gets extended by the experiments).-md or --module_dir: Path to the directory containing the module configurations (for the modular config).-ed or --experiments_dir: Path to the directory containing the experiment configurations (directory containing the experiments.yaml and run.yaml files).-ef or --experiments_file: Name of the experiments definition file. Defaults to experiments.yaml.-rf or --run_file: Name of the run definition file. Defaults to run.yaml.-td or --temp_dir: Name of the temporary directory for generated configs. Defaults to temp.-wd or --working_dir: Working directory for experiment execution.-d or --debug: Run in debug mode.The scheduler allows for customizable scenarios based on time-value patterns. This is done by defining a scenario path that begins with Scheduler in the Scenario component configuration and defining the parameters in the parameters section. For an example see customScenario.yaml. In the scenario component configuration the parameters are used to define the parameters of the Scheduler class.
The scheduler takes the following arguments:
duration: Duration of the schedule in seconds. It is used to determine the length of the schedule. The patterns are repeated over the duration. Defaults to 4 days.items: List of items to be scheduled. It is used to define the column names of the schedule (the column_prefix parameter of the Scheduler class is automatically prepended to each item to allow for custom column names). Defaults to DEFAULT_ITEMS.times: List of times at which the schedule changes. It is a list of integers and is used to define the times at which the schedule changes. If not specified the times are automatically determined by the length of the patterns and will be evenly spaced over the interval of the time unit (hours: over 24 hours, minutes: over 60 minutes, seconds: over 60 seconds). Defaults to None.patterns: Dictionary containing the time-value patterns. It is used to define the time-value patterns and must be of the same length as the times parameter as each pattern is applied at the corresponding time. Defaults to DEFAULT_PATTERNS.time_unit: Unit of the times parameter. Can be either "days", "hours", "minutes" or "seconds". Defaults to "hours".column_prefix: Prefix for the column names of the schedule. Defaults to "scen.out.".start_time: Start time of the schedule. Defaults to 0.For example the following configuration:
Scenario:
path:
- Scheduler: CustomScenario
parameters:
CustomScenario:
duration: 7200
items: ["u"]
times: [0, 30, 59]
patterns: {"1": [1, 0, 1]}
time_unit: minutes
column_prefix: "custom_"
start_time: 3600This will create a schedule where the value of u changes at times 1:00, 1:30 and 1:59, following the pattern [1, 0, 1]. This pattern starts at 3600 seconds (1:00) and repeats over the duration of 2 hours (7200 seconds).
In this example, notice that the pattern uses "1" as its key instead of "u". When pattern keys don't match the item names in the items list, the scheduler will map them in order. So here, the pattern "1": [1, 0, 1] gets mapped to the item "u".
This mapping behavior is particularly useful when running multiple simulations that use the same patterns but with different item names. You can keep the patterns constant in your scenario file and only update the items list to change which items receive those patterns.
Contributions are welcome! If you find a bug or have a feature request, please open an issue on GitHub. If you want to contribute code, please fork the repository and create a pull request.