From 0df3d9ced743ac3385dd710c7133a6cf369b051c Mon Sep 17 00:00:00 2001 From: Radu Nicolae Date: Mon, 16 Jun 2025 18:01:07 +0200 Subject: integrated M3SA, updated with tests and CpuPowerModels --- .../src/main/python/util/config.py | 186 +++++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 opendc-experiments/opendc-experiments-m3sa/src/main/python/util/config.py (limited to 'opendc-experiments/opendc-experiments-m3sa/src/main/python/util/config.py') diff --git a/opendc-experiments/opendc-experiments-m3sa/src/main/python/util/config.py b/opendc-experiments/opendc-experiments-m3sa/src/main/python/util/config.py new file mode 100644 index 00000000..e0d9827b --- /dev/null +++ b/opendc-experiments/opendc-experiments-m3sa/src/main/python/util/config.py @@ -0,0 +1,186 @@ +from json import JSONDecodeError, load +from warnings import warn +from numpy import mean, median +from typing import Callable +from enum import Enum +from sys import stderr +import os + +FUNCTIONS = { + "mean": mean, + "median": median, +} + + +class PlotType(Enum): + TIME_SERIES = "time_series" + CUMULATIVE = "cumulative" + CUMULATIVE_TIME_SERIES = "cumulative_time_series" + + def __str__(self) -> str: + return self.value + + +def get_plot_type(plot_type: str) -> PlotType: + """ + Returns the PlotType enum value for the given string + Args: + plot_type: the string representation of the plot type + Returns: + the PlotType enum value + """ + return next((pt for pt in PlotType if pt.value == plot_type), PlotType.TIME_SERIES) + + +class PlotAxis: + """ + This class represents an axis of a plot. It contains the label, value range, and number of ticks for the axis. + Attributes: + label (str): the label of the axis + value_range (tuple[float, float]): the range of values for the axis + ticks (int): the number of ticks on the axis + """ + + def __init__(self, label: str, value_range: tuple[float, float] | None, ticks: int | None): + self.label = label + self.value_range = value_range + self.ticks = ticks + + def has_range(self) -> bool: + """ + Checks if the axis has a value range + Returns: + True if the axis has a value range, False otherwise + """ + return self.value_range is not None + + def has_ticks(self) -> bool: + """ + Checks if the axis has a number of ticks + Returns: + True if the axis has a number of ticks, False otherwise + """ + return self.ticks is not None + + +class SimulationConfig: + """ + This class represents the configuration of a simulation. + It contains all the necessary parameters to run a simulation using multiple models. + + Attributes: + is_multimodel (bool): whether the simulation is multimodel + is_metamodel (bool): whether the simulation is a metamodel + metric (str): the metric to be used + window_function (function): the window function to be used + meta_function (function): the meta function to be used + window_size (int): the window size + samples_per_minute (int): the number of samples per minute + current_unit (str): the current unit + unit_scaling_magnitude (int): the unit scaling magnitude + plot_type (str): the plot type + plot_title (str): the plot title + x_axis (PlotAxis): the x-axis + y_axis (PlotAxis): the y-axis + seed (int): the seed + fig_size (tuple[int, int]): the figure size + """ + + def __init__(self, input_json: dict[str, any], output_path: str, simulation_path: str): + """ + Initializes the SimulationConfig object with the given input JSON + Args: + input_json: the input JSON object + Raises: + ValueError: if the input JSON is missing required + fields or has invalid values for certain fields + """ + + if "metric" not in input_json: + raise ValueError("Required field 'metric' is missing.") + if "meta_function" not in input_json and input_json["metamodel"]: + raise ValueError( + "Required field 'meta_function' is missing. Please select between 'mean' and 'median'. " + "Alternatively, disable metamodel in the config file." + ) + if input_json["meta_function"] not in FUNCTIONS: + raise ValueError( + "Invalid value for meta_function. Please select between 'mean' and 'median'." + ) + if "multimodel" not in input_json and input_json["metamodel"]: + warn("Warning: Missing 'multimodel' field. Defaulting to 'True'.") + + self.output_path: str = output_path + self.simulation_path: str = simulation_path + self.is_multimodel: bool = input_json.get("multimodel", True) + self.is_metamodel: bool = input_json.get("metamodel", False) + self.metric: str = input_json["metric"] + self.window_function: Callable[[any], float] = FUNCTIONS[input_json.get("window_function", "mean")] + self.meta_function: Callable[[any], float] = FUNCTIONS[input_json.get("meta_function", "mean")] + self.window_size: int = input_json.get("window_size", 1) + self.samples_per_minute: int = input_json.get("samples_per_minute", 0) + self.current_unit: str = input_json.get("current_unit", "") + self.unit_scaling_magnitude: int = input_json.get("unit_scaling_magnitude", 1) + self.plot_type: PlotType = next( + (pt for pt in PlotType if pt.value == input_json.get("plot_type", "time_series")), PlotType.TIME_SERIES) + self.plot_title: str = input_json.get("plot_title", "") + self.x_axis: PlotAxis = PlotAxis( + input_json.get("x_label", ""), + parse_range(input_json, "x"), + input_json.get("x_ticks_count", None) + ) + self.y_axis: PlotAxis = PlotAxis( + input_json.get("y_label", ""), + parse_range(input_json, "y"), + input_json.get("y_ticks_count", None) + ) + self.seed: int = input_json.get("seed", 0) + self.fig_size: tuple[int, int] = input_json.get("figsize", (20, 10)) + self.plot_colors: list[str] = input_json.get("plot_colors", []) + self.figure_export_name: str | None = input_json.get("figure_export_name", None) + + +def parse_range(user_input: dict[str, any], key: str) -> tuple[float, float] | None: + """ + Parses a range from the user input + Args: + user_input: the user input dictionary + key: the key of the range + + Returns: + a tuple containing the minimum and maximum values of the range + """ + + if f"{key}_min" not in user_input or f"{key}_max" not in user_input: + return None + + return user_input[f"{key}_min"], user_input[f"{key}_max"] + + +def parse_configuration(config_path: str, output_path: str, simulation_path: str) -> SimulationConfig: + """ + Reads the input JSON file and returns a SimulationConfig object + Args: + config_path: the path to the input JSON file + output_path: the path to the output folder + simulation_path: the path to the simulation folder + + Returns: + a SimulationConfig object + """ + + try: + with (open(config_path, 'r') as json): + input_json: dict[str, any] = load(json) + except JSONDecodeError: + stderr.write(f"Error decoding JSON in file: {config_path}") + exit(1) + except IOError: + stderr.write(f"Error reading file: {config_path}") + exit(1) + + try: + return SimulationConfig(input_json, output_path, simulation_path) + except ValueError as err: + print(f"Error parsing input JSON: {err}") + exit(1) -- cgit v1.2.3