Coverage for aixcalibuha/data_types.py: 69%
283 statements
« prev ^ index » next coverage.py v7.4.4, created at 2026-04-20 14:06 +0000
1"""
2Module containing data types to enable an automatic usage of
3different other modules in the Python package.
4"""
5import warnings
6import logging
7from typing import Union, Callable, List
8from copy import deepcopy
9import pandas as pd
10import numpy as np
11from ebcpy import TimeSeriesData
12from ebcpy.utils.statistics_analyzer import StatisticsAnalyzer
13from ebcpy.preprocessing import convert_datetime_index_to_float_index
15# pylint: disable=I1101
17logger = logging.getLogger(__name__)
20class Goals:
21 """
22 Class for one or multiple goals. Used to evaluate the
23 difference between current simulation and measured data
25 :param (ebcpy.data_types.TimeSeriesData, pd.DataFrame) meas_target_data:
26 The dataset of the measurement. It acts as a point of reference
27 for the simulation output. If the dimensions of the given DataFrame and later
28 added simulation-data are not equal, an error is raised.
29 Has to hold all variables listed under the MEASUREMENT_NAME variable in the
30 variable_names dict.
31 :param dict variable_names:
32 A dictionary to construct the goals-DataFrame using pandas MultiIndex-Functionality.
33 The dict has to follow the structure.
34 ``variable_names = {VARIABLE_NAME: [MEASUREMENT_NAME, SIMULATION_NAME]}``
36 - VARIABLE_NAME: A string which holds the actual name
37 of the variable you use as a goal.
38 E.g.: ``VARIABLE_NAME="Temperature_Condenser_Outflow"``
39 - MEASUREMENT_NAME: Is either a string or a tuple. Hold the name the variable
40 has inside the given meas_target_data. If you want to specify a tag you have
41 to pass a tuple, like: ``(MEASUREMENT_NAME, TAG_NAME)``. Else just pass a string.
42 E.g.: ``MEASUREMENT_NAME="HydraulicBench[4].T_Out"`` or
43 ``MEASUREMENT_NAME=("HydraulicBench[4].T_Out", "preprocessed")``
44 - SIMULATION_NAME is either a string or a tuple, just like MEASUREMENT_NAME.
45 E.g. (for Modelica): ``SIMULATION_NAME="HeatPump.Condenser.Vol.T"``
47 You may use a tuple instead of a list OR a dict
48 with key "meas" for measurement and key "sim" for simulation. These options may be
49 relevant for your own code readability.
50 E.g. ``variable_names =
51 {VARIABLE_NAME: {"meas":MEASUREMENT_NAME, "sim": SIMULATION_NAME}}``
52 :param str statistical_measure:
53 Measure to calculate the scalar of the objective,
54 One of the supported methods in
55 ebcpy.utils.statistics_analyzer.StatisticsAnalyzer
56 e.g. RMSE, MAE, NRMSE
57 :param list weightings:
58 Values between 0 and 1 to account for multiple Goals to be evaluated.
59 If multiple goals are selected, and weightings is None, each
60 weighting will be equal to 1/(Number of goals).
61 The weighting is scaled so that the sum will equal 1.
62 """
64 # Set default string for measurement reference
65 meas_tag_str = "meas"
66 sim_tag_str = "sim"
68 def __init__(self,
69 meas_target_data: Union[TimeSeriesData, pd.DataFrame],
70 variable_names: dict,
71 statistical_measure: Union[str, List[str]],
72 weightings: list = None):
73 """Initialize class-objects and check correct input."""
75 # Open the meas target data:
76 if not isinstance(meas_target_data, (TimeSeriesData, pd.DataFrame)):
77 raise TypeError(f"Given meas_target_data is of type {type(meas_target_data).__name__} "
78 "but TimeSeriesData is required.")
80 if not isinstance(variable_names, dict):
81 raise TypeError(f"Given variable_names is of type {type(variable_names).__name__} "
82 f"but a dict is required.")
84 # Extract the measurement-information out of the dict.
85 self.variable_names = variable_names
87 # Used to speed up the frequently used set_sim_target_data function
88 self._sim_var_matcher = {}
89 _columns = [] # Used to extract relevant part of df
91 _rename_cols_dict = {}
92 for var_name, meas_sim_info in self.variable_names.items():
93 # First extract the information about the measurement out of the dict
94 if isinstance(meas_sim_info, dict):
95 meas_info = meas_sim_info[self.meas_tag_str]
96 self._sim_var_matcher[var_name] = meas_sim_info[self.sim_tag_str]
97 elif isinstance(meas_sim_info, (list, tuple)):
98 meas_info = meas_sim_info[0]
99 self._sim_var_matcher[var_name] = meas_sim_info[1]
100 else:
101 raise TypeError(f"Variable {var_name} of variable_names has a value"
102 "neither being a dict, list or tuple.")
103 # Now get the info to extract the values out of the given tsd
104 # Convert string with into a list of tuples containing the relevant tag.
105 # If mulitple tags exist, and the default tag (self.meas_tag_str)
106 # is not present, an error is raised.
107 if isinstance(meas_info, str):
108 if isinstance(meas_target_data[meas_info], pd.Series):
109 raise TypeError("Given meas_target_data contains columns without a tag."
110 "Please only pass MultiIndex-DataFrame objects.")
111 tags = meas_target_data[meas_info].columns
112 _rename_cols_dict[meas_info] = var_name
113 if len(tags) != 1 and self.meas_tag_str not in tags:
114 raise TypeError("Not able to automatically select variables and tags. "
115 f"Variable {meas_info} has mutliple tags, none of which "
116 f"is specified as {self.meas_tag_str}.")
117 if self.meas_tag_str in tags:
118 _columns.append((meas_info, self.meas_tag_str))
119 else:
120 _columns.append((meas_info, tags[0]))
121 elif isinstance(meas_info, tuple):
122 _rename_cols_dict[meas_info[0]] = var_name
123 _columns.append(meas_info)
124 else:
125 raise TypeError(f"Measurement Info on variable {var_name} is "
126 "neither of type string or tuple.")
128 # Take the subset of the given tsd based on var_names and tags.
129 self._tsd = meas_target_data[_columns].copy()
131 # Rename all variables to the given var_name (key of self.variable_names)
132 self._tsd = self._tsd.rename(columns=_rename_cols_dict, level=0)
134 # Rename all tags to the default measurement name for consistency.
135 tags = dict(zip(self._tsd.columns.levels[1],
136 [self.meas_tag_str for _ in range(len(_columns))]))
137 self._tsd = self._tsd.rename(columns=tags, level=1)
139 # Save the tsd to a tsd_ref object
140 # Used to never lose the original dataframe.
141 # _tsd may be altered by relevant intervals, this object never!
142 self._tsd_ref = self._tsd.copy()
144 # Set the statistical analyzer:
145 self.statistical_measure = statistical_measure
147 # Set the weightings, if not specified.
148 self._num_goals = len(_columns)
149 if weightings is None:
150 self.weightings = np.array([1 / self._num_goals
151 for i in range(self._num_goals)])
152 else:
153 if not isinstance(weightings, (list, np.ndarray)):
154 raise TypeError(f"weightings is of type {type(weightings).__name__} "
155 f"but should be of type list.")
156 if len(weightings) != self._num_goals:
157 raise IndexError(f"The given number of weightings ({len(weightings)}) "
158 f"does not match the number of "
159 f"goals ({self._num_goals})")
160 self.weightings = np.array(weightings) / sum(weightings)
162 def __str__(self):
163 """Overwrite string method to present the Goals-Object more
164 nicely."""
165 return str(self._tsd)
167 @property
168 def statistical_measure(self):
169 """The statistical measure of this Goal instance"""
170 return self._stat_meas
172 @statistical_measure.setter
173 def statistical_measure(self,
174 statistical_measure: Union[str, Callable, List[Union[str, Callable]]]):
175 """
176 Set the new statistical measure. The value must be
177 supported by the method argument in the
178 ``StatisticsAnalyzer`` class of ``ebcpy``.
179 """
180 def _get_stat_meas(statistical_measure):
181 if callable(statistical_measure):
182 return statistical_measure.__name__
183 return statistical_measure
185 self._stat_meas = None
186 if not isinstance(statistical_measure, list):
187 statistical_measure = [statistical_measure] * len(self.variable_names)
188 self._stat_meas = _get_stat_meas(statistical_measure[0])
190 if len(statistical_measure) != len(self.variable_names):
191 raise ValueError("The number of statistical measures does not match the number of goals.")
193 if self._stat_meas is None:
194 self._stat_meas = '_'.join([_get_stat_meas(i) for i in statistical_measure])
196 self._stat_analyzer = {}
197 for n, goal_name in enumerate(self.variable_names.keys()):
198 self._stat_analyzer[goal_name] = StatisticsAnalyzer(method=statistical_measure[n])
201 def eval_difference(self, verbose=False, penaltyfactor=1):
202 """
203 Evaluate the difference of the measurement and simulated data based on the
204 chosen statistical_measure.
206 :param boolean verbose:
207 If True, a dict with difference-values of for all goals and the
208 corresponding weightings is returned together with the total difference.
209 This can be useful to better understand which goals is performing
210 well in an optimization and which goals needs further is not performing well.
211 :param float penaltyfactor:
212 Muliplty result with this factor to account for
213 penatlies of some sort.
214 :return: float total_difference
215 weighted ouput for all goals.
216 """
217 total_difference = 0
218 _verbose_calculation = {}
220 for i, goal_name in enumerate(self.variable_names.keys()):
221 if self._tsd.isnull().values.any():
222 raise ValueError("There are not valid values in the "
223 "simulated target data. Probably the time "
224 "interval of measured and simulated data "
225 "are not equal. \nPlease check the frequencies "
226 "in the toml file (output_interval & frequency).")
227 _diff = abs(self._stat_analyzer[goal_name].calc(
228 meas=self._tsd[(goal_name, self.meas_tag_str)],
229 sim=self._tsd[(goal_name, self.sim_tag_str)]
230 ))
231 # Apply penalty function
232 _diff = _diff * penaltyfactor
233 _verbose_calculation[goal_name] = (self.weightings[i], _diff)
234 total_difference += self.weightings[i] * _diff
235 if verbose:
236 return total_difference, _verbose_calculation
237 return total_difference
239 def set_sim_target_data(self, sim_target_data):
240 """Alter the object with new simulation data
241 self._sim_target_data based on the given dataframe
242 sim_target_data.
244 :param TimeSeriesData sim_target_data:
245 Object with simulation target data. This data should be
246 the output of a simulation, hence "sim"-target-data.
247 """
248 # Start with the base
249 self._tsd = self._tsd_ref.copy()
250 # Check index type
251 if not isinstance(sim_target_data.index, type(self._tsd.index)):
252 raise IndexError(
253 f"Given sim_target_data is using {type(sim_target_data.index).__name__}"
254 f" as an index, but the reference results (measured-data) was declared"
255 f" using the {type(self._tsd_ref.index).__name__}. Convert your"
256 f" measured-data index to solve this error."
257 )
258 # Three critical cases may occur:
259 # 1. sim_target_data is bigger (in len) than _tsd
260 # --> Only the first part is accepted
261 # 2. sim_target_data is smaller than _tsd
262 # --> Missing values become NaN, which is fine. If no other function eliminates
263 # the NaNs, an error is raised when doing eval_difference().
264 # 3. The index differs:
265 # --> All new values are NaN. However, this should raise an error, as an error
266 # in eval_difference would not lead back to this function.
267 # Check if index matches in relevant intersection:
268 sta = max(self._tsd.index[0], sim_target_data.index[0])
269 sto = min(self._tsd.index[-1], sim_target_data.index[-1])
270 if len(self._tsd.loc[sta:sto].index) != len(sim_target_data.loc[sta:sto].index):
271 raise ValueError(f"Given indexes have different lengths "
272 f"({len(self._tsd.loc[sta:sto].index)} vs "
273 f"{len(sim_target_data.loc[sta:sto].index)}). "
274 f"Can't compare them. ")
275 mask = self._tsd.loc[sta:sto].index != sim_target_data.loc[sta:sto].index
276 if np.any(mask):
277 diff = self._tsd.loc[sta:sto].index - sim_target_data.loc[sta:sto].index
278 raise IndexError(f"Measured and simulated data differ on {np.count_nonzero(mask)}"
279 f" index points. Affected index part: {diff[mask]}. "
280 f"This will lead to errors in evaluation, "
281 f"hence we raise the error already here. "
282 f"Check output_interval, equidistant_output and "
283 f"frequency of measured data to find the reason for "
284 f"this error. The have to match.")
286 # Resize simulation data to match to meas data
287 for goal_name in self.variable_names.keys():
288 _tsd_sim = sim_target_data.loc[sta:sto, self._sim_var_matcher[goal_name]]
289 if len(_tsd_sim.columns) > 1:
290 raise ValueError("Given sim_target_data contains multiple tags for variable "
291 f"{self._sim_var_matcher[goal_name]}. "
292 "Can't select one automatically.")
293 self._tsd.loc[sta:sto, (goal_name, self.sim_tag_str)] = _tsd_sim.values
294 # Sort the index for better visualisation
295 self._tsd = self._tsd.sort_index(axis=1)
297 def set_relevant_time_intervals(self, intervals):
298 """
299 For many calibration-uses cases, different time-intervals of the measured
300 and simulated data are relevant. Set the interval to be used with this function.
301 This will change both measured and simulated data. Therefore, the eval_difference
302 function can be called at every moment.
304 :param list intervals:
305 List with time-intervals. Each list element has to be a tuple
306 with the first element being the start_time as float or int and
307 the second item being the end_time of the interval as float or int.
308 E.g:
309 [(0, 100), [150, 200), (500, 600)]
310 """
311 _df_ref = self._tsd.copy()
312 # Create initial False mask
313 _mask = np.full(_df_ref.index.shape, False)
314 # Dynamically make mask for multiple possible time-intervals
315 for _start_time, _end_time in intervals:
316 _mask = _mask | ((_df_ref.index >= _start_time) & (_df_ref.index <= _end_time))
317 self._tsd = _df_ref.loc[_mask]
319 def get_goals_list(self):
320 """Get the internal list containing all goals."""
321 return list(self.variable_names.keys())
323 def get_goals_data(self):
324 """Get the current time-series-data object."""
325 return self._tsd.copy()
327 def get_sim_var_names(self):
328 """Get the names of the simulation variables.
330 :returns list sim_var_names:
331 Names of the simulation variables as a list
332 """
333 return list(self._sim_var_matcher.values())
335 def get_meas_frequency(self):
336 """
337 Get the frequency of the measurement data.
339 :returns:
340 float: Mean frequency of the index
341 """
342 mean, std = self._tsd_ref.frequency
343 if std >= 1e-8:
344 logger.critical("The index of your measurement data is not "
345 "equally sampled. The standard deviation is %s."
346 "The may lead to errors when mapping measurements to simulation "
347 "results.", mean.std())
348 return mean
351class TunerParas:
352 """
353 Class for tuner parameters.
354 Tuner parameters are parameters of a model which are
355 constant during simulation but are varied during calibration
356 or other analysis.
358 :param list names:
359 List of names of the tuner parameters
360 :param float,int initial_values:
361 Initial values for optimization.
362 Even though some optimization methods don't require an
363 initial guess, specifying a initial guess based on
364 expected values or experience is helpful to better
365 check the results of the calibration
366 :param list,tuple bounds:
367 Tuple or list of float or ints for lower and upper bound to the tuner parameter.
368 The bounds object is optional, however highly recommend
369 for calibration or optimization in general. As soon as you
370 tune parameters with different units, such as Capacity and
371 heat conductivity, the solver will fail to find good solutions.
374 Example:
376 >>> tuner_paras = TunerParas(names=["C", "m_flow_2", "heatConv_a"],
377 >>> initial_values=[5000, 0.02, 200],
378 >>> bounds=[(4000, 6000), (0.01, 0.1), (10, 300)])
379 >>> print(tuner_paras)
380 initial_value min max scale
381 names
382 C 5000.00 4000.00 6000.0 2000.00
383 m_flow_2 0.02 0.01 0.1 0.09
384 heatConv_a 200.00 10.00 300.0 290.00
385 """
387 def __init__(self, names, initial_values, bounds=None):
388 """Initialize class-objects and check correct input."""
389 # Check if the given input-parameters are of correct format. If not, raise an error.
390 for name in names:
391 if not isinstance(name, str):
392 raise TypeError(f"Given name is of type {type(name).__name__} "
393 "and not of type str.")
394 # Check if all names are unique:
395 if len(names) != len(set(names)):
396 raise ValueError("Given names contain duplicates. "
397 "This will yield errors in later stages"
398 "such as calibration of sensitivity analysis.")
399 try:
400 # Calculate the sum, as this will fail if the elements are not float or int.
401 sum(initial_values)
402 except TypeError as err:
403 raise TypeError("initial_values contains other "
404 "instances than float or int.") from err
405 if len(names) != len(initial_values):
406 raise ValueError(f"shape mismatch: names has length {len(names)}"
407 f" and initial_values {len(initial_values)}.")
408 self._bounds = bounds
409 if bounds is None:
410 _bound_min = -np.inf
411 _bound_max = np.inf
412 else:
413 if len(bounds) != len(names):
414 raise ValueError(f"shape mismatch: bounds has length {len(bounds)} "
415 f"and names {len(names)}.")
416 _bound_min, _bound_max = [], []
417 for bound in bounds:
418 _bound_min.append(bound[0])
419 _bound_max.append(bound[1])
421 self._df = pd.DataFrame({"names": names,
422 "initial_value": initial_values,
423 "min": _bound_min,
424 "max": _bound_max})
425 self._df = self._df.set_index("names")
426 self._set_scale()
428 def __str__(self):
429 """Overwrite string method to present the TunerParas-Object more
430 nicely."""
431 return str(self._df)
433 def scale(self, descaled):
434 """
435 Scales the given value to the bounds of the tuner parameter between 0 and 1
437 :param np.array,list descaled:
438 Value to be scaled
439 :return: np.array scaled:
440 Scaled value between 0 and 1
441 """
442 # If no bounds are given, scaling is not possible--> descaled = scaled
443 if self._bounds is None:
444 return descaled
445 _scaled = (descaled - self._df["min"]) / self._df["scale"]
446 if not all((_scaled >= 0) & (_scaled <= 1)):
447 warnings.warn("Given descaled values are outside "
448 "of bounds. Automatically limiting "
449 "the values with respect to the bounds.")
450 return np.clip(_scaled, a_min=0, a_max=1)
452 def descale(self, scaled):
453 """
454 Converts the given scaled value to an descaled one.
456 :param np.array,list scaled:
457 Scaled input value between 0 and 1
458 :return: np.array descaled:
459 descaled value based on bounds.
460 """
461 # If no bounds are given, scaling is not possible--> descaled = scaled
462 if not self._bounds:
463 return scaled
464 _scaled = np.array(scaled)
465 if not all((_scaled >= 0 - 1e4) & (_scaled <= 1 + 1e4)):
466 warnings.warn("Given scaled values are outside of bounds. "
467 "Automatically limiting the values with "
468 "respect to the bounds.")
469 _scaled = np.clip(_scaled, a_min=0, a_max=1)
470 return _scaled * self._df["scale"] + self._df["min"]
472 @property
473 def bounds(self):
474 """Get property bounds"""
475 return self._bounds
477 def get_names(self):
478 """Return the names of the tuner parameters"""
479 return list(self._df.index)
481 def get_initial_values(self):
482 """Return the initial values of the tuner parameters"""
483 return self._df["initial_value"].values
485 def get_bounds(self):
486 """Return the bound-values of the tuner parameters"""
487 return self._df["min"].values, self._df["max"].values
489 def get_value(self, name, col):
490 """Function to get a value of a specific tuner parameter"""
491 return self._df.loc[name, col]
493 def set_value(self, name, col, value):
494 """Function to set a value of a specific tuner parameter"""
495 if not isinstance(value, (float, int)):
496 raise ValueError(f"Given value is of type {type(value).__name__} "
497 "but float or int is required")
498 if col not in ["max", "min", "initial_value"]:
499 raise KeyError("Can only alter max, min and initial_value")
500 self._df.loc[name, col] = value
501 self._set_scale()
503 def remove_names(self, names):
504 """
505 Remove gives list of names from the Tuner-parameters
507 :param list names:
508 List with names inside of the TunerParas-dataframe
509 """
510 self._df = self._df.drop(names)
512 def _set_scale(self):
513 self._df["scale"] = self._df["max"] - self._df["min"]
514 if not self._df[self._df["scale"] <= 0].empty:
515 raise ValueError(
516 "The given lower bounds are greater equal "
517 "than the upper bounds, resulting in a "
518 f"negative scale: \n{str(self._df['scale'])}"
519 )
522class CalibrationClass:
523 """
524 Class used for calibration of time-series data.
526 :param str name:
527 Name of the class, e.g. 'device on'
528 :param float,int start_time:
529 Time at which the class starts
530 :param float,int stop_time:
531 Time at which the class ends
532 :param Goals goals:
533 Goals parameters which are relevant in this class.
534 As this class may be used in the classifier, a Goals-Class
535 may not be available at all times and can be added later.
536 :param TunerParas tuner_paras:
537 As this class may be used in the classifier, a TunerParas-Class
538 may not be available at all times and can be added later.
539 :param list relevant_intervals:
540 List with time-intervals relevant for the calibration.
541 Each list element has to be a tuple with the first element being
542 the start-time as float/int and the second item being the end-time
543 of the interval as float/int.
544 E.g:
545 For a class with start_time=0 and stop_time=1000, given following intervals
546 [(0, 100), [150, 200), (500, 600)]
547 will only evaluate the data between 0-100, 150-200 and 500-600.
548 The given intervals may overlap. Furthermore the intervals do not need
549 to be in an ascending order or be limited to
550 the start_time and end_time parameters.
551 :keyword (pd.DataFrame, ebcpy.data_types.TimeSeriesData) inputs:
552 TimeSeriesData or DataFrame that holds
553 input data for the simulation to run.
554 The time-index should be float index and match the overall
555 ranges set by start- and stop-time.
556 :keyword dict input_kwargs:
557 If inputs are provided, additional input keyword-args passed to the
558 simulation API can be specified.
559 Using FMUs, you don't need to specify anything.
560 Using DymolaAPI, you have to specify 'table_name' and 'file_name'
561 """
563 def __init__(self, name, start_time, stop_time, goals=None,
564 tuner_paras=None, relevant_intervals=None, **kwargs):
565 """Initialize class-objects and check correct input."""
566 self.name = name
567 self._start_time = start_time
568 self.stop_time = stop_time
569 self._goals = None
570 self._tuner_paras = None
571 if goals is not None:
572 self.goals = goals
573 if tuner_paras is not None:
574 self.tuner_paras = tuner_paras
575 if relevant_intervals is not None:
576 self.relevant_intervals = relevant_intervals
577 else:
578 # Then all is relevant
579 self.relevant_intervals = [(start_time, stop_time)]
580 self._inputs = None
581 inputs = kwargs.get('inputs', None)
582 if inputs is not None:
583 self.inputs = inputs # Trigger the property setter
584 self.input_kwargs = kwargs.get('input_kwargs', {})
586 @property
587 def name(self):
588 """Get name of calibration class"""
589 return self._name
591 @name.setter
592 def name(self, name: str):
593 """Set name of calibration class"""
594 if not isinstance(name, str):
595 raise TypeError(f"Name of CalibrationClass is {type(name)} "
596 f"but has to be of type str")
597 self._name = name
599 @property
600 def start_time(self) -> Union[float, int]:
601 """Get start time of calibration class"""
602 return self._start_time
604 @start_time.setter
605 def start_time(self, start_time: Union[float, int]):
606 """Set start time of calibration class"""
607 if not start_time <= self.stop_time:
608 raise ValueError("The given start-time is "
609 "higher than the stop-time.")
610 self._start_time = start_time
612 @property
613 def stop_time(self) -> Union[float, int]:
614 """Get stop time of calibration class"""
615 return self._stop_time
617 @stop_time.setter
618 def stop_time(self, stop_time: Union[float, int]):
619 """Set stop time of calibration class"""
620 if not self.start_time <= stop_time:
621 raise ValueError("The given stop-time is "
622 "lower than the start-time.")
623 self._stop_time = stop_time
625 @property
626 def tuner_paras(self) -> TunerParas:
627 """Get the tuner parameters of the calibration-class"""
628 return self._tuner_paras
630 @tuner_paras.setter
631 def tuner_paras(self, tuner_paras):
632 """
633 Set the tuner parameters for the calibration-class.
635 :param tuner_paras: TunerParas
636 """
637 if not isinstance(tuner_paras, TunerParas):
638 raise TypeError(f"Given tuner_paras is of type {type(tuner_paras).__name__} "
639 "but should be type TunerParas")
640 self._tuner_paras = deepcopy(tuner_paras)
642 @property
643 def goals(self) -> Goals:
644 """Get current goals instance"""
645 return self._goals
647 @goals.setter
648 def goals(self, goals: Goals):
649 """
650 Set the goals object for the calibration-class.
652 :param Goals goals:
653 Goals-data-type
654 """
655 if not isinstance(goals, Goals):
656 raise TypeError(f"Given goals parameter is of type {type(goals).__name__} "
657 "but should be type Goals")
658 self._goals = deepcopy(goals)
660 @property
661 def relevant_intervals(self) -> list:
662 """Get current relevant_intervals"""
663 return self._relevant_intervals
665 @relevant_intervals.setter
666 def relevant_intervals(self, relevant_intervals: list):
667 """Set current relevant_intervals"""
668 self._relevant_intervals = relevant_intervals
670 @property
671 def inputs(self) -> Union[TimeSeriesData, pd.DataFrame]:
672 """Get the inputs for this calibration class"""
673 return self._inputs
675 @inputs.setter
676 def inputs(self, inputs: Union[TimeSeriesData, pd.DataFrame]):
677 """Set the inputs for this calibration class"""
678 # Check correct index:
679 if not isinstance(inputs, (TimeSeriesData, pd.DataFrame)):
680 raise TypeError(f"Inputs need to be either TimeSeriesData "
681 f"or pd.DataFrame, but you passed {type(inputs)}")
682 if isinstance(inputs.index, pd.DatetimeIndex):
683 inputs = convert_datetime_index_to_float_index(inputs)
685 self._inputs = inputs
688def merge_calibration_classes(calibration_classes):
689 """
690 Given a list of multiple calibration-classes, this function merges given
691 objects by the "name" attribute. Relevant intervals are set, in order
692 to maintain the start and stop-time info.
694 :param list calibration_classes:
695 List containing multiple CalibrationClass-Objects
696 :return: list cal_classes_merged:
697 A list containing one CalibrationClass-Object for each different
698 "name" of class.
700 Example:
701 >>> cal_classes = [CalibrationClass("on", 0, 100),
702 >>> CalibrationClass("off", 100, 200),
703 >>> CalibrationClass("on", 200, 300)]
704 >>> merged_classes = merge_calibration_classes(cal_classes)
705 Is equal to:
706 >>> merged_classes = [CalibrationClass("on", 0, 300,
707 >>> relevant_intervals=[(0,100), (200,300)]),
708 >>> CalibrationClass("off", 100, 200)]
710 """
711 # Use a dict for easy name-access
712 temp_merged = {}
713 for cal_class in calibration_classes:
714 _name = cal_class.name
715 # First create dictionary with all calibration classes
716 if _name in temp_merged:
717 temp_merged[_name]["intervals"] += cal_class.relevant_intervals
718 else:
719 temp_merged[_name] = {"goals": cal_class.goals,
720 "tuner_paras": cal_class.tuner_paras,
721 "intervals": deepcopy(cal_class.relevant_intervals),
722 "inputs": deepcopy(cal_class.inputs),
723 "input_kwargs": deepcopy(cal_class.input_kwargs)
724 }
726 # Convert dict to actual calibration-classes
727 cal_classes_merged = []
728 for _name, values in temp_merged.items():
729 # Flatten the list of tuples and get the start- and stop-values
730 start_time = min(sum(values["intervals"], ()))
731 stop_time = max(sum(values["intervals"], ()))
732 cal_classes_merged.append(CalibrationClass(
733 _name, start_time, stop_time,
734 goals=values["goals"],
735 tuner_paras=values["tuner_paras"],
736 relevant_intervals=values["intervals"],
737 inputs=values["inputs"],
738 input_kwargs=values["input_kwargs"])
739 )
741 return cal_classes_merged