Coverage for aixcalibuha/data

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

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: 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.")

127

128 # Take the subset of the given tsd based on var_names and tags.

129 self._tsd = meas_target_data[_columns].copy()

130

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)

133

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)

138

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()

143

144 # Set the statistical analyzer:

145 self.statistical_measure = statistical_measure

146

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)

161

162 def __str__(self):

163 """Overwrite string method to present the Goals-Object more

164 nicely."""

165 return str(self._tsd)

166

167 @property

168 def statistical_measure(self):

169 """The statistical measure of this Goal instance"""

170 return self._stat_meas

171

172 @statistical_measure.setter

173 def statistical_measure(self, statistical_measure: Union[str, Callable]):

174 """

175 Set the new statistical measure. The value must be

176 supported by the method argument in the

177 ``StatisticsAnalyzer`` class of ``ebcpy``.

178 """

179 self._stat_analyzer = StatisticsAnalyzer(method=statistical_measure)

180 if callable(statistical_measure):

181 self._stat_meas = statistical_measure.__name__

182 else:

183 self._stat_meas = statistical_measure

184

185 def eval_difference(self, verbose=False, penaltyfactor=1):

186 """

187 Evaluate the difference of the measurement and simulated data based on the

188 chosen statistical_measure.

189

190 :param boolean verbose:

191 If True, a dict with difference-values of for all goals and the

192 corresponding weightings is returned together with the total difference.

193 This can be useful to better understand which goals is performing

194 well in an optimization and which goals needs further is not performing well.

195 :param float penaltyfactor:

196 Muliplty result with this factor to account for

197 penatlies of some sort.

198 :return: float total_difference

199 weighted ouput for all goals.

200 """

201 total_difference = 0

202 _verbose_calculation = {}

203

204 for i, goal_name in enumerate(self.variable_names.keys()):

205 if self._tsd.isnull().values.any():

206 raise ValueError("There are not valid values in the "

207 "simulated target data. Probably the time "

208 "interval of measured and simulated data "

209 "are not equal. \nPlease check the frequencies "

210 "in the toml file (output_interval & frequency).")

211 _diff = self._stat_analyzer.calc(

212 meas=self._tsd[(goal_name, self.meas_tag_str)],

213 sim=self._tsd[(goal_name, self.sim_tag_str)]

214 )

215 # Apply penalty function

216 _diff = _diff * penaltyfactor

217 _verbose_calculation[goal_name] = (self.weightings[i], _diff)

218 total_difference += self.weightings[i] * _diff

219 if verbose:

220 return total_difference, _verbose_calculation

221 return total_difference

222

223 def set_sim_target_data(self, sim_target_data):

224 """Alter the object with new simulation data

225 self._sim_target_data based on the given dataframe

226 sim_target_data.

227

228 :param TimeSeriesData sim_target_data:

229 Object with simulation target data. This data should be

230 the output of a simulation, hence "sim"-target-data.

231 """

232 # Start with the base

233 self._tsd = self._tsd_ref.copy()

234 # Check index type

235 if not isinstance(sim_target_data.index, type(self._tsd.index)):

236 raise IndexError(

237 f"Given sim_target_data is using {type(sim_target_data.index).__name__}"

238 f" as an index, but the reference results (measured-data) was declared"

239 f" using the {type(self._tsd_ref.index).__name__}. Convert your"

240 f" measured-data index to solve this error."

241 )

242 # Three critical cases may occur:

243 # 1. sim_target_data is bigger (in len) than _tsd

244 # --> Only the first part is accepted

245 # 2. sim_target_data is smaller than _tsd

246 # --> Missing values become NaN, which is fine. If no other function eliminates

247 # the NaNs, an error is raised when doing eval_difference().

248 # 3. The index differs:

249 # --> All new values are NaN. However, this should raise an error, as an error

250 # in eval_difference would not lead back to this function.

251 # Check if index matches in relevant intersection:

252 sta = max(self._tsd.index[0], sim_target_data.index[0])

253 sto = min(self._tsd.index[-1], sim_target_data.index[-1])

254 if len(self._tsd.loc[sta:sto].index) != len(sim_target_data.loc[sta:sto].index):

255 raise ValueError(f"Given indexes have different lengths "

256 f"({len(self._tsd.loc[sta:sto].index)} vs "

257 f"{len(sim_target_data.loc[sta:sto].index)}). "

258 f"Can't compare them. ")

