Coverage for agentlib_flexquant/utils/interactive.py: 32%

1import webbrowser 

2from typing import Optional, Union, get_args 

3 

4import pandas as pd 

5from agentlib.core.agent import AgentConfig 

6from agentlib_mpc.utils import TIME_CONVERSION, TimeConversionTypes 

7from agentlib_mpc.utils.analysis import mpc_at_time_step 

8from agentlib_mpc.utils.plotting.interactive import get_port 

9from dash import Dash, Input, Output, ctx, dcc, html 

10from plotly import graph_objects as go 

11from pydantic import FilePath 

12 

13import agentlib_flexquant.data_structures.flex_results as flex_results 

14import agentlib_flexquant.data_structures.globals as glbs 

15from agentlib_flexquant.data_structures.flex_kpis import FlexibilityKPIs 

16from agentlib_flexquant.data_structures.flex_offer import OfferStatus 

17from agentlib_flexquant.data_structures.flexquant import FlexQuantConfig 

18 

19 

20class CustomBound: 

21 """Dataclass to let the user define custom bounds for the mpc variables. 

22 

23 for_variable -- The name of the variable to plot the bounds into 

24 lower bound -- The lower bound of the variable as the name of the lower bound variable 

25 in the MPC 

26 upper bound -- The upper bound of the variable as the name of the upper bound variable 

27 in the MPC 

28 

29 """ 

30 

31 for_variable: str 

32 lower_bound: Optional[str] 

33 upper_bound: Optional[str] 

34 

35 def __init__( 

36 self, 

37 for_variable: str, 

38 lb_name: Optional[str] = None, 

39 ub_name: Optional[str] = None, 

40 ): 

41 self.for_variable = for_variable 

42 self.lower_bound = lb_name 

43 self.upper_bound = ub_name 

44 

45 

46class Dashboard(flex_results.Results): 

47 """Class for the dashboard of flexquant""" 

48 

49 # Constants for plotting variables 

50 MPC_ITERATIONS: str = "stats_iter_count" 

51 

52 # Label for the positive and negative flexibilities 

53 label_positive: str = "positive" 

54 label_negative: str = "negative" 

55 

56 # Keys for line properties 

57 bounds_key: str = "bounds" 

58 characteristic_times_current_key: str = "characteristic_times_current" 

59 characteristic_times_accepted_key: str = "characteristic_times_accepted" 

60 

61 # Custom settings 

62 custom_bounds: list[CustomBound] = [] 

63 

64 def __init__( 

65 self, 

66 flex_config: Optional[Union[str, FilePath, FlexQuantConfig]] = None, 

67 simulator_agent_config: Optional[Union[str, FilePath, AgentConfig]] = None, 

68 generated_flex_files_base_path: Optional[Union[str, FilePath]] = None, 

69 results: Union[str, FilePath, dict[str, dict[str, pd.DataFrame]]] = None, 

70 to_timescale: TimeConversionTypes = "hours", 

71 port: int = None, 

72 ): 

73 super().__init__( 

74 flex_config=flex_config, 

75 simulator_agent_config=simulator_agent_config, 

76 generated_flex_files_base_path=generated_flex_files_base_path, 

77 results=results, 

78 to_timescale=to_timescale, 

79 ) 

80 self.port = port 

81 self.current_timescale_input = self.current_timescale_of_data 

82 # remove FMU from dataclass to enable pickling/multiprocessing of Dashboard 

83 if hasattr(self.simulator_module_config, "model"): 

84 self.simulator_module_config.model_config["frozen"] = False 

85 delattr(self.simulator_module_config, "model") 

86 

87 # Define line properties 

88 self.LINE_PROPERTIES: dict = { 

89 self.simulator_agent_config.id: {"color": "black",}, 

90 self.baseline_agent_config.id: {"color": "black",}, 

91 self.neg_flex_agent_config.id: {"color": "red",}, 

92 self.pos_flex_agent_config.id: {"color": "blue",}, 

93 self.bounds_key: {"color": "grey",}, 

94 self.characteristic_times_current_key: {"color": "grey", "dash": "dash",}, 

95 self.characteristic_times_accepted_key: {"color": "yellow",}, 

96 } 

97 

98 # KPIS 

99 kpis_pos = FlexibilityKPIs(direction="positive") 

100 self.kpi_names_pos = kpis_pos.get_name_dict() 

