Example of farn API usage¶

The following gives an example of how the farn API can be used in conjunction with OSP cosim and the ospx package.

Case¶

Case is a simple dataclass designed to hold case-specific information as instance attributes.
Below example uses the most typical / important ones:

from pathlib import Path
from typing import Dict, List
from farn.core import Case, Parameter

case_name: str = "test_001"

case_folder: Path = Path("path/to/your/experiment/cases") / case_name

case_parameters: List[Parameter] = [
    Parameter("x1", 1.0),
    Parameter("x2", 2.0),
    Parameter("x3", 3.0),
]

command_sets: Dict[str, List[str]] = {
    "prepare": [
        "copy path/to/template/dir/caseDict .",  # copy a (case-agnostic) ospx config file from a template directory into the case folder
        "dictParser caseDict",  # parse the ospx config file. This will make it case-specific
        "ospCaseBuilder parsed.caseDict",  # build the (case-specific) OspSystemStructure.xml
    ],
    "run": [
        "cosim run OspSystemStructure.xml -b 0 -d 10",  # run OSP cosim
    ],
    "post": [
        "watchCosim -d watchDict",  # optional post-processing. watchCosim creates a sub-folder 'results' in the case folder
    ],
}

case: Case = Case(
    case=case_name,
    path=case_folder,
    is_leaf=True,
    parameters=case_parameters,
    command_sets=command_sets,
)

Cases¶

Many farn functions expect not a single Case object, but a list of Case objects passed in.
So, even if you create only a single Case object, you will likely need to wrap it in a Cases list for processing.

from farn.core import Cases

cases: Cases = Cases()
cases.append(case)

Create case folder(s)¶

from farn.farn import create_case_folders

_ = create_case_folders(cases)

Create paramDict file in case folder(s)¶

It is a convention used by both farn and ospx that a ‘paramDict’ file is expected to exist in the case folder which contains the case-specific values of all varied parameters.
create_param_dict_files() will create this paramDict file in the case folder(s):

from farn.farn import create_param_dict_files

_ = create_param_dict_files(cases)

Prepare OSP simulation case(s)¶

Execute the command_set that performs preparatory tasks in your case folder(s).

from farn.farn import execute_command_set

_ = execute_command_set(
    cases=cases,
    command_set="prepare",
)

Run OSP simulation case(s)¶

Execute the command_set that actually runs your simulation case(s).

_ = execute_command_set(
    cases=cases,
    command_set="run",
)

Execute post-processing (optional)¶

Execute the command_set that performs post-processing tasks in your case folder(s) (if any).

_ = execute_command_set(
    cases=cases,
    command_set="post",
)

Read results¶

Reading results after your simulation cases finished can be rather individual and there will surely be multiple ways to accomplish this. Two exemplary ways being

a) You can read the *.csv files that cosim.exe generates, i.e. by reading them into a Pandas DataFrame.

b) In case you executed watchCosim as a post-processing step, a sub-folder ‘results’ will exist in each case folder that contains the results additionally as a resultDict file.

Variant a) is less complex and can be done with Pandas (no example given here).
Variant b) offers some additional possibilities but requires also additional code.
Below hence an example for variant b), i.e. how to read a resultDict file generated by watchCosim

from typing import Any, Dict, List
from pandas import DataFrame, Series
from dictIO import DictReader
from dictIO.utils.path import relative_path

# @NOTE: Adjust 'component_name' and 'variable_name' to what is defined in your (FMU) model
# @NOTE: 'y' is just an example. Add mappings for more column names / variables as you like and need.
mapping: Dict[str, Dict[str, Any]] = {
    "y": {  # column name you want to map a variable to
        "key": "component_name|variable_name:latestValue",  # variable in FMU
        "unit": 1,  # usually 1, unless you want to apply a scaling factor
    }
}

# column names
names: List[str] = [name for name in mapping if not re.search("(^_|COMMENT)", name)]

series: Dict[str, Series] = {
    "path": Series(data=None, dtype=dtype(str), name="path"),
}

for index, case in cases:

    case_folder: Path = case.path
    result_folder: Path = case_folder / "results"
    result_dict_file: Path = result_folder / "watchDict-test_project-resultDict"  # adapt to output of watchCosim.

    series["path"].loc[index] = str(relative_path(Path.cwd(), case_folder))

    result_dict = DictReader.read(result_dict_file, includes=False, comments=False)

    for name in names:
        value: Any = None
        value_eval_string = "result_dict['" + "']['".join(mapping[name]["key"].split(":")) + "']"
        try:
            value = eval(value_eval_string)
        except Exception:
            logger.warning(f'"{value_eval_string}" not in {result_dict_file}')
            continue

        if name not in series:
            series[name] = Series(data=None, dtype=dtype(type(value)), name=name)

        if value is not None:
            series[name].loc[index] = value

df: DataFrame = DataFrame(data=series)

df.set_index("path", inplace=True)  # optional. Makes the 'path' column the DataFrame's index