259 mask = self._tsd.loc[sta:sto].index != sim_target_data.loc[sta:sto].index

260 if np.any(mask):

261 diff = self._tsd.loc[sta:sto].index - sim_target_data.loc[sta:sto].index

262 raise IndexError(f"Measured and simulated data differ on {np.count_nonzero(mask)}"

263 f" index points. Affected index part: {diff[mask]}. "

264 f"This will lead to errors in evaluation, "

265 f"hence we raise the error already here. "

266 f"Check output_interval, equidistant_output and "

267 f"frequency of measured data to find the reason for "

268 f"this error. The have to match.")

269

270 # Resize simulation data to match to meas data

271 for goal_name in self.variable_names.keys():

272 _tsd_sim = sim_target_data.loc[sta:sto, self._sim_var_matcher[goal_name]]

273 if len(_tsd_sim.columns) > 1:

274 raise ValueError("Given sim_target_data contains multiple tags for variable "

275 f"{self._sim_var_matcher[goal_name]}. "

276 "Can't select one automatically.")

277 self._tsd.loc[sta:sto, (goal_name, self.sim_tag_str)] = _tsd_sim.values

278 # Sort the index for better visualisation

279 self._tsd = self._tsd.sort_index(axis=1)

280

281 def set_relevant_time_intervals(self, intervals):

282 """

283 For many calibration-uses cases, different time-intervals of the measured

284 and simulated data are relevant. Set the interval to be used with this function.

285 This will change both measured and simulated data. Therefore, the eval_difference

286 function can be called at every moment.

287

288 :param list intervals:

289 List with time-intervals. Each list element has to be a tuple

290 with the first element being the start_time as float or int and

291 the second item being the end_time of the interval as float or int.

292 E.g:

293 [(0, 100), [150, 200), (500, 600)]

294 """

295 _df_ref = self._tsd.copy()

296 # Create initial False mask

297 _mask = np.full(_df_ref.index.shape, False)

298 # Dynamically make mask for multiple possible time-intervals

299 for _start_time, _end_time in intervals:

300 _mask = _mask | ((_df_ref.index >= _start_time) & (_df_ref.index <= _end_time))

301 self._tsd = _df_ref.loc[_mask]

302

303 def get_goals_list(self):

304 """Get the internal list containing all goals."""

305 return list(self.variable_names.keys())

306

307 def get_goals_data(self):

308 """Get the current time-series-data object."""

309 return self._tsd.copy()

310

311 def get_sim_var_names(self):

312 """Get the names of the simulation variables.

313

314 :returns list sim_var_names:

315 Names of the simulation variables as a list

316 """

317 return list(self._sim_var_matcher.values())

318

319 def get_meas_frequency(self):

320 """

321 Get the frequency of the measurement data.

322

323 :returns:

324 float: Mean frequency of the index

325 """

326 mean, std = self._tsd_ref.frequency

327 if std >= 1e-8:

328 logger.critical("The index of your measurement data is not "

329 "equally sampled. The standard deviation is %s."

330 "The may lead to errors when mapping measurements to simulation "

331 "results.", mean.std())

332 return mean

333

334

335class TunerParas:

336 """

337 Class for tuner parameters.

338 Tuner parameters are parameters of a model which are

339 constant during simulation but are varied during calibration

340 or other analysis.

341

342 :param list names:

343 List of names of the tuner parameters

344 :param float,int initial_values:

345 Initial values for optimization.

346 Even though some optimization methods don't require an

347 initial guess, specifying a initial guess based on

348 expected values or experience is helpful to better

349 check the results of the calibration

350 :param list,tuple bounds:

351 Tuple or list of float or ints for lower and upper bound to the tuner parameter.

352 The bounds object is optional, however highly recommend

353 for calibration or optimization in general. As soon as you

354 tune parameters with different units, such as Capacity and

355 heat conductivity, the solver will fail to find good solutions.

356

357

358 Example:

359

360 >>> tuner_paras = TunerParas(names=["C", "m_flow_2", "heatConv_a"],

361 >>> initial_values=[5000, 0.02, 200],

362 >>> bounds=[(4000, 6000), (0.01, 0.1), (10, 300)])

363 >>> print(tuner_paras)

364 initial_value min max scale

365 names

366 C 5000.00 4000.00 6000.0 2000.00

367 m_flow_2 0.02 0.01 0.1 0.09

368 heatConv_a 200.00 10.00 300.0 290.00

369 """

