Coverage for agentlib_flexquant/modules/flexibility_indicator.py: 95%
217 statements
« prev ^ index » next coverage.py v7.4.4, created at 2025-10-20 14:09 +0000
1"""
2Flexibility indicator module for calculating and distributing energy flexibility offers.
4This module processes power and energy profiles from baseline and shadow MPCs to
5calculate flexibility KPIs, validate profile consistency, and generate flexibility
6offers for energy markets. It handles both positive and negative flexibility with
7optional cost calculations and energy storage corrections.
8"""
9import logging
10import os
11from pathlib import Path
12from typing import Optional
14import agentlib
15import numpy as np
16import pandas as pd
17from pydantic import BaseModel, ConfigDict, Field, model_validator
18from agentlib_flexquant.utils.data_handling import fill_nans, MEAN
20import agentlib_flexquant.data_structures.globals as glbs
21from agentlib_flexquant.data_structures.flex_kpis import (
22 FlexibilityData,
23 FlexibilityKPIs,
24)
25from agentlib_flexquant.data_structures.flex_offer import FlexOffer
28class InputsForCorrectFlexCosts(BaseModel):
29 """Configuration for flexibility cost correction."""
31 enable_energy_costs_correction: bool = Field(
32 name="enable_energy_costs_correction",
33 description=(
34 "Variable determining whether to correct the costs of the flexible energy "
35 "Define the variable for stored electrical energy in the base MPC model and "
36 "config as output if the correction of costs is enabled"
37 ),
38 default=False,
39 )
41 absolute_power_deviation_tolerance: float = Field(
42 name="absolute_power_deviation_tolerance",
43 default=0.1,
44 description="Absolute tolerance in kW within which no warning is thrown",
45 )
47 stored_energy_variable: Optional[str] = Field(
48 name="stored_energy_variable",
49 default=None,
50 description="Name of the variable representing the stored electrical energy in the "
51 "baseline config"
52 )
55class InputsForCalculateFlexCosts(BaseModel):
56 """Configuration for flexibility cost calculation with optional constant pricing."""
58 use_constant_electricity_price: bool = Field(
59 default=False, description="Use constant electricity price"
60 )
61 calculate_flex_costs: bool = Field(
62 default=True, description="Calculate the flexibility cost"
63 )
64 const_electricity_price: float = Field(
65 default=np.nan, description="constant electricity price in ct/kWh"
66 )
68 @model_validator(mode="after")
69 def validate_constant_price(self):
70 """Validate that a valid constant electricity price is provided
71 when constant pricing is enabled."""
72 if self.use_constant_electricity_price and np.isnan(
73 self.const_electricity_price
74 ):
75 raise ValueError(
76 (
77 f"Constant electricity price must have a valid value in float if it is "
78 f"to be used for calculation. "
79 f'Received "use_constant_electricity_price": true, '
80 f'"const_electricity_price": {self.const_electricity_price}. '
81 f'Please specify them correctly in "calculate_costs" field in flex config.'
82 )
83 )
84 return self
87# Pos and neg kpis to get the right names for plotting
88kpis_pos = FlexibilityKPIs(direction="positive")
89kpis_neg = FlexibilityKPIs(direction="negative")
92class FlexibilityIndicatorModuleConfig(agentlib.BaseModuleConfig):
93 """Configuration for flexibility indicator module with power/energy inputs,
94 KPI outputs, and cost calculation settings."""
96 model_config = ConfigDict(extra="forbid")
98 inputs: list[agentlib.AgentVariable] = [
99 agentlib.AgentVariable(
100 name=glbs.POWER_ALIAS_BASE,
101 unit="W",
102 type="pd.Series",
103 description="The power input to the system",
104 ),
105 agentlib.AgentVariable(
106 name=glbs.POWER_ALIAS_NEG,
107 unit="W",
108 type="pd.Series",
109 description="The power input to the system",
110 ),
111 agentlib.AgentVariable(
112 name=glbs.POWER_ALIAS_POS,
113 unit="W",
114 type="pd.Series",
115 description="The power input to the system",
116 ),
117 agentlib.AgentVariable(
118 name=glbs.STORED_ENERGY_ALIAS_BASE,
119 unit="kWh",
120 type="pd.Series",
121 description="Energy stored in the system w.r.t. 0K",
122 ),
123 agentlib.AgentVariable(
124 name=glbs.STORED_ENERGY_ALIAS_NEG,
125 unit="kWh",
126 type="pd.Series",
127 description="Energy stored in the system w.r.t. 0K",
128 ),
129 agentlib.AgentVariable(
130 name=glbs.STORED_ENERGY_ALIAS_POS,
131 unit="kWh",
132 type="pd.Series",
133 description="Energy stored in the system w.r.t. 0K",
134 ),
135 ]
137 outputs: list[agentlib.AgentVariable] = [
138 # Flexibility offer
139 agentlib.AgentVariable(name=glbs.FLEXIBILITY_OFFER, type="FlexOffer"),
140 # Power KPIs
141 agentlib.AgentVariable(
142 name=kpis_neg.power_flex_full.get_kpi_identifier(),
143 unit="W",
144 type="pd.Series",
145 description="Negative power flexibility",
146 ),
147 agentlib.AgentVariable(
148 name=kpis_pos.power_flex_full.get_kpi_identifier(),
149 unit="W",
150 type="pd.Series",
151 description="Positive power flexibility",
152 ),
153 agentlib.AgentVariable(
154 name=kpis_neg.power_flex_offer.get_kpi_identifier(),
155 unit="W",
156 type="pd.Series",
157 description="Negative power flexibility",
158 ),
159 agentlib.AgentVariable(
160 name=kpis_pos.power_flex_offer.get_kpi_identifier(),
161 unit="W",
162 type="pd.Series",
163 description="Positive power flexibility",
164 ),
165 agentlib.AgentVariable(
166 name=kpis_neg.power_flex_offer_min.get_kpi_identifier(),
167 unit="W",
168 type="float",
169 description="Minimum of negative power flexibility",
170 ),
171 agentlib.AgentVariable(
172 name=kpis_pos.power_flex_offer_min.get_kpi_identifier(),
173 unit="W",
174 type="float",
175 description="Minimum of positive power flexibility",
176 ),
177 agentlib.AgentVariable(
178 name=kpis_neg.power_flex_offer_max.get_kpi_identifier(),
179 unit="W",
180 type="float",
181 description="Maximum of negative power flexibility",
182 ),
183 agentlib.AgentVariable(
184 name=kpis_pos.power_flex_offer_max.get_kpi_identifier(),
185 unit="W",
186 type="float",
187 description="Maximum of positive power flexibility",
188 ),
189 agentlib.AgentVariable(
190 name=kpis_neg.power_flex_offer_avg.get_kpi_identifier(),
191 unit="W",
192 type="float",
193 description="Average of negative power flexibility",
194 ),
195 agentlib.AgentVariable(
196 name=kpis_pos.power_flex_offer_avg.get_kpi_identifier(),
197 unit="W",
198 type="float",
199 description="Average of positive power flexibility",
200 ),
201 agentlib.AgentVariable(
202 name=kpis_neg.power_flex_within_boundary.get_kpi_identifier(),
203 unit="-",
204 type="bool",
205 description=(
206 "Variable indicating whether the baseline power and flex power "
207 "align at the horizon end"
208 ),
209 ),
210 agentlib.AgentVariable(
211 name=kpis_pos.power_flex_within_boundary.get_kpi_identifier(),
212 unit="-",
213 type="bool",
214 description=(
215 "Variable indicating whether the baseline power and flex power "
216 "align at the horizon end"
217 ),
218 ),
219 # Energy KPIs
220 agentlib.AgentVariable(
221 name=kpis_neg.energy_flex.get_kpi_identifier(),
222 unit="kWh",
223 type="float",
224 description="Negative energy flexibility",
225 ),
226 agentlib.AgentVariable(
227 name=kpis_pos.energy_flex.get_kpi_identifier(),
228 unit="kWh",
229 type="float",
230 description="Positive energy flexibility",
231 ),
232 # Costs KPIs
233 agentlib.AgentVariable(
234 name=kpis_neg.costs.get_kpi_identifier(),
235 unit="ct",
236 type="float",
237 description="Saved costs due to baseline",
238 ),
239 agentlib.AgentVariable(
240 name=kpis_pos.costs.get_kpi_identifier(),
241 unit="ct",
242 type="float",
243 description="Saved costs due to baseline",
244 ),
245 agentlib.AgentVariable(
246 name=kpis_neg.corrected_costs.get_kpi_identifier(),
247 unit="ct",
248 type="float",
249 description="Corrected saved costs due to baseline",
250 ),
251 agentlib.AgentVariable(
252 name=kpis_pos.corrected_costs.get_kpi_identifier(),
253 unit="ct",
254 type="float",
255 description="Corrected saved costs due to baseline",
256 ),
257 agentlib.AgentVariable(
258 name=kpis_neg.costs_rel.get_kpi_identifier(),
259 unit="ct/kWh",
260 type="float",
261 description="Saved costs due to baseline",
262 ),
263 agentlib.AgentVariable(
264 name=kpis_pos.costs_rel.get_kpi_identifier(),
265 unit="ct/kWh",
266 type="float",
267 description="Saved costs due to baseline",
268 ),
269 agentlib.AgentVariable(
270 name=kpis_neg.corrected_costs_rel.get_kpi_identifier(),
271 unit="ct/kWh",
272 type="float",
273 description="Corrected saved costs per energy due to baseline",
274 ),
275 agentlib.AgentVariable(
276 name=kpis_pos.corrected_costs_rel.get_kpi_identifier(),
277 unit="ct/kWh",
278 type="float",
279 description="Corrected saved costs per energy due to baseline",
280 ),
281 ]
283 parameters: list[agentlib.AgentVariable] = [
284 agentlib.AgentVariable(name=glbs.PREP_TIME, unit="s",
285 description="Preparation time"),
286 agentlib.AgentVariable(name=glbs.MARKET_TIME, unit="s",
287 description="Market time"),
288 agentlib.AgentVariable(name=glbs.FLEX_EVENT_DURATION, unit="s",
289 description="time to switch objective"),
290 agentlib.AgentVariable(name=glbs.TIME_STEP, unit="s",
291 description="timestep of the mpc solution"),
292 agentlib.AgentVariable(name=glbs.PREDICTION_HORIZON, unit="-",
293 description="prediction horizon of the mpc solution"),
294 agentlib.AgentVariable(name=glbs.COLLOCATION_TIME_GRID, alias=glbs.COLLOCATION_TIME_GRID,
295 description="Time grid of the mpc model output")
296 ]
298 results_file: Optional[Path] = Field(
299 default=Path("flexibility_indicator.csv"),
300 description="User specified results file name",
301 )
302 save_results: Optional[bool] = Field(
303 validate_default=True,
304 default=True
305 )
306 price_variable: str = Field(
307 default="c_pel", description="Name of the price variable sent by a predictor",
308 )
309 power_unit: str = Field(
310 default="kW",
311 description="Unit of the power variable"
312 )
313 integration_method: glbs.INTEGRATION_METHOD = Field(
314 default=glbs.LINEAR,
315 description="Method set to integrate series variable"
316 )
317 shared_variable_fields: list[str] = ["outputs"]
319 correct_costs: InputsForCorrectFlexCosts = InputsForCorrectFlexCosts()
320 calculate_costs: InputsForCalculateFlexCosts = InputsForCalculateFlexCosts()
322 @model_validator(mode="after")
323 def check_results_file_extension(self):
324 """Validate that results_file has a .csv extension."""
325 if self.results_file and self.results_file.suffix != ".csv":
326 raise ValueError(
327 f"Invalid file extension for 'results_file': '{self.results_file}'. "
328 f"Expected a '.csv' file."
329 )
330 return self
333class FlexibilityIndicatorModule(agentlib.BaseModule):
334 """Module for calculating flexibility KPIs and generating flexibility offers
335 from MPC power/energy profiles."""
337 config: FlexibilityIndicatorModuleConfig
338 data: FlexibilityData
340 def __init__(self, *args, **kwargs):
341 super().__init__(*args, **kwargs)
342 self.var_list = []
343 for variable in self.variables:
344 if variable.name in [glbs.FLEXIBILITY_OFFER]:
345 continue
346 self.var_list.append(variable.name)
347 self.time = []
348 self.in_provision = False
349 self.offer_count = 0
350 self.data = FlexibilityData(
351 prep_time=self.get(glbs.PREP_TIME).value,
352 market_time=self.get(glbs.MARKET_TIME).value,
353 flex_event_duration=self.get(glbs.FLEX_EVENT_DURATION).value,
354 time_step=self.get(glbs.TIME_STEP).value,
355 prediction_horizon=self.get(glbs.PREDICTION_HORIZON).value,
356 )
357 self.df = pd.DataFrame(columns=pd.Series(self.var_list))
359 def register_callbacks(self):
360 inputs = self.config.inputs
361 for var in inputs:
362 self.agent.data_broker.register_callback(
363 name=var.name, alias=var.name, callback=self.callback
364 )
365 self.agent.data_broker.register_callback(
366 name="in_provision", alias="in_provision", callback=self.callback
367 )
369 def process(self):
370 """Yield control to the simulation environment and wait for events."""
371 yield self.env.event()
373 def callback(self, inp, name):
374 """Handle incoming data by storing power/energy profiles and triggering
375 flexibility calculations when all required inputs are available."""
376 if name == "in_provision":
377 self.in_provision = inp.value
378 if self.in_provision:
379 self._set_inputs_to_none()
381 if not self.in_provision:
382 if name == glbs.POWER_ALIAS_BASE:
383 self.data.power_profile_base = self.data.unify_inputs(inp.value)
384 elif name == glbs.POWER_ALIAS_NEG:
385 self.data.power_profile_flex_neg = self.data.unify_inputs(inp.value)
386 elif name == glbs.POWER_ALIAS_POS:
387 self.data.power_profile_flex_pos = self.data.unify_inputs(inp.value)
388 elif name == glbs.STORED_ENERGY_ALIAS_BASE:
389 self.data.stored_energy_profile_base = self.data.unify_inputs(inp.value)
390 elif name == glbs.STORED_ENERGY_ALIAS_NEG:
391 self.data.stored_energy_profile_flex_neg = self.data.unify_inputs(inp.value)
392 elif name == glbs.STORED_ENERGY_ALIAS_POS:
393 self.data.stored_energy_profile_flex_pos = self.data.unify_inputs(inp.value)
394 elif name == self.config.price_variable:
395 if not self.config.calculate_costs.use_constant_electricity_price:
396 # price comes from predictor
397 self.data.electricity_price_series = self.data.unify_inputs(inp.value,
398 mpc=False)
400 # set the constant electricity price series if given
401 if (
402 self.config.calculate_costs.use_constant_electricity_price
403 and self.data.electricity_price_series is None
404 ):
405 # get the index for the electricity price series
406 n = self.get(glbs.PREDICTION_HORIZON).value
407 ts = self.get(glbs.TIME_STEP).value
408 grid = np.arange(0, n * ts + ts, ts)
409 # fill the electricity_price_series with values
410 self.data.electricity_price_series = pd.Series(
411 [self.config.calculate_costs.const_electricity_price for i in grid], index=grid)
413 necessary_input_for_calc_flex = [
414 self.data.power_profile_base,
415 self.data.power_profile_flex_neg,
416 self.data.power_profile_flex_pos,
417 ]
419 if self.config.calculate_costs.calculate_flex_costs:
420 necessary_input_for_calc_flex.append(self.data.electricity_price_series)
422 if (all(var is not None for var in necessary_input_for_calc_flex) and
423 len(necessary_input_for_calc_flex) == 4):
424 # align the index of price variable to the index of inputs from mpc;
425 # electricity price signal is usually steps
426 necessary_input_for_calc_flex[-1] = self.data.electricity_price_series.reindex(
427 self.data.power_profile_base.index).ffill()
429 if self.config.correct_costs.enable_energy_costs_correction:
430 necessary_input_for_calc_flex.extend(
431 [
432 self.data.stored_energy_profile_base,
433 self.data.stored_energy_profile_flex_neg,
434 self.data.stored_energy_profile_flex_pos,
435 ]
436 )
438 if all(var is not None for var in necessary_input_for_calc_flex):
440 # check the power profile end deviation
441 if not self.config.correct_costs.enable_energy_costs_correction:
442 self.check_power_end_deviation(
443 tol=self.config.correct_costs.absolute_power_deviation_tolerance
444 )
446 # Calculate the flexibility, send the offer, write and save the results
447 self.calc_and_send_offer()
449 # set the values to None to reset the callback
450 self._set_inputs_to_none()
452 def get_results(self) -> Optional[pd.DataFrame]:
453 """Open results file of flexibility_indicator.py."""
454 results_file = self.config.results_file
455 try:
456 results = pd.read_csv(results_file, header=[0], index_col=[0, 1])
457 return results
458 except FileNotFoundError:
459 self.logger.error("Results file %s was not found.", results_file)
460 return None
462 def write_results(self, df: pd.DataFrame, ts: float, n: int) -> pd.DataFrame:
463 """Write every data of variables in self.var_list in an DataFrame.
465 DataFrame will be updated every time step
467 Args:
468 df: DataFrame which is initialised as an empty DataFrame with columns
469 according to self.var_list
470 ts: time step
471 n: number of time steps during prediction horizon
473 Returns:
474 DataFrame with results of every variable in self.var_list
476 """
477 results = []
478 now = self.env.now
480 # First, collect all series and their indices
481 all_series = []
482 for name in self.var_list:
483 # Get the appropriate values based on name
484 if name == glbs.POWER_ALIAS_BASE:
485 values = self.data.power_profile_base
486 elif name == glbs.POWER_ALIAS_NEG:
487 values = self.data.power_profile_flex_neg
488 elif name == glbs.POWER_ALIAS_POS:
489 values = self.data.power_profile_flex_pos
490 elif name == glbs.STORED_ENERGY_ALIAS_BASE:
491 values = self.data.stored_energy_profile_base
492 elif name == glbs.STORED_ENERGY_ALIAS_NEG:
493 values = self.data.stored_energy_profile_flex_neg
494 elif name == glbs.STORED_ENERGY_ALIAS_POS:
495 values = self.data.stored_energy_profile_flex_pos
496 elif name == self.config.price_variable:
497 values = self.data.electricity_price_series
498 elif name == glbs.COLLOCATION_TIME_GRID:
499 value = self.get(name).value
500 values = pd.Series(index=value, data=value)
501 else:
502 values = self.get(name).value
504 # Convert to Series if not already
505 if not isinstance(values, pd.Series):
506 values = pd.Series(values)
508 all_series.append((name, values))
510 # Create the standard grid for reference
511 standard_grid = np.arange(0, n * ts, ts)
513 # Find the union of all indices to create a comprehensive grid
514 all_indices = set(standard_grid)
515 for _, series in all_series:
516 all_indices.update(series.index)
517 combined_index = sorted(all_indices)
519 # Reindex all series to the combined grid
520 for i, (name, series) in enumerate(all_series):
521 # Reindex to the comprehensive grid
522 reindexed = series.reindex(combined_index)
523 results.append(reindexed)
525 if not now % ts:
526 self.time.append(now)
527 new_df = pd.DataFrame(results).T
528 new_df.columns = self.var_list
529 # Rename time_step variable column
530 new_df.rename(
531 columns={glbs.TIME_STEP: f"{glbs.TIME_STEP}_mpc"}, inplace=True
532 )
533 new_df.index.direction = "time"
534 new_df[glbs.TIME_STEP] = now
535 new_df.set_index([glbs.TIME_STEP, new_df.index], inplace=True)
536 df = pd.concat([df, new_df])
537 # set the indices once again as concat cant handle indices properly
538 indices = pd.MultiIndex.from_tuples(
539 df.index, names=[glbs.TIME_STEP, "time"]
540 )
541 df.set_index(indices, inplace=True)
542 # Drop column time_step and keep it as an index only
543 if glbs.TIME_STEP in df.columns:
544 df.drop(columns=[glbs.TIME_STEP], inplace=True)
546 return df
548 def cleanup_results(self):
549 """Remove the existing result files."""
550 results_file = self.config.results_file
551 if not results_file:
552 return
553 os.remove(results_file)
555 def calc_and_send_offer(self):
556 """Calculate the flexibility KPIs for current predictions, send the flex offer
557 and set the outputs, write and save the results."""
558 # Calculate the flexibility KPIs for current predictions
559 collocation_time_grid = self.get(glbs.COLLOCATION_TIME_GRID).value
560 self.data.calculate(
561 enable_energy_costs_correction=self.config.correct_costs.enable_energy_costs_correction,
562 calculate_flex_cost=self.config.calculate_costs.calculate_flex_costs,
563 integration_method=self.config.integration_method,
564 collocation_time_grid=collocation_time_grid)
566 # get the full index during flex enevt including mpc_time_grid index and the
567 # collocation index
568 full_index = np.sort(np.concatenate([collocation_time_grid, self.data.mpc_time_grid]))
569 flex_begin = self.get(glbs.MARKET_TIME).value + self.get(glbs.PREP_TIME).value
570 flex_end = flex_begin + self.get(glbs.FLEX_EVENT_DURATION).value
571 full_flex_offer_index = full_index[(full_index >= flex_begin) & (full_index <= flex_end)]
573 # reindex the power profiles to not send the simulation points to the market, but only
574 # the values on the collocation points and the forward mean of them
575 base_power_profile = self.data.power_profile_base.reindex(
576 collocation_time_grid).reindex(full_flex_offer_index)
577 pos_diff_profile = self.data.kpis_pos.power_flex_offer.value.reindex(
578 collocation_time_grid).reindex(full_flex_offer_index)
579 neg_diff_profile = self.data.kpis_neg.power_flex_offer.value.reindex(
580 collocation_time_grid).reindex(full_flex_offer_index)
582 # fill the mpc_time_grid with forward mean
583 base_power_profile = fill_nans(base_power_profile, method=MEAN)
584 pos_diff_profile = fill_nans(pos_diff_profile, method=MEAN)
585 neg_diff_profile = fill_nans(neg_diff_profile, method=MEAN)
587 # Send flex offer
588 self.send_flex_offer(
589 name=glbs.FLEXIBILITY_OFFER,
590 base_power_profile=base_power_profile,
591 pos_diff_profile=pos_diff_profile,
592 pos_price=self.data.kpis_pos.costs.value,
593 neg_diff_profile=neg_diff_profile,
594 neg_price=self.data.kpis_neg.costs.value,
595 )
597 # set outputs
598 for kpi in self.data.get_kpis().values():
599 if kpi.get_kpi_identifier() not in [
600 kpis_pos.power_flex_within_boundary.get_kpi_identifier(),
601 kpis_neg.power_flex_within_boundary.get_kpi_identifier(),
602 ]:
603 for output in self.config.outputs:
604 if output.name == kpi.get_kpi_identifier():
605 self.set(output.name, kpi.value)
607 # write results
608 self.df = self.write_results(
609 df=self.df,
610 ts=self.get(glbs.TIME_STEP).value,
611 n=self.get(glbs.PREDICTION_HORIZON).value,
612 )
614 # save results
615 if self.config.save_results:
616 self.df.to_csv(self.config.results_file)
618 def send_flex_offer(
619 self,
620 name: str,
621 base_power_profile: pd.Series,
622 pos_diff_profile: pd.Series,
623 pos_price: float,
624 neg_diff_profile: pd.Series,
625 neg_price: float,
626 timestamp: float = None,
627 ):
628 """Send a flex offer as an agent Variable.
630 The first offer is dismissed, since the different MPCs need one time step
631 to fully initialize.
633 Args:
634 name: name of the agent variable
635 base_power_profile: time series of power from baseline mpc
636 pos_diff_profile: power profile for the positive difference (base-pos)
637 in flexibility event time grid
638 pos_price: price for positive flexibility
639 neg_diff_profile: power profile for the negative difference (neg-base)
640 in flexibility event time grid
641 neg_price: price for negative flexibility
642 timestamp: the time offer was generated
644 """
645 if self.offer_count > 0:
646 var = self._variables_dict[name]
647 var.value = FlexOffer(
648 base_power_profile=base_power_profile,
649 pos_diff_profile=pos_diff_profile,
650 pos_price=pos_price,
651 neg_diff_profile=neg_diff_profile,
652 neg_price=neg_price,
653 )
654 if timestamp is None:
655 timestamp = self.env.time
656 var.timestamp = timestamp
657 self.agent.data_broker.send_variable(
658 variable=var.copy(update={"source": self.source}), copy=False,
659 )
660 self.offer_count += 1
662 def _set_inputs_to_none(self):
663 self.data.power_profile_base = None
664 self.data.power_profile_flex_neg = None
665 self.data.power_profile_flex_pos = None
666 self.data.electricity_price_series = None
667 self.data.stored_energy_profile_base = None
668 self.data.stored_energy_profile_flex_neg = None
669 self.data.stored_energy_profile_flex_pos = None
671 def check_power_end_deviation(self, tol: float):
672 """Calculate the deviation of the final value of the power profiles
673 and warn the user if it exceeds the tolerance."""
674 logger = logging.getLogger(__name__)
675 dev_pos = np.mean(
676 self.data.power_profile_flex_pos.values[-4:]
677 - self.data.power_profile_base.values[-4:]
678 )
679 dev_neg = np.mean(
680 self.data.power_profile_flex_neg.values[-4:]
681 - self.data.power_profile_base.values[-4:]
682 )
683 if abs(dev_pos) > tol:
684 logger.warning(
685 "There is an average deviation of %.6f kW between the final values of "
686 "power profiles of positive shadow MPC and the baseline. "
687 "Correction of energy costs might be necessary.",
688 dev_pos,
689 )
690 self.set(kpis_pos.power_flex_within_boundary.get_kpi_identifier(), False)
691 else:
692 self.set(kpis_pos.power_flex_within_boundary.get_kpi_identifier(), True)
693 if abs(dev_neg) > tol:
694 logger.warning(
695 "There is an average deviation of %.6f kW between the final values of "
696 "power profiles of negative shadow MPC and the baseline. "
697 "Correction of energy costs might be necessary.",
698 dev_neg,
699 )
700 self.set(kpis_neg.power_flex_within_boundary.get_kpi_identifier(), False)
701 else:
702 self.set(kpis_neg.power_flex_within_boundary.get_kpi_identifier(), True)