1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
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)
|
