create_benchmark

The create_benchmark.py script enables the automatic generation of datasets, based on the Cyber-Physical Process Plant setups saved in the benchmark_config.json file.

The script sequentially calls the following functionalities:
  • create_ds_dir(setup: str)

    creating a dataset directory, which is named after the dataset name defined in the benchmark_config.json and in which the results will be saved.

  • create_call_file(setup: str)

    creating a customized call.mos file for running the custom simulation and saving the results in a custom directory.

  • create_process_plant(config_ds: dict)

    creating a customized process_plant.mo file, based on the Cyber-Physical Process Plant defined in the benchmark_config.json.

  • create_supermodel(faulty_module: str, fault: str)

    creating a customized superModel of the faulty module, to induce time-dependent faults.

  • run_simulation(setup: str, fault: str)

    run individual simulations for all faults indicated in the benchmark_config.json.

  • clean_ds(setup: str, fault: str)

    clean the dataset from unnecessary values and only keep values of interest.

  • save_connection_model(config_dict: dict, setup: str)

    saving the Cyber-Physical Process Plant configuration as *.json in the custom dataset directory.

  • create_dataset(config_ds: dict, setup: str)

    running all functionality calls in a function.

Functions

  • def create_ds_dir(setup: str)

    creating a dataset directory, which is named after the dataset name defined in the benchmark_config.json and in which the results will be saved.

    1def create_ds_dir(setup: str):
2    os.system("cd ../datasets && mkdir " + setup)
3    return print("Directory ./datasets/" + setup + " was created ...\n")
  • def create_call_file(setup: str)

    creating a customized call.mos file for running the custom simulation and saving the results in a custom directory.

     1def create_call_file(config_dict: dict, setup: str):
 2    """
 3    creating a call.mos file with updated dataset directory
 4    """
 5    file = open("call.mos", "r")
 6    replacement = ""
 7    for line in file:
 8        if 'cd("ds' in line:
 9            changes = 'cd("' + setup + '"); \n'
10        elif 'simulate(' in line:
11            changes = 'simulate(process_plant, startTime =' + str(config_dict["startTime"]) + ' , stopTime=' + str(config_dict["stopTime"]) + ' , numberOfIntervals=' + str(config_dict["numberOfIntervals"]) + ' , outputFormat="csv", simflags="-maxIntegrationOrder=1");\n'
12        else:
13            changes = line
14        replacement = replacement + changes
15    file.close()
16    # write mode
17    fout = open("call.mos", "w")
18    fout.write(replacement)
19    fout.close()
20    return
  • def create_process_plant(config_ds: dict)

    creating a customized process_plant.mo file, based on the Cyber-Physical Process Plant defined in the benchmark_config.json.

     1def create_process_plant(config_dict: dict):
 2    """
 3    Creates a process plant model in modelica as .mo file,
 4    based on the setup, that is encoded in the config_ds dict.
 5
 6    The file is saved as process_plant.mo
 7    """
 8
 9    intro = ''\
10            'model process_plant\n' \
11            'inner Modelica.Fluid.System system(energyDynamics = Modelica.Fluid.Types.Dynamics.FixedInitial, m_flow_start = 1, massDynamics = Modelica.Fluid.Types.Dynamics.FixedInitial, momentumDynamics = Modelica.Fluid.Types.Dynamics.FixedInitial, p_ambient(displayUnit = "Pa"));\n' \
12            'replaceable package Medium = Modelica.Media.Water.StandardWater;\n\n'
13    models = ''
14    equations = '\nequation\n'
15    outro = 'end process_plant;\n'
16
17    mod_list = config_dict["model"]["nodes"]
18    equ_list = config_dict["model"]["edges"]
19
20    for i in mod_list:
21        models = models + (i[:-1] + " " + i + ";\n") #"(replaceable package Medium = Medium);\n")
22
23    l, m = 1, 1
24    for j in equ_list:
25        if "mixer" in j[1]:
26            equations = equations + ("connect(" + str(j[0]) + ".port_out, " + str(j[1]) + ".port_in" + str(l) + ");\n")
27            l += 1
28        if "still" in j[0]:
29            equations = equations + ("connect(" + str(j[0]) + ".port_out" + str(m) + ", " + str(j[1]) + ".port_in);\n")
30            m += 1
31        if "mixer" not in j[1] and "still" not in j[0]:
32            equations = equations + ("connect(" + str(j[0]) + ".port_out, " + str(j[1]) + ".port_in);\n")
33
34    with open("../simulation_models/process_plant.mo", "w") as f:
35        f.write(intro)
36        f.write(models)
37        f.write(equations)
38        f.write(outro)
39    return
  • def create_supermodel(faulty_module: str, fault: str)

    creating a customized superModel of the faulty module, to induce time-dependent faults.

     1def create_supermodel(config_dict: dict, fault: str):
 2    """
 3    Within the supermodels fault induction can be turned off and on.
 4    Hence, for each fault setup a separate supermodel must be created.
 5    """
 6    str_superModel = "../simulation_models/" + str(config_dict["faulty_module"][0][:-9]) + "_superModel.mo"
 7    ok_str1 = "valve_leaking_simulator = if time >=" + str(config_dict["stopTime"]) + " then 0.0001 else 0; // leaking"
 8    nok_str1 = "valve_leaking_simulator = if time >=" + str(config_dict["errorInduction"]) + " then 0.0001 else 0; // leaking"
 9    ok_str2 = "valve_clogging_simulator = if time >=" + str(config_dict["stopTime"]) + " then 0.01 else 1; // clogging"
