Coordinator

The Coordinator is the main class in CoFmuPy. It controls the blocks internally in order to ease the usage of the library. For end users, this is the only interface to start with: from a JSON configuration file, the coordinator will instantiate and monitor all the required components.

from cofmupy import Coordinator

# Instantiate the Coordinator
my_coordinator = Coordinator()

# Start the Coordinator (and all its components) from a JSON configuration file
my_coordinator.start(conf_path="my_config_file.json")

The Coordinator can then run the simulation using do_step() or run_simulation(). Internally, the Master component will execute the steps of the co-simulation.

n_steps = 100
for _ in range(N):
    my_coordinator.do_step(step_size=0.05)

It is then possible to get the simulation results as a Pandas dataframe :

results_df = my_coordinator.get_results()

Source code in cofmupy/coordinator.py

def __init__(self):
    self.config_parser = None
    self.graph_engine = None
    self.master = None
    self.stream_handlers = None
    self.storage_handler: StorageHandler

do_step

do_step(step_size: float, save_data=False)

Perform a simulation step.

PARAMETER	DESCRIPTION
step_size	simulation step size TYPE: float
save_data	whether to save the data in the default CSV data storage. Defaults to False. TYPE: bool DEFAULT: False

Source code in cofmupy/coordinator.py

def do_step(self, step_size: float, save_data=False):
    """
    Perform a simulation step.

    Args:
        step_size (float): simulation step size
        save_data (bool): whether to save the data in the default CSV data storage.
            Defaults to False.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")

    # Format: [{(fmu1, var1): value}, {(fmu2, var2): value, (fmu3, var3): value}]
    data = [dh.get_data(self.master.current_time) for dh in self.stream_handlers]

    # Format: {(fmu1, var1): value}, (fmu2, var2): value, (fmu3, var3): value}
    data = {k: v for d in data for k, v in d.items()}

    # Format: {fmu1: {var1: value1}, fmu2: {var2: value2}, fmu3: {var3: value3}}
    data_for_master = defaultdict(dict)
    for (fmu, var), val in data.items():
        data_for_master[fmu][var] = [val]

    # Do step in the master
    outputs = self.master.do_step(step_size, input_dict=data_for_master)

    # Save results and data
    if save_data:
        for input_key, input_value in data_for_master.items():
            for variable_name, variable_value in input_value.items():
                outputs[input_key][variable_name] = variable_value
        self.storage_handler.notify_results(
            "file", self.master.current_time, outputs
        )

get_causality

get_causality(name: tuple) -> str

Gets the causality of the given variable.

PARAMETER	DESCRIPTION
name	variable name as (fmu_id, var_name). TYPE: tuple

RETURNS	DESCRIPTION
str	causality of the variable. TYPE: str

Source code in cofmupy/coordinator.py

def get_causality(self, name: tuple) -> str:
    """
    Gets the causality of the given variable.

    Args:
        name (tuple): variable name as (fmu_id, var_name).

    Returns:
        str: causality of the variable.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")

    return self.master.get_causality(name)

get_results

get_results() -> dict

Get the results from the simulation.

RETURNS	DESCRIPTION
dict	dataframe with the results. TYPE: dict

Source code in cofmupy/coordinator.py

def get_results(self) -> dict:
    """
    Get the results from the simulation.

    Returns:
        dict: dataframe with the results.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")
    return self.master.get_results()

get_variable

get_variable(name: tuple) -> list

Get the value of the given tuple fmu/variable.

PARAMETER	DESCRIPTION
name	variable name as (fmu_id, var_name). TYPE: tuple

RETURNS	DESCRIPTION
list	value of the variable, as a list. TYPE: list

Source code in cofmupy/coordinator.py

def get_variable(self, name: tuple) -> list:
    """
    Get the value of the given tuple fmu/variable.

    Args:
        name (tuple): variable name as (fmu_id, var_name).

    Returns:
        list: value of the variable, as a list.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")

    return self.master.get_variable(name)

get_variable_names

get_variable_names() -> list

Get the names of all variables in the system.

RETURNS	DESCRIPTION
list	list of variable names as (fmu_id, var_name) tuples. TYPE: list

Source code in cofmupy/coordinator.py

def get_variable_names(self) -> list:
    """
    Get the names of all variables in the system.

    Returns:
        list: list of variable names as (fmu_id, var_name) tuples.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")
    return self.master.variable_names

get_variable_type

get_variable_type(name: tuple) -> str

Get the type of the given variable.

PARAMETER	DESCRIPTION
name	variable name as (fmu_id, var_name). TYPE: tuple

RETURNS	DESCRIPTION
str	type of the variable. TYPE: str

Source code in cofmupy/coordinator.py

def get_variable_type(self, name: tuple) -> str:
    """
    Get the type of the given variable.

    Args:
        name (tuple): variable name as (fmu_id, var_name).

    Returns:
        str: type of the variable.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")

    return self.master.get_variable_type(name)