101 kpis_neg = FlexibilityKPIs(direction="negative") 

102 self.kpi_names_neg = kpis_neg.get_name_dict() 

103 

104 # Get variables for plotting 

105 # MPC stats 

106 self.plotting_variables = [self.MPC_ITERATIONS] 

107 # MPC and sim variables 

108 self.intersection_mpcs_sim = self.get_intersection_mpcs_sim() 

109 self.plotting_variables.extend( 

110 [key for key in self.intersection_mpcs_sim.keys()] 

111 ) 

112 # Flexibility kpis 

113 self.plotting_variables.append(kpis_pos.energy_flex.name) 

114 self.plotting_variables.append(kpis_pos.costs.name) 

115 # for kpi in kpis_pos.get_kpi_dict(direction_name=False).values(): 

116 # if not isinstance(kpi.value, pd.Series): 

117 # self.plotting_variables.append(kpi.name) 

118 

119 def _plot_mpc_stats(self, fig: go.Figure, variable: str) -> go.Figure: 

120 """ plot the statics of the baseline and shadow mpcs. 

121 

122 Args: 

123 fig: the figure to be updated 

124 variable: the statics variable to be plotted 

125 

126 Returns: 

127 The updated figure 

128 

129 """ 

130 fig.add_trace( 

131 go.Scatter( 

132 name=self.baseline_agent_config.id, 

133 x=self.df_baseline_stats.index, 

134 y=self.df_baseline_stats[variable], 

135 mode="markers", 

136 line=self.LINE_PROPERTIES[self.baseline_agent_config.id], 

137 ) 

138 ) 

139 fig.add_trace( 

140 go.Scatter( 

141 name=self.pos_flex_agent_config.id, 

142 x=self.df_pos_flex_stats.index, 

143 y=self.df_pos_flex_stats[variable], 

144 mode="markers", 

145 line=self.LINE_PROPERTIES[self.pos_flex_agent_config.id], 

146 ) 

147 ) 

148 fig.add_trace( 

149 go.Scatter( 

150 name=self.neg_flex_agent_config.id, 

151 x=self.df_neg_flex_stats.index, 

152 y=self.df_neg_flex_stats[variable], 

153 mode="markers", 

154 line=self.LINE_PROPERTIES[self.neg_flex_agent_config.id], 

155 ) 

156 ) 

157 return fig 

158 

159 def _plot_one_mpc_variable( 

160 self, fig: go.Figure, variable: str, time_step: float 

161 ) -> go.Figure: 

162 """Plot the mpc series for the specified variable at the specified time step. 

163 

164 Args: 

165 fig: the figure to be updated 

166 variable: the variable to be plotted 

167 time_step: the time step to be plotted 

168 

169 Returns: 

170 The updated figure 

171 

172 """ 

173 # Get the mpc data for the plot 

174 series_neg = mpc_at_time_step( 

175 data=self.df_neg_flex, 

176 time_step=time_step, 

177 variable=self.intersection_mpcs_sim[variable][ 

178 self.neg_flex_module_config.module_id 

179 ], 

180 index_offset=False, 

181 ) 

182 series_pos = mpc_at_time_step( 

183 data=self.df_pos_flex, 

184 time_step=time_step, 

185 variable=self.intersection_mpcs_sim[variable][ 

186 self.pos_flex_module_config.module_id 

187 ], 

188 index_offset=False, 

189 ) 

190 series_bas = mpc_at_time_step( 

191 data=self.df_baseline, 

192 time_step=time_step, 

193 variable=self.intersection_mpcs_sim[variable][ 

194 self.baseline_module_config.module_id 

195 ], 

196 index_offset=False, 

197 ) 

198 

199 def _add_step_to_data(s: pd.Series) -> pd.Series: 

200 """ shift the index of the series """ 

201 s_concat = s.copy().shift(periods=1) 

202 s_concat.index = s.index - 0.01 * (s.index[1] - s.index[0]) 

203 for ind, val in s_concat.items(): 

204 s[ind] = val 

205 s.sort_index(inplace=True) 

206 return s 

207 

208 # Manage nans 

209 for series in [series_neg, series_pos, series_bas]: 

210 if variable in [ 

211 control.name for control in self.baseline_module_config.controls 

212 ]: 

213 series.dropna(inplace=True) 

214 series = _add_step_to_data(s=series) 