10    nok_str2 = "valve_clogging_simulator = if time >=" + str(config_dict["errorInduction"]) + " then 0.01 else 1; // clogging"
11
12    # replace all strings with ok strings before introducing nok strings
13    file = open(str_superModel, "r")
14    replacement = ""
15    for line in file:
16        if "valve_leaking_simulator =" in line:
17            changes = ok_str1
18        elif "valve_clogging_simulator =" in line:
19            changes = ok_str2
20        else:
21            changes = line
22        replacement = replacement + changes
23    file.close()
24    fout = open(str_superModel, "w")
25    fout.write(replacement)
26    fout.close()
27
28    # read mode
29    file = open(str_superModel, "r")
30    replacement = ""
31    for line in file:
32        if fault == "l":
33            if ok_str1 in line:
34                changes = line.replace(ok_str1, nok_str1)
35            elif nok_str2 in line:
36                changes = line.replace(nok_str2, ok_str2)
37            else:
38                changes = line
39        elif fault == "c":
40            if nok_str1 in line:
41                changes = line.replace(nok_str1, ok_str1)
42            elif ok_str2 in line:
43                changes = line.replace(ok_str2, nok_str2)
44            else:
45                changes = line
46        elif fault == "lc":
47            if ok_str1 in line:
48                changes = line.replace(ok_str1, nok_str1)
49            elif ok_str2 in line:
50                changes = line.replace(ok_str2, nok_str2)
51            else:
52                changes = line
53        replacement = replacement + changes
54    file.close()
55    # write mode
56    fout = open(str_superModel, "w")
57    fout.write(replacement)
58    fout.close()
59    return
  • def run_simulation(setup: str, fault: str)

    run individual simulations for all faults indicated in the benchmark_config.json.

     1def run_simulation(setup: str, fault: str):
 2    """
 3
 4    This file runs the simulation by calling the call.mos file from console
 5    saves the results .csv as setup + fault .csv
 6    removes other unnecessary simulation files
 7    """
 8    os.system("omc call.mos")
 9
10    # delete unnecessarey simulation information and rename file
11    os.system("cd ../datasets/" + setup + " && mv process_plant_res.csv " + setup + fault + ".csv")
12    filelist = [f for f in os.listdir("../datasets/" + setup) if not f.endswith(".csv")]
13    for f in filelist:
14        os.remove(os.path.join("../datasets/" + setup, f))
15    return
  • def clean_ds(setup: str, fault: str)

    clean the dataset from unnecessary values and only keep values of interest.

     1def clean_ds(setup: str, fault: str):
 2    """
 3    reading csv as df and cleaning everything unnecessary
 4    """
 5    csv_file = "../datasets/" + setup + "/" + setup + fault + ".csv"
 6    channels_of_interest = ["time", "v_flow", "level", "m_flow", "fluidVolume", "N_in", "opening", "heatTransfer.Ts", "medium.t", "port_a.p", "port_b.p"]
 7
 8    df = pd.read_csv(csv_file)
 9
10    for column in df:
11        flag = False
12        for i in channels_of_interest:
13            if i in column:
14                flag = True
15        if flag == False:
16            df = df.drop(column, 1)
17
18    df.to_csv(csv_file)
19    print("dataset " + csv_file + " was created ...\n")
20    return
  • def save_connection_model(config_dict: dict, setup: str)

    saving the Cyber-Physical Process Plant configuration as *.json in the custom dataset directory.

     1def save_connection_model(config_dict: dict, setup: str):
 2    """
 3
 4    saves the simulation configuration as .txt file in dataset directory
 5    """
 6    file = "../datasets/" + setup + "/" + setup + "_config.json"
 7    with open(file, "w") as json_file:
 8        json.dump(config_dict, json_file)
 9
10    print("Saved config-file ... \n")
11    return
  • def create_dataset(config_ds: dict, setup: str)

    running all functionality calls in a function.

     1def create_dataset(config_ds: dict, setup: str):
 2    """
 3    This funcion creates a single .csv dataset by
 4        - building a modelica model and supermodel of the desired process plant setup (including the adapted parametrization)
 5        - running the simulation (regarding to the desired fault setup)
 6        - cleaning the dataset from unnecessary values
 7        - and saving it in a separate directory
 8
 9    in:
10        - a dict of the current ds config
11
12    returns:
13        - a csv in a directory named after the process plant build
14    """
15
16    create_ds_dir(setup=setup)                                          # creating a directory for the dataset setup
17    create_call_file(config_dict=config_ds, setup=setup)                # creating an adapted call.mos file to run Modelica simulations
18    create_process_plant(config_dict=config_ds)                         # creating a process plant setup following the config file
19
20    for fault in config_ds["faults"]:
21        create_supermodel(config_dict=config_ds, fault=fault)           # creating a supermodel inducing the fault "fault"
22        run_simulation(setup=setup, fault=fault)                        # running the om simulation
23        clean_ds(setup=setup, fault=fault)                              # cleaning resulting ds from unnecessary values
24        print("simulation run " + str(setup) + str(fault) + " finished ... \n")
25    save_connection_model(config_dict=config_ds, setup=setup)           # saving the connection model and simulation setup as file in ds-directory
26    return