370

371 def __init__(self, names, initial_values, bounds=None):

372 """Initialize class-objects and check correct input."""

373 # Check if the given input-parameters are of correct format. If not, raise an error.

374 for name in names:

375 if not isinstance(name, str):

376 raise TypeError(f"Given name is of type {type(name).__name__} "

377 "and not of type str.")

378 # Check if all names are unique:

379 if len(names) != len(set(names)):

380 raise ValueError("Given names contain duplicates. "

381 "This will yield errors in later stages"

382 "such as calibration of sensitivity analysis.")

383 try:

384 # Calculate the sum, as this will fail if the elements are not float or int.

385 sum(initial_values)

386 except TypeError as err:

387 raise TypeError("initial_values contains other "

388 "instances than float or int.") from err

389 if len(names) != len(initial_values):

390 raise ValueError(f"shape mismatch: names has length {len(names)}"

391 f" and initial_values {len(initial_values)}.")

392 self._bounds = bounds

393 if bounds is None:

394 _bound_min = -np.inf

395 _bound_max = np.inf

396 else:

397 if len(bounds) != len(names):

398 raise ValueError(f"shape mismatch: bounds has length {len(bounds)} "

399 f"and names {len(names)}.")

400 _bound_min, _bound_max = [], []

401 for bound in bounds:

402 _bound_min.append(bound[0])

403 _bound_max.append(bound[1])

404

405 self._df = pd.DataFrame({"names": names,

406 "initial_value": initial_values,

407 "min": _bound_min,

408 "max": _bound_max})

409 self._df = self._df.set_index("names")

410 self._set_scale()

411

412 def __str__(self):

413 """Overwrite string method to present the TunerParas-Object more

414 nicely."""

415 return str(self._df)

416

417 def scale(self, descaled):

418 """

419 Scales the given value to the bounds of the tuner parameter between 0 and 1

420

421 :param np.array,list descaled:

422 Value to be scaled

423 :return: np.array scaled:

424 Scaled value between 0 and 1

425 """

426 # If no bounds are given, scaling is not possible--> descaled = scaled

427 if self._bounds is None:

428 return descaled

429 _scaled = (descaled - self._df["min"]) / self._df["scale"]

430 if not all((_scaled >= 0) & (_scaled <= 1)):

431 warnings.warn("Given descaled values are outside "

432 "of bounds. Automatically limiting "

433 "the values with respect to the bounds.")

434 return np.clip(_scaled, a_min=0, a_max=1)

435

436 def descale(self, scaled):

437 """

438 Converts the given scaled value to an descaled one.

439

440 :param np.array,list scaled:

441 Scaled input value between 0 and 1

442 :return: np.array descaled:

443 descaled value based on bounds.

444 """

445 # If no bounds are given, scaling is not possible--> descaled = scaled

446 if not self._bounds:

447 return scaled

448 _scaled = np.array(scaled)

449 if not all((_scaled >= 0 - 1e4) & (_scaled <= 1 + 1e4)):

450 warnings.warn("Given scaled values are outside of bounds. "

451 "Automatically limiting the values with "

452 "respect to the bounds.")

453 _scaled = np.clip(_scaled, a_min=0, a_max=1)

454 return _scaled * self._df["scale"] + self._df["min"]

455

456 @property

457 def bounds(self):

458 """Get property bounds"""

459 return self._bounds

460

461 def get_names(self):

462 """Return the names of the tuner parameters"""

463 return list(self._df.index)

464

465 def get_initial_values(self):

466 """Return the initial values of the tuner parameters"""

467 return self._df["initial_value"].values

468

469 def get_bounds(self):

470 """Return the bound-values of the tuner parameters"""

471 return self._df["min"].values, self._df["max"].values

472

473 def get_value(self, name, col):

474 """Function to get a value of a specific tuner parameter"""

475 return self._df.loc[name, col]

476

477 def set_value(self, name, col, value):

478 """Function to set a value of a specific tuner parameter"""

479 if not isinstance(value, (float, int)):

480 raise ValueError(f"Given value is of type {type(value).__name__} "

481 "but float or int is required")

482 if col not in ["max", "min", "initial_value"]:

483 raise KeyError("Can only alter max, min and initial_value")

484 self._df.loc[name, col] = value

485 self._set_scale()

486

487 def remove_names(self, names):