215 series.dropna(inplace=True) 

216 

217 # Plot the data 

218 try: 

219 df_sim = self.df_simulation[ 

220 self.intersection_mpcs_sim[variable][ 

221 self.simulator_module_config.module_id 

222 ] 

223 ] 

224 fig.add_trace( 

225 go.Scatter( 

226 name=self.simulator_agent_config.id, 

227 x=df_sim.index, 

228 y=df_sim, 

229 mode="lines", 

230 line=self.LINE_PROPERTIES[self.simulator_agent_config.id], 

231 zorder=2, 

232 ) 

233 ) 

234 except KeyError: 

235 # E.g. when the simulator variable name was not found from the intersection 

236 pass 

237 

238 fig.add_trace( 

239 go.Scatter( 

240 name=self.baseline_agent_config.id, 

241 x=series_bas.index, 

242 y=series_bas, 

243 mode="lines", 

244 line=self.LINE_PROPERTIES[self.baseline_agent_config.id] 

245 | {"dash": "dash"}, 

246 zorder=3, 

247 ) 

248 ) 

249 fig.add_trace( 

250 go.Scatter( 

251 name=self.neg_flex_agent_config.id, 

252 x=series_neg.index, 

253 y=series_neg, 

254 mode="lines", 

255 line=self.LINE_PROPERTIES[self.neg_flex_agent_config.id] 

256 | {"dash": "dash"}, 

257 zorder=4, 

258 ) 

259 ) 

260 fig.add_trace( 

261 go.Scatter( 

262 name=self.pos_flex_agent_config.id, 

263 x=series_pos.index, 

264 y=series_pos, 

265 mode="lines", 

266 line=self.LINE_PROPERTIES[self.pos_flex_agent_config.id] 

267 | {"dash": "dash"}, 

268 zorder=4, 

269 ) 

270 ) 

271 

272 # Get the data for the bounds 

273 def _get_mpc_series(var_type: str, var_name: str) -> pd.Series: 

274 return self.df_baseline[(var_type, var_name)].xs(0, level=1) 

275 

276 def _get_bound(var_name: str) -> Optional[pd.Series]: 

277 if var_name in self.df_baseline.columns.get_level_values(1): 

278 try: 

279 bound = _get_mpc_series(var_type="variable", var_name=var_name) 

280 except KeyError: 

281 bound = _get_mpc_series(var_type="parameter", var_name=var_name) 

282 else: 

283 bound = None 

284 return bound 

285 

286 df_lb = None 

287 df_ub = None 

288 for custom_bound in self.custom_bounds: 

289 if variable == custom_bound.for_variable: 

290 df_lb = _get_bound(custom_bound.lower_bound) 

291 df_ub = _get_bound(custom_bound.upper_bound) 

292 if variable in [ 

293 control.name for control in self.baseline_module_config.controls 

294 ]: 

295 df_lb = _get_mpc_series(var_type="lower", var_name=variable) 

296 df_ub = _get_mpc_series(var_type="upper", var_name=variable) 

297 

298 # Plot bounds 

299 if df_lb is not None: 

300 fig.add_trace( 

301 go.Scatter( 

302 name="Lower bound", 

303 x=df_lb.index, 

304 y=df_lb, 

305 mode="lines", 

306 line=self.LINE_PROPERTIES[self.bounds_key], 

307 zorder=1, 

308 ) 

309 ) 

310 if df_ub is not None: 

311 fig.add_trace( 

312 go.Scatter( 

313 name="Upper bound", 

314 x=df_ub.index, 

315 y=df_ub, 

316 mode="lines", 

317 line=self.LINE_PROPERTIES[self.bounds_key], 

318 zorder=1, 

319 ) 

320 ) 

321 

322 return fig 

323 

324 def _plot_flexibility_kpi(self, fig: go.Figure, variable: str) -> go.Figure: 

325 """Plot the flexibility kpi. 

326 

327 Args: 

328 fig: the figure to be updated 

329 variable: the kpi variable to be plotted 

330 

331 Returns: 

332 The updated figure 

333 

334 """ 

335 df_ind = self.df_indicator.xs(0, level=1) 

336 # if the variable only has NaN, don't plot 

337 if df_ind[self.kpi_names_pos[variable]].isna().all(): 

338 return 

