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