get_variables

get_variables(names: list) -> dict

Get the values of the given variables.

PARAMETER	DESCRIPTION
names	list of variable names as (fmu_id, var_name) to get, e.g. [("fmu1", "var3"), ("fmu2", "var1")]. TYPE: list

RETURNS	DESCRIPTION
dict	dictionary with the variable names and their values. TYPE: dict

Source code in cofmupy/coordinator.py

def get_variables(self, names: list) -> dict:
    """
    Get the values of the given variables.

    Args:
        names (list): list of variable names as (fmu_id, var_name) to get,
            e.g. [("fmu1", "var3"), ("fmu2", "var1")].

    Returns:
        dict: dictionary with the variable names and their values.
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")

    var_values = {}
    for name in names:
        var_values[name] = self.get_variable(name)

    return var_values

load_data_storages

load_data_storages(data_storages_config: dict)

Load the data storages from the given dictionary of configurations.

PARAMETER	DESCRIPTION
data_storages_config	dictionary containing the configurations for the data storages. TYPE: dict

Source code in cofmupy/coordinator.py

def load_data_storages(self, data_storages_config: dict):
    """
    Load the data storages from the given dictionary of configurations.

    Args:
        data_storages_config (dict): dictionary containing the configurations for
            the data storages.
    """
    for storage in data_storages_config.values():
        self.storage_handler.register_storage(storage["type"], storage["config"])

load_stream_handlers

load_stream_handlers(stream_handlers: dict)

Load the stream handlers from the given dictionary of configurations.

PARAMETER	DESCRIPTION
stream_handlers	dictionary containing the configurations for the stream handlers. TYPE: dict

Source code in cofmupy/coordinator.py

def load_stream_handlers(self, stream_handlers: dict):
    """
    Load the stream handlers from the given dictionary of configurations.

    Args:
        stream_handlers (dict): dictionary containing the configurations for the
            stream handlers.
    """

    self.stream_handlers = []

    for key, config in stream_handlers.items():
        var_name = config["config"].pop("variable", "")
        # Check if the stream handler already exists
        for dh in self.stream_handlers:
            if dh.is_equivalent_stream(**config["config"]):
                break
        else:
            # If not (i.e. break not executed), create a new stream handler
            dh = BaseDataStreamHandler.create_handler(config)
            self.stream_handlers.append(dh)

        # Add variable to the stream handler for mapping
        dh.add_variable(key, var_name)

parse_config

parse_config(config_path: str)

Start the configuration parser to parse the given configuration file.

PARAMETER	DESCRIPTION
config_path	path to the configuration file TYPE: str

Source code in cofmupy/coordinator.py

def parse_config(self, config_path: str):
    """
    Start the configuration parser to parse the given configuration file.

    Args:
        config_path (str): path to the configuration file
    """

    self.config_parser = ConfigParser(config_path)

run_simulation

run_simulation(
    step_size: float,
    end_time: float,
    save_data: bool = False,
)

Run the simulation until the given end time.

PARAMETER	DESCRIPTION
step_size	simulation step size TYPE: float
end_time	simulation end time TYPE: float
save_data (boolean)	whether to save data into configured storages or not

Source code in cofmupy/coordinator.py

def run_simulation(
    self, step_size: float, end_time: float, save_data: bool = False
):
    """
    Run the simulation until the given end time.

    Args:
        step_size (float): simulation step size
        end_time (float): simulation end time
        save_data (boolean) : whether to save data into configured storages or not
    """
    if self.master is None:
        raise RuntimeError("Coordinator not initialized. Call start() first.")

    while self.master.current_time < end_time:
        self.do_step(step_size, save_data=save_data)

save_results

save_results(filename: str)

Save the results to a CSV file.

PARAMETER	DESCRIPTION
filename	name of the CSV file to save the results to. TYPE: str

Source code in cofmupy/coordinator.py

def save_results(self, filename: str):
    """
    Save the results to a CSV file.

    Args:
        filename (str): name of the CSV file to save the results to.
    """
    df_results = pd.DataFrame.from_dict(self.get_results())

    # Sort the columns starting with "time" and then alphabetically
    columns = df_results.columns.tolist()
    columns.remove("time")
    columns = ["time"] + sorted(columns)

    # Set headers of the CSV file where tuple (fmu, var_name) is replaced by
    # "fmu.var_name"
    headers = list(columns)  # copy of the mutable list
    for i, col_header in enumerate(headers):
        if isinstance(col_header, tuple):
            headers[i] = f"{col_header[0]}.{col_header[1]}"

    df_results.to_csv(filename, columns=columns, header=headers, index=False)

start

start(
    conf_path: str,
    fixed_point_init=False,
    fixed_point_kwargs=None,
)

Start the coordinator with the given configuration file.

PARAMETER	DESCRIPTION
conf_path	path to the configuration file. TYPE: str
fixed_point_init	whether to use the fixed-point initialization method. TYPE: bool DEFAULT: False
fixed_point_kwargs	keyword arguments for the fixed point initialization method if fixed_point is set to True. Defaults to None, in which case the default values are used "solver": "fsolve", "time_step": minimum_default_step_size, and "xtol": 1e-5. TYPE: dict DEFAULT: None

Source code in cofmupy/coordinator.py

def start(self, conf_path: str, fixed_point_init=False, fixed_point_kwargs=None):
    """
    Start the coordinator with the given configuration file.

    Args:
        conf_path (str): path to the configuration file.
        fixed_point_init (bool): whether to use the fixed-point initialization method.
        fixed_point_kwargs (dict): keyword arguments for the fixed point initialization
            method if fixed_point is set to True. Defaults to None, in which
            case the default values are used "solver": "fsolve",
            "time_step": minimum_default_step_size, and "xtol": 1e-5.
    """

    # 1. Start ConfigParser and parse the configuration file
    self.parse_config(conf_path)

    # 2. Start GraphEngine
    self.start_graph_engine(self.config_parser.graph_config)

    # 3. Start Master
    self.config_parser.master_config["sequence_order"] = (
        self.graph_engine.sequence_order
    )
    self.start_master(
        self.config_parser.master_config,
        fixed_point_init=fixed_point_init,
        fixed_point_kwargs=fixed_point_kwargs,
    )

    # 4. Create DataStreamHandlers
    self.load_stream_handlers(self.config_parser.stream_handlers)

    # 5. Create DataStorages
    self.storage_handler = StorageHandler()
    self.load_data_storages(self.config_parser.data_storages)

start_graph_engine

start_graph_engine(config: dict)

Start the graph engine with the given configuration.

PARAMETER	DESCRIPTION
config	configuration for the graph engine containing the FMUs, connections, and edge separation. TYPE: dict

Source code in cofmupy/coordinator.py

def start_graph_engine(self, config: dict):
    """
    Start the graph engine with the given configuration.

    Args:
        config (dict): configuration for the graph engine containing the FMUs,
            connections, and edge separation.
    """
    self.graph_engine = GraphEngine(
        config["fmus"],
        config["symbolic_nodes"],
        config["connections"],
        config["edge_sep"],
    )

start_master

start_master(
    config: dict,
    fixed_point_init=False,
    fixed_point_kwargs=None,
)

Start the master algorithm with the given configuration.

PARAMETER	DESCRIPTION
config	configuration for the master algorithm containing the FMUs, connections, sequence order, and loop method. TYPE: dict
fixed_point_init	whether to use the fixed-point initialization method. TYPE: bool DEFAULT: False
fixed_point_kwargs	keyword arguments for the fixed point initialization method if fixed_point is set to True. Defaults to None, in which case the default values are used "solver": "fsolve", "time_step": minimum_default_step_size, and "xtol": 1e-5. TYPE: dict DEFAULT: None

Source code in cofmupy/coordinator.py

def start_master(
    self, config: dict, fixed_point_init=False, fixed_point_kwargs=None
):
    """
    Start the master algorithm with the given configuration.

    Args:
        config (dict): configuration for the master algorithm containing the FMUs,
            connections, sequence order, and loop method.
        fixed_point_init (bool): whether to use the fixed-point initialization method.
        fixed_point_kwargs (dict): keyword arguments for the fixed point initialization
            method if fixed_point is set to True. Defaults to None, in which
            case the default values are used "solver": "fsolve",
            "time_step": minimum_default_step_size, and "xtol": 1e-5.
    """
    self.master = DefaultMaster(
        fmu_config_list=config["fmus"],
        connections=config["connections"],
        sequence_order=config["sequence_order"],
        cosim_method=config["cosim_method"],
        iterative=config["iterative"],
        fixed_point=fixed_point_init,
        fixed_point_kwargs=fixed_point_kwargs,
    )
    self.master.init_simulation(input_dict={})