339 fig.add_trace( 

340 go.Scatter( 

341 name=self.label_positive, 

342 x=df_ind.index, 

343 y=df_ind[self.kpi_names_pos[variable]], 

344 mode="lines+markers", 

345 line=self.LINE_PROPERTIES[self.pos_flex_agent_config.id], 

346 ) 

347 ) 

348 fig.add_trace( 

349 go.Scatter( 

350 name=self.label_negative, 

351 x=df_ind.index, 

352 y=df_ind[self.kpi_names_neg[variable]], 

353 mode="lines+markers", 

354 line=self.LINE_PROPERTIES[self.neg_flex_agent_config.id], 

355 ) 

356 ) 

357 return fig 

358 

359 def _plot_market_results(self, fig: go.Figure, variable: str) -> go.Figure: 

360 """Plot the market results. 

361 

362 Args: 

363 fig: the figure to be updated 

364 variable: the variable to be plotted 

365 

366 Returns: 

367 The updated figure 

368 

369 """ 

370 df_flex_market_index = self.df_market.index.droplevel("time") 

371 if variable in self.df_market.columns: 

372 fig.add_trace( 

373 go.Scatter( 

374 x=df_flex_market_index, 

375 y=self.df_market[variable], 

376 mode="lines+markers", 

377 line=self.LINE_PROPERTIES[self.pos_flex_agent_config.id], 

378 ) 

379 ) 

380 else: 

381 pos_var = f"pos_{variable}" 

382 neg_var = f"neg_{variable}" 

383 fig.add_trace( 

384 go.Scatter( 

385 name=self.label_positive, 

386 x=df_flex_market_index, 

387 y=self.df_market[pos_var], 

388 mode="lines+markers", 

389 line=self.LINE_PROPERTIES[self.pos_flex_agent_config.id], 

390 ) 

391 ) 

392 fig.add_trace( 

393 go.Scatter( 

394 name=self.label_negative, 

395 x=df_flex_market_index, 

396 y=self.df_market[neg_var], 

397 mode="lines+markers", 

398 line=self.LINE_PROPERTIES[self.neg_flex_agent_config.id], 

399 ) 

400 ) 

401 return fig 

402 

403 def _get_characteristic_times(self, at_time_step: float) -> (float, float, float): 

404 """Get the characteristic times. 

405 

406 Args: 

407 at_time_step: the time at which we want to get the characteristic times 

408 

409 Returns: 

410 market_time, prep_time and flex_event_duration 

411 

412 """ 

413 df_characteristic_times = self.df_indicator.xs(0, level="time") 

414 rel_market_time = ( 

415 df_characteristic_times.loc[at_time_step, glbs.MARKET_TIME] 

416 / TIME_CONVERSION[self.current_timescale_of_data] 

417 ) 

418 rel_prep_time = ( 

419 df_characteristic_times.loc[at_time_step, glbs.PREP_TIME] 

420 / TIME_CONVERSION[self.current_timescale_of_data] 

421 ) 

422 flex_event_duration = ( 

423 df_characteristic_times.loc[at_time_step, glbs.FLEX_EVENT_DURATION] 

424 / TIME_CONVERSION[self.current_timescale_of_data] 

425 ) 

426 return rel_market_time, rel_prep_time, flex_event_duration 

427 

428 def _mark_time( 

429 self, fig: go.Figure, at_time_step: float, line_prop: dict 

430 ) -> go.Figure: 

431 fig.add_vline(x=at_time_step, line=line_prop, layer="below") 

432 return fig 

433 

434 def _mark_characteristic_times( 

435 self, fig: go.Figure, offer_time: Union[float, int] = 0, line_prop: dict = None 

436 ) -> go.Figure: 

437 """Add markers of the characteristic times to the plot for a time step. 

438 

439 Args: 

440 fig: the figure to plot the results into 

441 offer_time: When to show the markers 

442 line_prop: the graphic properties of the lines as in plotly 

443 

444 Returns: 

445 The updated figure 

446 

447 """ 

448 if line_prop is None: 

449 line_prop = self.LINE_PROPERTIES[self.characteristic_times_current_key] 

450 try: 

451 ( 

452 rel_market_time, 

453 rel_prep_time, 

454 flex_event_duration, 

455 ) = self._get_characteristic_times(offer_time) 

456 self._mark_time(fig=fig, at_time_step=offer_time, line_prop=line_prop) 