488 """

489 Remove gives list of names from the Tuner-parameters

490

491 :param list names:

492 List with names inside of the TunerParas-dataframe

493 """

494 self._df = self._df.drop(names)

495

496 def _set_scale(self):

497 self._df["scale"] = self._df["max"] - self._df["min"]

498 if not self._df[self._df["scale"] <= 0].empty:

499 raise ValueError(

500 "The given lower bounds are greater equal "

501 "than the upper bounds, resulting in a "

502 f"negative scale: \n{str(self._df['scale'])}"

503 )

504

505

506class CalibrationClass:

507 """

508 Class used for calibration of time-series data.

509

510 :param str name:

511 Name of the class, e.g. 'device on'

512 :param float,int start_time:

513 Time at which the class starts

514 :param float,int stop_time:

515 Time at which the class ends

516 :param Goals goals:

517 Goals parameters which are relevant in this class.

518 As this class may be used in the classifier, a Goals-Class

519 may not be available at all times and can be added later.

520 :param TunerParas tuner_paras:

521 As this class may be used in the classifier, a TunerParas-Class

522 may not be available at all times and can be added later.

523 :param list relevant_intervals:

524 List with time-intervals relevant for the calibration.

525 Each list element has to be a tuple with the first element being

526 the start-time as float/int and the second item being the end-time

527 of the interval as float/int.

528 E.g:

529 For a class with start_time=0 and stop_time=1000, given following intervals

530 [(0, 100), [150, 200), (500, 600)]

531 will only evaluate the data between 0-100, 150-200 and 500-600.

532 The given intervals may overlap. Furthermore the intervals do not need

533 to be in an ascending order or be limited to

534 the start_time and end_time parameters.

535 :keyword (pd.DataFrame, ebcpy.data_types.TimeSeriesData) inputs:

536 TimeSeriesData or DataFrame that holds

537 input data for the simulation to run.

538 The time-index should be float index and match the overall

539 ranges set by start- and stop-time.

540 :keyword dict input_kwargs:

541 If inputs are provided, additional input keyword-args passed to the

542 simulation API can be specified.

543 Using FMUs, you don't need to specify anything.

544 Using DymolaAPI, you have to specify 'table_name' and 'file_name'

545 """

546

547 def __init__(self, name, start_time, stop_time, goals=None,

548 tuner_paras=None, relevant_intervals=None, **kwargs):

549 """Initialize class-objects and check correct input."""

550 self.name = name

551 self._start_time = start_time

552 self.stop_time = stop_time

553 self._goals = None

554 self._tuner_paras = None

555 if goals is not None:

556 self.goals = goals

557 if tuner_paras is not None:

558 self.tuner_paras = tuner_paras

559 if relevant_intervals is not None:

560 self.relevant_intervals = relevant_intervals

561 else:

562 # Then all is relevant

563 self.relevant_intervals = [(start_time, stop_time)]

564 self._inputs = None

565 inputs = kwargs.get('inputs', None)

566 if inputs is not None:

567 self.inputs = inputs # Trigger the property setter

568 self.input_kwargs = kwargs.get('input_kwargs', {})

569

570 @property

571 def name(self):

572 """Get name of calibration class"""

573 return self._name

574

575 @name.setter

576 def name(self, name: str):

577 """Set name of calibration class"""

578 if not isinstance(name, str):

579 raise TypeError(f"Name of CalibrationClass is {type(name)} "

580 f"but has to be of type str")

581 self._name = name

582

583 @property

584 def start_time(self) -> Union[float, int]:

585 """Get start time of calibration class"""

586 return self._start_time

587

588 @start_time.setter

589 def start_time(self, start_time: Union[float, int]):

590 """Set start time of calibration class"""

591 if not start_time <= self.stop_time:

592 raise ValueError("The given start-time is "

593 "higher than the stop-time.")

594 self._start_time = start_time

595

596 @property

597 def stop_time(self) -> Union[float, int]:

598 """Get stop time of calibration class"""

599 return self._stop_time

600

601 @stop_time.setter

602 def stop_time(self, stop_time: Union[float, int]):

603 """Set stop time of calibration class"""

604 if not self.start_time <= stop_time:

605 raise ValueError("The given stop-time is "

606 "lower than the start-time.")

607 self._stop_time = stop_time

608

609 @property

610 def tuner_paras(self) -> TunerParas:

611 """Get the tuner parameters of the calibration-class"""

