1 files changed, 172 insertions, 0 deletions
diff --git a/opendc-experiments/opendc-experiments-m3sa/src/main/python/models/meta_model.py b/opendc-experiments/opendc-experiments-m3sa/src/main/python/models/meta_model.py
new file mode 100644
index 00000000..a6d0fded
--- /dev/null
+++ b/opendc-experiments/opendc-experiments-m3sa/src/main/python/models/meta_model.py
@@ -0,0 +1,172 @@
+import os
+import pandas as pd
+from models import Model, MultiModel
+from typing import Callable
+from util import PlotType
+
+
+class MetaModel:
+    """
+    A class that aggregates results from multiple simulation models based on user-defined functions, producing
+    consolidated outputs for analysis.
+
+    Attributes:
+        multi_model (MultiModel): The container of models whose results are aggregated.
+        meta_model (Model): Model instance that stores aggregated results.
+        meta_function (function): Function used to calculate aggregated data.
+        min_raw_model_len (int): Minimum length of raw data arrays across all models.
+        min_processed_model_len (int): Minimum length of processed data arrays across all models.
+        number_of_models (int): Number of models being aggregated.
+        function_map (dict): Mapping of aggregation function names to function implementations.
+    """
+
+    META_MODEL_ID = 'M'
+
+    def __init__(self, multi_model: MultiModel, meta_function: Callable[[any], float] = None):
+        """
+        Initializes the Metamodel with a MultiModel instance and prepares aggregation functions based on configuration.
+
+        :param multi_model: MultiModel instance containing the models to aggregate.
+        :raise ValueError: If metamodel functionality is not enabled in the configuration.
+        """
+        if not multi_model.config.is_metamodel:
+            raise ValueError("Metamodel is not enabled in the config file")
+
+        self.multi_model = multi_model
+        self.meta_model: Model = Model(
+            raw_sim_data=[],
+            identifier=self.META_MODEL_ID,
+        )
+
+        self.meta_function: Callable[
+            [any], float] = self.multi_model.config.meta_function if meta_function is None else meta_function
+
+        self.min_raw_model_len = min([len(model.raw_sim_data) for model in self.multi_model.models])
+        self.min_processed_model_len = min([len(model.processed_sim_data) for model in self.multi_model.models])
+        self.number_of_models = len(self.multi_model.models)
+
+    def output(self) -> None:
+        """
+        Generates outputs by plotting the aggregated results and exporting the metamodel data to a file.
+        :return: None
+        :side effect: Outputs data to files and generates plots.
+        """
+        self.plot()
+        self.output_metamodel()
+
+    def compute(self) -> None:
+        """
+        Computes aggregated data based on the specified plot type from the configuration.
+        :raise ValueError: If an unsupported plot type is specified in the configuration.
+        """
+        match self.multi_model.config.plot_type:
+            case PlotType.TIME_SERIES:
+                self.compute_time_series()
+            case PlotType.CUMULATIVE:
+                self.compute_cumulative()
+            case PlotType.CUMULATIVE_TIME_SERIES:
+                self.compute_cumulative_time_series()
+
+    def plot(self) -> None:
+        """
+        Plots the aggregated data according to the specified plot type from the configuration.
+        :raise ValueError: If an unsupported plot type is specified.
+        """
+
+        match self.multi_model.config.plot_type:
+            case PlotType.TIME_SERIES:
+                self.plot_time_series()
+            case PlotType.CUMULATIVE:
+                self.plot_cumulative()
+            case PlotType.CUMULATIVE_TIME_SERIES:
+                self.plot_cumulative_time_series()
+
+    def compute_time_series(self):
+        """
+        Aggregates time series data across models using the specified aggregation function.
+        :return: None
+        :side effect: Updates the meta_model's processed data with aggregated results.
+        """
+        for i in range(0, self.min_processed_model_len):
+            data_entries = []
+            for model in self.multi_model.models:
+                data_entries.append(model.processed_sim_data[i])
+            self.meta_model.processed_sim_data.append(self.meta_function(data_entries))
+        self.meta_model.raw_sim_data = self.meta_model.processed_sim_data
+
+    def plot_time_series(self):
+        """
+        Generates a time series plot of the aggregated data.
+        :return: None
+        :side effect: Displays a time series plot using the multi_model's plotting capabilities.
+        """
+        self.multi_model.models.append(self.meta_model)
+        self.multi_model.generate_plot()
+
+    def compute_cumulative(self):
+        """
+        Aggregates cumulative data entries across all models.
+        :return: None
+        :side effect: Updates the meta_model's cumulative data with aggregated results.
+        """
+        for i in range(0, self.min_raw_model_len):
+            data_entries = []
+            for model in self.multi_model.models:
+                sim_data = model.raw_sim_data
+                ith_element = sim_data[i]
+                data_entries.append(ith_element)
+            self.meta_model.cumulated += self.meta_function(data_entries)
+
+        self.meta_model.cumulated = round(self.meta_model.cumulated, 2)
+
+    def plot_cumulative(self):
+        """
+        Generates a cumulative plot of the aggregated data.
+        :return: None
+        :side effect: Displays a cumulative plot using the multi_model's plotting capabilities.
+        """
+        self.multi_model.models.append(self.meta_model)
+        self.multi_model.generate_plot()
+
+    def compute_cumulative_time_series(self):
+        """
+        Aggregates cumulative time series data entries across models using the specified aggregation function.
+        :return: None
+        :side effect: Updates the meta_model's processed data with cumulative aggregated results.
+        """
+        for i in range(0, self.min_processed_model_len):
+            data_entries = []
+            for model in self.multi_model.models:
+                data_entries.append(model.processed_sim_data[i])
+            self.meta_model.processed_sim_data.append(self.meta_function(data_entries))
+
+    def plot_cumulative_time_series(self):
+        """
+        Generates a cumulative time series plot of the aggregated data.
+        :return: None
+        :side effect: Displays a cumulative time series plot using the multi_model's plotting capabilities.
+        """
+        self.multi_model.models.append(self.meta_model)
+        self.multi_model.generate_plot()
+
+    def output_metamodel(self):
+        """
+        Exports the processed sim data of the metamodel to a parquet file for further analysis or record keeping.
+        :return: None
+        :side effect: Writes data to a parquet file at the specified directory path.
+        """
+        directory_path = os.path.join(self.multi_model.config.output_path, "raw-output/metamodel/seed=0")
+        try:
+            os.makedirs(directory_path, exist_ok=True)
+        except OSError as e:
+            print(f"Error creating directory: {e}")
+            exit(1)
+
+        current_path = os.path.join(directory_path, f"{self.multi_model.config.metric}.parquet")
+        minimum = min(len(self.multi_model.timestamps), len(self.meta_model.processed_sim_data))
+
+        df = pd.DataFrame({
+            "timestamp": self.multi_model.timestamps[:minimum],
+            self.multi_model.config.metric: self.meta_model.processed_sim_data[:minimum]
+        })
+        df.to_parquet(current_path, index=False)