457 self._mark_time( 

458 fig=fig, 

459 at_time_step=offer_time + rel_market_time, 

460 line_prop=line_prop, 

461 ) 

462 self._mark_time( 

463 fig=fig, 

464 at_time_step=offer_time + rel_prep_time + rel_market_time, 

465 line_prop=line_prop, 

466 ) 

467 self._mark_time( 

468 fig=fig, 

469 at_time_step=offer_time 

470 + rel_prep_time 

471 + rel_market_time 

472 + flex_event_duration, 

473 line_prop=line_prop, 

474 ) 

475 except KeyError: 

476 pass # No data of characteristic times available, e.g. if offer accepted 

477 return fig 

478 

479 def _mark_characteristic_times_of_accepted_offers(self, fig: go.Figure) -> go.Figure: 

480 """Add markers of the characteristic times for accepted offers to the plot.""" 

481 if self.df_market is not None: 

482 if ( 

483 self.df_market["status"] 

484 .isin( 

485 [ 

486 OfferStatus.ACCEPTED_NEGATIVE.value, 

487 OfferStatus.ACCEPTED_POSITIVE.value, 

488 ] 

489 ) 

490 .any() 

491 ): 

492 df_accepted_offers = self.df_market["status"].str.contains( 

493 pat="OfferStatus.accepted" 

494 ) 

495 for i in df_accepted_offers.index.to_list(): 

496 if df_accepted_offers[i]: 

497 fig = self._mark_characteristic_times( 

498 fig=fig, 

499 offer_time=i[0], 

500 line_prop=self.LINE_PROPERTIES[ 

501 self.characteristic_times_accepted_key 

502 ], 

503 ) 

504 return fig 

505 

506 def _create_plot( 

507 self, 

508 variable: str, 

509 at_time_step: float, 

510 show_accepted_characteristic_times: bool = True, 

511 show_current_characteristic_times: bool = True, 

512 zoom_to_offer_window: bool = False, 

513 zoom_to_prediction_interval: bool = False, 

514 ) -> go.Figure: 

515 """Create a plot for one variable 

516 

517 Args: 

518 variable: the variable to plot 

519 at_time_step: the time_step to show the mpc predictions 

520 and the characteristic times 

521 show_accepted_characteristic_times: whether to show 

522 the accepted characteristic times 

523 show_current_characteristic_times: whether to show 

524 the current characteristic times 

525 zoom_to_offer_window: whether to zoom to offer window 

526 zoom_to_prediction_interval: wether to zoom to prediction interval 

527 

528 Returns: 

529 The created figure 

530 

531 """ 

532 # Create the figure 

533 fig = go.Figure() 

534 self._mark_time(fig=fig, at_time_step=at_time_step, line_prop={"color": "green"}) 

535 if show_accepted_characteristic_times: 

536 self._mark_characteristic_times_of_accepted_offers(fig=fig) 

537 

538 # Plot variable 

539 if variable in self.df_baseline_stats.columns: 

540 self._plot_mpc_stats(fig=fig, variable=variable) 

541 elif variable in self.intersection_mpcs_sim.keys(): 

542 self._plot_one_mpc_variable( 

543 fig=fig, variable=variable, time_step=at_time_step 

544 ) 

545 if show_current_characteristic_times: 

546 self._mark_characteristic_times(fig=fig, offer_time=at_time_step) 

547 elif any(variable in label for label in self.df_indicator.columns): 

548 self._plot_flexibility_kpi(fig=fig, variable=variable) 

549 elif any(variable in label for label in self.df_market.columns): 

550 self._plot_market_results(fig=fig, variable=variable) 

551 else: 

552 raise ValueError(f"No plotting function found for variable {variable}") 

553 

554 # Set layout 

555 if zoom_to_offer_window: 

556 try: 

557 ( 

558 rel_market_time, 

559 rel_prep_time, 

560 flex_event_duration, 

561 ) = self._get_characteristic_times(at_time_step) 

562 ts = ( 

563 self.baseline_module_config.time_step 

564 / TIME_CONVERSION[self.current_timescale_of_data] 

565 ) 

566 

567 xlim_left = at_time_step 

568 xlim_right = ( 

569 at_time_step 

570 + rel_market_time 

571 + rel_prep_time 

572 + flex_event_duration 

573 + 4 * ts 

574 ) 

575 except KeyError: 

576 # No data of characteristic times available for this time step, 

577 # fall back to default zoom 