612 return self._tuner_paras

613

614 @tuner_paras.setter

615 def tuner_paras(self, tuner_paras):

616 """

617 Set the tuner parameters for the calibration-class.

618

619 :param tuner_paras: TunerParas

620 """

621 if not isinstance(tuner_paras, TunerParas):

622 raise TypeError(f"Given tuner_paras is of type {type(tuner_paras).__name__} "

623 "but should be type TunerParas")

624 self._tuner_paras = deepcopy(tuner_paras)

625

626 @property

627 def goals(self) -> Goals:

628 """Get current goals instance"""

629 return self._goals

630

631 @goals.setter

632 def goals(self, goals: Goals):

633 """

634 Set the goals object for the calibration-class.

635

636 :param Goals goals:

637 Goals-data-type

638 """

639 if not isinstance(goals, Goals):

640 raise TypeError(f"Given goals parameter is of type {type(goals).__name__} "

641 "but should be type Goals")

642 self._goals = deepcopy(goals)

643

644 @property

645 def relevant_intervals(self) -> list:

646 """Get current relevant_intervals"""

647 return self._relevant_intervals

648

649 @relevant_intervals.setter

650 def relevant_intervals(self, relevant_intervals: list):

651 """Set current relevant_intervals"""

652 self._relevant_intervals = relevant_intervals

653

654 @property

655 def inputs(self) -> Union[TimeSeriesData, pd.DataFrame]:

656 """Get the inputs for this calibration class"""

657 return self._inputs

658

659 @inputs.setter

660 def inputs(self, inputs: Union[TimeSeriesData, pd.DataFrame]):

661 """Set the inputs for this calibration class"""

662 # Check correct index:

663 if not isinstance(inputs, (TimeSeriesData, pd.DataFrame)):

664 raise TypeError(f"Inputs need to be either TimeSeriesData "

665 f"or pd.DataFrame, but you passed {type(inputs)}")

666 if isinstance(inputs.index, pd.DatetimeIndex):

667 inputs = convert_datetime_index_to_float_index(inputs)

668

669 self._inputs = inputs

670

671

672def merge_calibration_classes(calibration_classes):

673 """

674 Given a list of multiple calibration-classes, this function merges given

675 objects by the "name" attribute. Relevant intervals are set, in order

676 to maintain the start and stop-time info.

677

678 :param list calibration_classes:

679 List containing multiple CalibrationClass-Objects

680 :return: list cal_classes_merged:

681 A list containing one CalibrationClass-Object for each different

682 "name" of class.

683

684 Example:

685 >>> cal_classes = [CalibrationClass("on", 0, 100),

686 >>> CalibrationClass("off", 100, 200),

687 >>> CalibrationClass("on", 200, 300)]

688 >>> merged_classes = merge_calibration_classes(cal_classes)

689 Is equal to:

690 >>> merged_classes = [CalibrationClass("on", 0, 300,

691 >>> relevant_intervals=[(0,100), (200,300)]),

692 >>> CalibrationClass("off", 100, 200)]

693

694 """

695 # Use a dict for easy name-access

696 temp_merged = {}

697 for cal_class in calibration_classes:

698 _name = cal_class.name

699 # First create dictionary with all calibration classes

700 if _name in temp_merged:

701 temp_merged[_name]["intervals"] += cal_class.relevant_intervals

702 else:

703 temp_merged[_name] = {"goals": cal_class.goals,

704 "tuner_paras": cal_class.tuner_paras,

705 "intervals": deepcopy(cal_class.relevant_intervals),

706 "inputs": deepcopy(cal_class.inputs),

707 "input_kwargs": deepcopy(cal_class.input_kwargs)

708 }

709

710 # Convert dict to actual calibration-classes

711 cal_classes_merged = []

712 for _name, values in temp_merged.items():

713 # Flatten the list of tuples and get the start- and stop-values

714 start_time = min(sum(values["intervals"], ()))

715 stop_time = max(sum(values["intervals"], ()))

716 cal_classes_merged.append(CalibrationClass(

717 _name, start_time, stop_time,

718 goals=values["goals"],

719 tuner_paras=values["tuner_paras"],

720 relevant_intervals=values["intervals"],

721 inputs=values["inputs"],

722 input_kwargs=values["input_kwargs"])

723 )

724

725 return cal_classes_merged

Coverage for aixcalibuha/data_types.py: 86%

272 statements