578 xlim_left = self.df_simulation.index[0] 

579 xlim_right = ( 

580 self.df_simulation.index[-1] + self.df_baseline.index[-1][-1] 

581 ) 

582 elif zoom_to_prediction_interval: 

583 xlim_left = at_time_step 

584 xlim_right = at_time_step + self.df_baseline.index[-1][-1] 

585 else: 

586 xlim_left = self.df_simulation.index[0] 

587 xlim_right = ( 

588 self.df_simulation.index[-1] + self.df_baseline.index[-1][-1] 

589 ) 

590 

591 fig.update_layout( 

592 yaxis_title=variable, 

593 xaxis_title=f"Time in {self.current_timescale_of_data}", 

594 xaxis_range=[xlim_left, xlim_right], 

595 height=350, 

596 margin=dict(t=20, b=20), 

597 ) 

598 fig.update_xaxes( 

599 dtick=round(self.baseline_module_config.prediction_horizon / 6) 

600 * self.baseline_module_config.time_step 

601 / TIME_CONVERSION[self.current_timescale_of_data] 

602 ) 

603 fig.update_yaxes(tickformat="~r") 

604 return fig 

605 

606 def _create_layout(self) -> list: 

607 """Create the layout for the Dash app. 

608 

609 Returns: 

610 The layout as a list of Dash components. 

611 

612 """ 

613 return [ 

614 html.H1("Results"), 

615 html.H3("Settings"), 

616 html.Div( 

617 children=[ 

618 html.Code(f"{option}: {setting}, ") 

619 for option, setting in self.baseline_module_config.optimization_backend[ 

620 "discretization_options" 

621 ].items() 

622 ] 

623 ), 

624 # Options 

625 html.Div( 

626 children=[ 

627 html.H3("Options"), 

628 html.Div( 

629 children=[ 

630 dcc.Checklist( 

631 id="accepted_characteristic_times", 

632 options=[ 

633 { 

634 "label": "Show characteristic times (accepted)", 

635 "value": True, 

636 } 

637 ], 

638 value=[True], 

639 style={ 

640 "display": "inline-block", 

641 "padding-right": "10px", 

642 }, 

643 ), 

644 dcc.Checklist( 

645 id="current_characteristic_times", 

646 options=[ 

647 { 

648 "label": "Show characteristic times (current)", 

649 "value": True, 

650 } 

651 ], 

652 value=[True], 

653 style={ 

654 "display": "inline-block", 

655 "padding-right": "10px", 

656 }, 

657 ), 

658 ], 

659 ), 

660 html.Div( 

661 children=[ 

662 dcc.Checklist( 

663 id="zoom_to_offer_window", 

664 options=[ 

665 { 

666 "label": "Zoom to flexibility offer window", 

667 "value": False, 

668 } 

669 ], 

670 style={ 

671 "display": "inline-block", 

672 "padding-right": "10px", 

673 }, 

674 ), 

675 dcc.Checklist( 

676 id="zoom_to_prediction_interval", 

677 options=[ 

678 { 

679 "label": "Zoom to mpc prediction interval", 

680 "value": False, 

681 } 

682 ], 

683 style={"display": "inline-block"}, 

684 ), 

685 ], 

686 ), 

687 # Time input 

688 html.Div( 

689 children=[ 

690 html.H3( 

691 children=f"Time:", 

692 style={ 

693 "display": "inline-block", 

694 "padding-right": "10px", 

695 }, 

696 ), 

697 dcc.Input( 

698 id="time_typing", 

699 type="number", 

700 min=0, 

701 max=1, 

702 value=0, # will be updated in the callback 

703 style={"display": "inline-block"}, 

704 ), 

705 dcc.Dropdown( 

706 id="time_unit", 

707 options=get_args(TimeConversionTypes), 

708 value=self.current_timescale_input, 

709 style={ 

710 "display": "inline-block", 

711 "verticalAlign": "middle", 

712 "padding-left": "10px", 

713 "width": "100px", 

714 }, 

715 ), 

716 ], 

717 ), 

718 dcc.Slider( 

719 id="time_slider", 

720 min=0, 

721 max=1, 

722 value=0, # will be updated in the callback 

723 tooltip={"placement": "bottom", "always_visible": True}, 

724 marks=None, 

725 updatemode="drag", 

726 ), 

727 ], 

728 style={ 

729 "width": "88%", 

730 "padding-left": "0%", 

731 "padding-right": "12%", 

732 # Make the options sticky to the top of the page 

733 "position": "sticky", 

734 "top": "0", 

735 "overflow-y": "visible", 

736 "z-index": "100", 

737 "background-color": "white", 

738 }, 

739 ), 

740 # Container for the graphs, will be updated in the callback 

741 html.Div(id="graphs_container_variables", children=[]), 

742 ] 

743 

744 def _register_callbacks(self, app: Dash) -> None: 

745 """Register all callbacks for the Dash app. 

746 

747 Args: 

748 app: The Dash application instance to register callbacks on. 

749 

750 """ 

751 # Callbacks 

752 # Update the time value or the time unit 

753 @app.callback( 

754 Output(component_id="time_slider", component_property="value"), 

755 Output(component_id="time_slider", component_property="min"), 

756 Output(component_id="time_slider", component_property="max"), 

757 Output(component_id="time_slider", component_property="step"), 

758 Output(component_id="time_typing", component_property="value"), 

759 Output(component_id="time_typing", component_property="min"), 

760 Output(component_id="time_typing", component_property="max"), 

761 Output(component_id="time_typing", component_property="step"), 

762 Input(component_id="time_typing", component_property="value"), 

763 Input(component_id="time_slider", component_property="value"), 

764 Input(component_id="time_unit", component_property="value"), 

765 ) 

766 def update_time_index_of_input( 

767 time_typing: float, time_slider: float, time_unit: TimeConversionTypes 

768 ) -> (int, float, float, float, int, float, float, float): 

769 # get trigger id 

770 trigger_id = ctx.triggered[0]["prop_id"].split(".")[0] 

771 

772 # Get the value for the sliders 

773 if trigger_id == "time_slider": 

774 value = time_slider 

775 elif trigger_id == "time_unit": 

776 value = ( 

777 time_typing 

778 * TIME_CONVERSION[self.current_timescale_input] 

779 / TIME_CONVERSION[time_unit] 

780 ) 

781 else: 

782 value = time_typing 

783 

784 # Convert the index to the given time unit if necessary 

785 if trigger_id == "time_unit": 

786 self.convert_timescale_of_dataframe_index(to_timescale=time_unit) 

787 

788 # Get the index for the slider types 

789 times = self.df_baseline.index.get_level_values(0).unique() 

790 minimum = times[0] 

791 maximum = times[-1] 

792 step = times[1] - times[0] 

793 

794 self.current_timescale_input = time_unit 

795 

796 return (value, minimum, maximum, step, value, minimum, maximum, step) 

797 

798 # Update the graphs 

799 @app.callback( 

800 Output( 

801 component_id="graphs_container_variables", component_property="children" 

802 ), 

803 Input(component_id="time_typing", component_property="value"), 

804 Input( 

805 component_id="accepted_characteristic_times", component_property="value" 

806 ), 

807 Input( 

808 component_id="current_characteristic_times", component_property="value" 

809 ), 

810 Input(component_id="zoom_to_offer_window", component_property="value"), 

811 Input( 

812 component_id="zoom_to_prediction_interval", component_property="value" 

813 ), 

814 ) 

815 def update_graph( 

816 at_time_step: float, 

817 show_accepted_characteristic_times: bool, 

818 show_current_characteristic_times: bool, 

819 zoom_to_offer_window: bool, 

820 zoom_to_prediction_interval: bool, 

821 ) -> list[dcc.Graph]: 

822 """Update all graphs based on the options and slider values""" 

823 figs = [] 

824 for variable in self.plotting_variables: 

825 fig = self._create_plot( 

826 variable=variable, 

827 at_time_step=at_time_step, 

828 show_accepted_characteristic_times=show_accepted_characteristic_times, 

829 show_current_characteristic_times=show_current_characteristic_times, 

830 zoom_to_offer_window=zoom_to_offer_window, 

831 zoom_to_prediction_interval=zoom_to_prediction_interval, 

832 ) 

833 figs.append(dcc.Graph(id=f"graph_{variable}", figure=fig)) 

834 return figs 

835 

836 def create_app( 

837 self, custom_bounds: Union[CustomBound, list[CustomBound]] = None 

838 ) -> Dash: 

839 """Create and return the Dash app without running it. 

840 

841 This method sets up the entire app layout and callbacks but doesn't