Coverage for backend/django/Economics/results/services/chart_datasets.py: 88%

1"""Build v1 chart datasets from persisted Economics result rows. 

2 

3Charts are presentation summaries only. This service reads 

4``EconomicsResultLine`` rows produced by the result lifecycle service and 

5materializes compact ``EconomicsChartDataset`` rows for later API/export use. 

6It does not calculate new financial meaning; values, warning references, 

7assumptions, and drill-back row identifiers all come from the result table rows. 

8""" 

9 

10from __future__ import annotations 

11 

12from decimal import Decimal, InvalidOperation 

13from typing import Literal, TypeAlias 

14 

15from django.db import transaction 

16from pydantic import BaseModel, ConfigDict 

17 

18from Economics.results.models import EconomicsChartDataset, EconomicsResultLine, EconomicsResultRun 

19from Economics.results.services.financial_metrics.metric_catalog import ( 

20 FinancialMetricKey, 

21 default_comparison_metric_keys, 

22 financial_metric_spec, 

23 required_financial_metric_spec, 

24) 

25from Economics.scheduling.series import ( 

26 ScheduleTimelineBucket, 

27 ScheduleTimelineStep, 

28 study_schedule_timeline_buckets, 

29 study_schedule_timeline_step_count, 

30 study_schedule_timeline_steps, 

31) 

32 

33from Economics.shared.choices import OperatingLineEconomicEffect, OperatingLineCategory, ResultLineKind 

34 

35 

36CHART_CASH_FLOW_NPV = "cash_flow_npv" 

37CHART_CAPEX_BREAKDOWN = "capex_breakdown" 

38CHART_OPEX_BREAKDOWN = "opex_breakdown" 

39CHART_MANUAL_BASELINE_COMPARISON = "manual_baseline_comparison" 

40CHART_OPERATING_COST_CUMULATIVE = "operating_cost_cumulative" 

41CHART_OPERATING_COST_PROFILE = "operating_cost_profile" 

42MAX_OPERATING_COST_TIMELINE_POINTS = 5000 

43V1_CHART_KEYS = ( 

44 CHART_CASH_FLOW_NPV, 

45 CHART_CAPEX_BREAKDOWN, 

46 CHART_OPEX_BREAKDOWN, 

47 CHART_MANUAL_BASELINE_COMPARISON, 

48 CHART_OPERATING_COST_CUMULATIVE, 

49 CHART_OPERATING_COST_PROFILE, 

50) 

51COMPARISON_METRIC_KEYS = default_comparison_metric_keys() 

52ChartScalar: TypeAlias = str | int | Decimal | bool | None 

53 

54 

55class EconomicsContract(BaseModel): 

56 model_config = ConfigDict(frozen=True) 

57 

58 

59class ChartWarningRef(EconomicsContract): 

60 code: str 

61 severity: str 

62 message: str 

63 source_row_key: str | None = None 

64 

65 

66class ChartSourceRow(EconomicsContract): 

67 id: int 

68 row_key: str 

69 label: str 

70 

71 

72class ChartAssumptionRecord(EconomicsContract): 

73 """Normalized tooltip assumption copied from result-line JSON boundaries.""" 

74 

75 key: str 

76 value: ChartScalar 

77 

78 

79class CashFlowPointMetadata(EconomicsContract): 

80 point_type: Literal["cash_flow"] 

81 year: int | None 

82 present_value: Decimal | None 

83 

84 

85class RankedBreakdownPointMetadata(EconomicsContract): 

86 point_type: Literal["ranked_breakdown"] 

87 rank: int 

88 group: str 

89 

90 

91class ComparisonPointMetadata(EconomicsContract): 

92 point_type: Literal["comparison"] 

93 category: str 

94 series_key: str 

95 

96 

97class OperatingCostTimelinePointMetadata(EconomicsContract): 

98 point_type: Literal["operating_cost_timeline"] 

99 elapsed_hours: Decimal 

100 duration_hours: Decimal 

101 row_index: int | None 

102 series_key: str 

103 cumulative: bool 

104 

105 

106ChartPointMetadata: TypeAlias = ( 

107 CashFlowPointMetadata 

108 | RankedBreakdownPointMetadata 

109 | ComparisonPointMetadata 

110 | OperatingCostTimelinePointMetadata 

111) 

112 

113 

114class ChartDatum(EconomicsContract): 

115 key: str 

116 label: str 

117 value: Decimal | None 

118 unit: str 

119 source_row: ChartSourceRow | None = None 

120 assumptions: tuple[ChartAssumptionRecord, ...] 

121 warning_refs: tuple[ChartWarningRef, ...] 

122 metadata: ChartPointMetadata 

123 

124 

125class ChartSeries(EconomicsContract): 

126 key: str 

127 label: str 

128 unit: str 

129 points: tuple[ChartDatum, ...] 

130 

131 

132class CashFlowRenderingMetadata(EconomicsContract): 

133 chart_family: Literal["cash_flow_npv"] 

134 x_axis: Literal["project_year"] 

135 zero_reference: Decimal 

136 payback_years: Decimal | None 

137 npv: Decimal | None 

138 warning_refs: tuple[ChartWarningRef, ...] 

139 

140 

141class RankedBreakdownRenderingMetadata(EconomicsContract): 

142 chart_family: Literal["ranked_breakdown"] 

143 ranking: Literal["amount_desc"] 

144 warning_refs: tuple[ChartWarningRef, ...] 

145 

146 

147class ComparisonRenderingMetadata(EconomicsContract): 

148 chart_family: Literal["manual_baseline_comparison"] 

149 categories: tuple[str, ...] 

150 series_keys: tuple[str, ...] 

151 warning_refs: tuple[ChartWarningRef, ...] 

152 

153 

154class OperatingCostTimelineRenderingMetadata(EconomicsContract): 

155 chart_family: Literal["operating_cost_timeline"] 

156 x_axis: Literal["operating_hours"] 

157 cumulative: bool 

158 interval_hours: Decimal | None 

159 annual_operating_hours: Decimal | None 

160 schedule_scenario_id: int | None 

161 point_limit: int | None = None 

162 estimated_point_count: int | None = None 

163 coerced_point_count: int | None = None 

164 downsampled: bool = False 

165 message: str = "" 

166 warning_refs: tuple[ChartWarningRef, ...] 

167 

168 

169class ChartDataPayload(EconomicsContract): 

170 chart_key: str 

171 title: str 

172 chart_type: str 

173 series: tuple[ChartSeries, ...] 

174 

175 

176ChartRenderingMetadata: TypeAlias = ( 

177 CashFlowRenderingMetadata 

178 | RankedBreakdownRenderingMetadata 

179 | ComparisonRenderingMetadata 

180 | OperatingCostTimelineRenderingMetadata 

181) 

182 

183 

184class ChartDatasetContract(EconomicsContract): 

185 chart_key: str 

186 title: str 

187 chart_type: str 

188 source_row_keys: tuple[str, ...] 

189 series: tuple[ChartSeries, ...] 

190 rendering_metadata: ChartRenderingMetadata 

191 

192 def chart_data_payload(self) -> ChartDataPayload: 

193 """Return the typed payload that is serialized into ``EconomicsChartDataset.chart_data``.""" 

194 return ChartDataPayload( 

195 chart_key=self.chart_key, 

196 title=self.title, 

197 chart_type=self.chart_type, 

198 series=self.series, 

199 ) 

200 

201 def rendering_metadata_payload(self) -> ChartRenderingMetadata: 

202 """Return typed compact chart configuration for the rendering metadata JSON boundary.""" 

203 return self.rendering_metadata 

204 

205 

206def build_chart_datasets(result_run: EconomicsResultRun) -> tuple[ChartDatasetContract, ...]: 

207 """Return deterministic v1 chart datasets derived from a result run's rows.""" 

208 lines = list( 

209 result_run.lines.select_related("source_capital_line", "source_operating_line").order_by( 

210 "sort_order", 

211 "created_at", 

212 "pk", 

213 ) 

214 ) 

215 line_by_row_key = {line.row_key: line for line in lines} 

216 return ( 

217 _cash_flow_npv_dataset(lines=lines, line_by_row_key=line_by_row_key), 

218 _ranked_breakdown_dataset( 

219 chart_key=CHART_CAPEX_BREAKDOWN, 

220 title="Capital Cost Breakdown", 

221 line_kind=ResultLineKind.CAPITAL, 

222 source_group="capital_lines", 

223 lines=lines, 

224 ), 

225 _ranked_breakdown_dataset( 

226 chart_key=CHART_OPEX_BREAKDOWN, 

227 title="Operating Cost Breakdown", 

228 line_kind=ResultLineKind.OPERATING, 

229 source_group="operating_lines", 

230 lines=lines, 

231 ), 

232 _manual_baseline_comparison_dataset(line_by_row_key=line_by_row_key), 

233 *_operating_cost_timeline_datasets(result_run=result_run, lines=lines), 

234 ) 

235 

236 

237def materialize_chart_datasets(result_run: EconomicsResultRun) -> tuple[ChartDatasetContract, ...]: 

238 """Upsert the required v1 chart datasets for ``result_run`` transactionally.""" 

239 datasets = build_chart_datasets(result_run) 

240 with transaction.atomic(): 

241 for dataset in datasets: 

242 EconomicsChartDataset.objects.update_or_create( 

243 flowsheet_state=result_run.flowsheet_state, 

244 result_run=result_run, 

245 chart_key=dataset.chart_key, 

246 defaults={ 

247 "title": dataset.title, 

248 "chart_type": dataset.chart_type, 

249 "source_row_keys": list(dataset.source_row_keys), 

250 "chart_data": dataset.chart_data_payload().model_dump(mode="json"), 

251 "rendering_metadata": dataset.rendering_metadata_payload().model_dump(mode="json"), 

252 }, 

253 ) 

254 return datasets 

255 

256 

257def _cash_flow_npv_dataset( 

258 *, 

259 lines: list[EconomicsResultLine], 

260 line_by_row_key: dict[str, EconomicsResultLine], 

261) -> ChartDatasetContract: 

262 cash_flow_lines = [line for line in lines if line.kind == ResultLineKind.CASH_FLOW] 

263 annual_points = [] 

264 cumulative_points = [] 

265 for line in cash_flow_lines: 

266 year = _year_from_cash_flow_key(line.row_key) 

267 metadata = CashFlowPointMetadata(point_type="cash_flow", year=year, present_value=line.amount) 

268 annual_points.append( 

269 _datum_from_line( 

270 line, 

271 key=f"{line.row_key}.cash_flow", 

272 value=_decimal_from_payload(line.warning_payload, "cash_flow", fallback=line.amount), 

273 metadata=metadata, 

274 ) 

275 ) 

276 cumulative_points.append( 

277 _datum_from_line( 

278 line, 

279 key=f"{line.row_key}.cumulative_discounted", 

280 value=_decimal_from_payload(line.warning_payload, "cumulative_present_value", fallback=line.amount), 

281 metadata=metadata, 

282 ) 

283 ) 

284 

285 payback_line = line_by_row_key.get(required_financial_metric_spec(FinancialMetricKey.SIMPLE_PAYBACK_YEARS).row_key) 

286 npv_line = line_by_row_key.get(required_financial_metric_spec(FinancialMetricKey.NPV).row_key) 

287 source_row_keys = _source_row_keys(cash_flow_lines + [line for line in (payback_line, npv_line) if line is not None]) 

288 return ChartDatasetContract( 

289 chart_key=CHART_CASH_FLOW_NPV, 

290 title="Cash Flow And NPV", 

291 chart_type="combined_bar_line", 

292 source_row_keys=source_row_keys, 

293 series=( 

294 ChartSeries( 

295 key="annual_net_cash_flow", 

296 label="Annual Net Cash Flow", 

297 unit=_first_unit(cash_flow_lines), 

298 points=tuple(annual_points), 

299 ), 

300 ChartSeries( 

301 key="cumulative_discounted_cash_flow", 

302 label="Cumulative Discounted Cash Flow", 

303 unit=_first_unit(cash_flow_lines), 

304 points=tuple(cumulative_points), 

305 ), 

306 ), 

307 rendering_metadata=CashFlowRenderingMetadata( 

308 chart_family="cash_flow_npv", 

309 x_axis="project_year", 

310 zero_reference=Decimal("0"), 

311 payback_years=payback_line.amount if payback_line and payback_line.amount is not None else None, 

312 npv=npv_line.amount if npv_line and npv_line.amount is not None else None, 

313 warning_refs=_run_warning_refs(lines), 

314 ), 

315 ) 

316 

317 

318def _ranked_breakdown_dataset( 

319 *, 

320 chart_key: str, 

321 title: str, 

322 line_kind: str, 

323 source_group: str, 

324 lines: list[EconomicsResultLine], 

325) -> ChartDatasetContract: 

326 breakdown_lines = [ 

327 line 

328 for line in lines 

329 if line.kind == line_kind and line.group == source_group and line.amount is not None 

330 ] 

331 if line_kind == ResultLineKind.OPERATING and source_group == "operating_lines": 

332 breakdown_lines = [ 

333 line 

334 for line in breakdown_lines 

335 if not _operating_line_is_revenue(line) 

336 ] 

337 breakdown_lines.sort(key=lambda line: (-abs(line.amount or Decimal("0")), line.label, line.row_key)) 

338 points = tuple( 

339 _datum_from_line( 

340 line, 

341 key=line.row_key, 

342 value=line.amount, 

343 metadata=RankedBreakdownPointMetadata(point_type="ranked_breakdown", rank=index, group=source_group), 

344 ) 

345 for index, line in enumerate(breakdown_lines, start=1) 

346 ) 

347 return ChartDatasetContract( 

348 chart_key=chart_key, 

349 title=title, 

350 chart_type="ranked_bar", 

351 source_row_keys=_source_row_keys(breakdown_lines), 

352 series=( 

353 ChartSeries( 

354 key="amount", 

355 label=title, 

356 unit=_first_unit(breakdown_lines), 

357 points=points, 

358 ), 

359 ), 

360 rendering_metadata=RankedBreakdownRenderingMetadata( 

361 chart_family="ranked_breakdown", 

362 ranking="amount_desc", 

363 warning_refs=_run_warning_refs(breakdown_lines), 

364 ), 

365 ) 

366 

367 

368def _manual_baseline_comparison_dataset( 

369 *, 

370 line_by_row_key: dict[str, EconomicsResultLine], 

371) -> ChartDatasetContract: 

372 points: list[ChartDatum] = [] 

373 capex_line = line_by_row_key.get(required_financial_metric_spec(FinancialMetricKey.CAPEX).row_key) 

374 incremental_capex_line = line_by_row_key.get(required_financial_metric_spec(FinancialMetricKey.INCREMENTAL_CAPEX).row_key) 

375 if capex_line is not None: 

376 points.append(_comparison_datum(capex_line, series_key="target", category="capex", value=capex_line.amount)) 

377 baseline_capex = _baseline_capex(capex_line=capex_line, incremental_capex_line=incremental_capex_line) 

378 if incremental_capex_line is not None and baseline_capex is not None: 

379 points.append( 

380 _comparison_datum( 

381 incremental_capex_line, 

382 series_key="baseline", 

383 category="capex", 

384 value=baseline_capex, 

385 label="Baseline Capex", 

386 ) 

387 ) 

388 

389 opex_line = line_by_row_key.get(required_financial_metric_spec(FinancialMetricKey.ANNUAL_OPEX).row_key) 

390 annual_savings_line = line_by_row_key.get(required_financial_metric_spec(FinancialMetricKey.ANNUAL_SAVINGS).row_key) 

391 if opex_line is not None: 

392 points.append(_comparison_datum(opex_line, series_key="target", category="annual_opex", value=opex_line.amount)) 

393 baseline_opex = _decimal_from_assumptions(annual_savings_line, "baseline_annual_opex") 

394 if annual_savings_line is not None and baseline_opex is not None: 

395 points.append( 

396 _comparison_datum( 

397 annual_savings_line, 

398 series_key="baseline", 

399 category="annual_opex", 

400 value=baseline_opex, 

401 label="Baseline Annual Opex", 

402 ) 

403 ) 

404 

405 for metric_key in COMPARISON_METRIC_KEYS: 

406 if metric_key in {"capex", "annual_opex"}: 

407 continue 

408 spec = financial_metric_spec(metric_key) 

409 line = line_by_row_key.get(spec.row_key) if spec is not None else None 

410 if line is not None: 

411 points.append(_comparison_datum(line, series_key="result", category=metric_key, value=line.amount)) 

412 

413 source_rows = [point.source_row.row_key for point in points if point.source_row is not None] 

414 return ChartDatasetContract( 

415 chart_key=CHART_MANUAL_BASELINE_COMPARISON, 

416 title="Manual Baseline Comparison", 

417 chart_type="grouped_bar", 

418 source_row_keys=tuple(dict.fromkeys(source_rows)), 

419 series=( 

420 ChartSeries( 

421 key="comparison_values", 

422 label="Comparison Values", 

423 unit="mixed", 

424 points=tuple(points), 

425 ), 

426 ), 

427 rendering_metadata=ComparisonRenderingMetadata( 

428 chart_family="manual_baseline_comparison", 

429 categories=COMPARISON_METRIC_KEYS, 

430 series_keys=("target", "baseline", "result"), 

431 warning_refs=_run_warning_refs([line for line in line_by_row_key.values() if line.row_key in source_rows]), 

432 ), 

433 ) 

434 

435 

436def _operating_cost_timeline_datasets( 

437 *, 

438 result_run: EconomicsResultRun, 

439 lines: list[EconomicsResultLine], 

440) -> tuple[ChartDatasetContract, ChartDatasetContract]: 

441 """Build cumulative and per-step operating-cost timeline datasets from schedule rows.""" 

442 operating_lines = [ 

443 line 

444 for line in lines 

445 if line.kind == ResultLineKind.OPERATING 

446 and line.group == "operating_lines" 

447 and line.amount is not None 

448 and not _operating_line_is_revenue(line) 

449 ] 

450 study = result_run.study 

451 estimated_timestep_count = study_schedule_timeline_step_count(study=study) 

452 estimated_chart_point_count = _operating_timeline_chart_point_count( 

453 timestep_count=estimated_timestep_count, 

454 operating_line_count=len(operating_lines), 

455 ) 

456 downsampled = operating_lines and estimated_chart_point_count > MAX_OPERATING_COST_TIMELINE_POINTS 

457 if downsampled: 

458 bucket_count = _operating_timeline_bucket_count( 

459 operating_line_count=len(operating_lines), 

460 point_limit=MAX_OPERATING_COST_TIMELINE_POINTS, 

461 ) 

462 timeline_buckets = study_schedule_timeline_buckets(study=study, bucket_count=bucket_count) 

463 timeline_steps = _timeline_steps_from_buckets(timeline_buckets) 

464 else: 

465 timeline_buckets = () 

466 timeline_steps = study_schedule_timeline_steps(study=study) 

467 if not operating_lines or not timeline_steps: 

468 return ( 

469 _empty_operating_timeline_dataset( 

470 chart_key=CHART_OPERATING_COST_CUMULATIVE, 

471 title="Cumulative Operating Cost", 

472 cumulative=True, 

473 result_run=result_run, 

474 estimated_point_count=estimated_chart_point_count, 

475 ), 

476 _empty_operating_timeline_dataset( 

477 chart_key=CHART_OPERATING_COST_PROFILE, 

478 title="Operating Cost Profile", 

479 cumulative=False, 

480 result_run=result_run, 

481 estimated_point_count=estimated_chart_point_count, 

482 ), 

483 ) 

484 

485 if timeline_buckets: 

486 total_annual_hours = _active_timeline_bucket_hours(timeline_buckets) 

487 line_amounts_by_step = { 

488 line.pk: tuple( 

489 _operating_line_bucket_amount( 

490 line=line, 

491 bucket=bucket, 

492 total_annual_hours=total_annual_hours, 

493 ) 

494 for bucket in timeline_buckets 

495 ) 

496 for line in operating_lines 

497 } 

498 coerced_point_count = _operating_timeline_chart_point_count( 

499 timestep_count=len(timeline_steps), 

500 operating_line_count=len(operating_lines), 

501 ) 

502 else: 

503 total_annual_hours = sum((step.duration_hours for step in timeline_steps), Decimal("0")) 

504 line_amounts_by_step = { 

505 line.pk: tuple( 

506 _operating_line_step_amount( 

507 line=line, 

508 step=step, 

509 total_annual_hours=total_annual_hours, 

510 ) 

511 for step in timeline_steps 

512 ) 

513 for line in operating_lines 

514 } 

515 coerced_point_count = None 

516 currency = result_run.result_currency 

517 cumulative_series = [ 

518 ChartSeries( 

519 key="total", 

520 label="Total operating cost", 

521 unit=currency, 

522 points=_operating_total_timeline_points( 

523 timeline_steps=timeline_steps, 

524 line_amounts_by_step=line_amounts_by_step, 

525 unit=currency, 

526 cumulative=True, 

527 ), 

528 ) 

529 ] 

530 cumulative_series.extend( 

531 _operating_line_timeline_series( 

532 line=line, 

533 timeline_steps=timeline_steps, 

534 amounts=line_amounts_by_step[line.pk], 

535 unit=currency, 

536 cumulative=True, 

537 ) 

538 for line in operating_lines 

539 ) 

540 profile_series = ( 

541 ChartSeries( 

542 key="total", 

543 label="Total operating cost", 

544 unit=currency, 

545 points=_operating_total_timeline_points( 

546 timeline_steps=timeline_steps, 

547 line_amounts_by_step=line_amounts_by_step, 

548 unit=currency, 

549 cumulative=False, 

550 ), 

551 ), 

552 ) 

553 source_row_keys = _source_row_keys(operating_lines) 

554 warning_refs = _run_warning_refs(operating_lines) 

555 interval_hours = _timeline_interval_hours(timeline_steps) 

556 annual_operating_hours = total_annual_hours if total_annual_hours > 0 else None 

557 return ( 

558 ChartDatasetContract( 

559 chart_key=CHART_OPERATING_COST_CUMULATIVE, 

560 title="Cumulative Operating Cost", 

561 chart_type="multi_line", 

562 source_row_keys=source_row_keys, 

563 series=tuple(cumulative_series), 

564 rendering_metadata=OperatingCostTimelineRenderingMetadata( 

565 chart_family="operating_cost_timeline", 

566 x_axis="operating_hours", 

567 cumulative=True, 

568 interval_hours=interval_hours, 

569 annual_operating_hours=annual_operating_hours, 

570 schedule_scenario_id=study.schedule_scenario_id, 

571 point_limit=MAX_OPERATING_COST_TIMELINE_POINTS, 

572 estimated_point_count=estimated_chart_point_count, 

573 coerced_point_count=coerced_point_count, 

574 downsampled=downsampled, 

575 message=_operating_timeline_downsampled_message(downsampled), 

576 warning_refs=warning_refs, 

577 ), 

578 ), 

579 ChartDatasetContract( 

580 chart_key=CHART_OPERATING_COST_PROFILE, 

581 title="Operating Cost Profile", 

582 chart_type="line", 

583 source_row_keys=source_row_keys, 

584 series=profile_series, 

585 rendering_metadata=OperatingCostTimelineRenderingMetadata( 

586 chart_family="operating_cost_timeline", 

587 x_axis="operating_hours", 

588 cumulative=False, 

589 interval_hours=interval_hours, 

590 annual_operating_hours=annual_operating_hours, 

591 schedule_scenario_id=study.schedule_scenario_id, 

592 point_limit=MAX_OPERATING_COST_TIMELINE_POINTS, 

593 estimated_point_count=estimated_chart_point_count, 

594 coerced_point_count=coerced_point_count, 

595 downsampled=downsampled, 

596 message=_operating_timeline_downsampled_message(downsampled), 

597 warning_refs=warning_refs, 

598 ), 

599 ), 

600 ) 

601 

602 

603def _operating_timeline_chart_point_count(*, timestep_count: int, operating_line_count: int) -> int: 

604 """Estimate combined point count for cumulative and profile timeline charts.""" 

605 if timestep_count <= 0: 

606 return 0 

607 cumulative_point_count = (timestep_count + 1) * (operating_line_count + 1) 

608 profile_point_count = timestep_count 

609 return cumulative_point_count + profile_point_count 

610 

611 

612def _operating_timeline_bucket_count(*, operating_line_count: int, point_limit: int) -> int: 

613 """Return the largest timeline bucket count that fits both timeline charts.""" 

614 

615 fixed_cumulative_origin_points = operating_line_count + 1 

616 variable_points_per_bucket = operating_line_count + 2 

617 if point_limit <= fixed_cumulative_origin_points or variable_points_per_bucket <= 0: 617 ↛ 618line 617 didn't jump to line 618 because the condition on line 617 was never true

618 return 1 

619 return max(1, (point_limit - fixed_cumulative_origin_points) // variable_points_per_bucket) 

620 

621 

622def _timeline_steps_from_buckets( 

623 timeline_buckets: tuple[ScheduleTimelineBucket, ...], 

624) -> tuple[ScheduleTimelineStep, ...]: 

625 """Adapt bounded schedule buckets to the existing chart point builders.""" 

626 

627 return tuple( 

628 ScheduleTimelineStep( 

629 index=bucket.index, 

630 row_index=bucket.row_index, 

631 elapsed_hours=bucket.elapsed_hours, 

632 duration_hours=bucket.duration_hours, 

633 ) 

634 for bucket in timeline_buckets 

635 ) 

636 

637 

638def _active_timeline_bucket_hours(timeline_buckets: tuple[ScheduleTimelineBucket, ...]) -> Decimal: 

639 """Return active operating hours represented by grouped timeline buckets.""" 

640 

641 return sum( 

642 ( 

643 sum((segment.duration_hours for segment in bucket.segments), Decimal("0")) 

644 for bucket in timeline_buckets 

645 ), 

646 Decimal("0"), 

647 ) 

648 

649 

650def _operating_timeline_downsampled_message(downsampled: bool) -> str: 

651 if not downsampled: 

652 return "" 

653 return "Production schedule timeline grouped to the chart point limit." 

654 

655 

656def _empty_operating_timeline_dataset( 

657 *, 

658 chart_key: str, 

659 title: str, 

660 cumulative: bool, 

661 result_run: EconomicsResultRun, 

662 estimated_point_count: int | None = None, 

663) -> ChartDatasetContract: 

664 """Return a valid empty timeline dataset when no schedule-backed chart can be drawn.""" 

665 return ChartDatasetContract( 

666 chart_key=chart_key, 

667 title=title, 

668 chart_type="multi_line" if cumulative else "line", 

669 source_row_keys=(), 

670 series=(), 

671 rendering_metadata=OperatingCostTimelineRenderingMetadata( 

672 chart_family="operating_cost_timeline", 

673 x_axis="operating_hours", 

674 cumulative=cumulative, 

675 interval_hours=None, 

676 annual_operating_hours=None, 

677 schedule_scenario_id=result_run.study.schedule_scenario_id, 

678 point_limit=MAX_OPERATING_COST_TIMELINE_POINTS, 

679 estimated_point_count=estimated_point_count, 

680 message="", 

681 warning_refs=(), 

682 ), 

683 ) 

684 

685 

686def _operating_line_step_amount( 

687 *, 

688 line: EconomicsResultLine, 

689 step: ScheduleTimelineStep, 

690 total_annual_hours: Decimal, 

691) -> Decimal: 

692 """Allocate one annual operating-line amount into a single schedule step.""" 

693 schedule_contribution = _schedule_contribution_for_row(line, row_index=step.row_index) 

694 if schedule_contribution is not None: 

695 annual_hours = schedule_contribution["annual_hours"] 

696 if annual_hours <= 0: 696 ↛ 697line 696 didn't jump to line 697 because the condition on line 696 was never true

697 return Decimal("0") 

698 return schedule_contribution["amount"] / annual_hours * step.duration_hours 

699 if total_annual_hours <= 0: 699 ↛ 700line 699 didn't jump to line 700 because the condition on line 699 was never true

700 return Decimal("0") 

701 return (line.amount or Decimal("0")) / total_annual_hours * step.duration_hours 

702 

703 

704def _operating_line_bucket_amount( 

705 *, 

706 line: EconomicsResultLine, 

707 bucket: ScheduleTimelineBucket, 

708 total_annual_hours: Decimal, 

709) -> Decimal: 

710 """Allocate an operating-line amount into one grouped schedule bucket.""" 

711 

712 bucket_amount = Decimal("0") 

713 for segment in bucket.segments: 

714 schedule_contribution = _schedule_contribution_for_row(line, row_index=segment.row_index) 

715 bucket_amount += _operating_line_segment_amount( 

716 line=line, 

717 schedule_contribution=schedule_contribution, 

718 duration_hours=segment.duration_hours, 

719 total_annual_hours=total_annual_hours, 

720 ) 

721 return bucket_amount 

722 

723 

724def _operating_line_segment_amount( 

725 *, 

726 line: EconomicsResultLine, 

727 schedule_contribution: dict[str, Decimal] | None, 

728 duration_hours: Decimal, 

729 total_annual_hours: Decimal, 

730) -> Decimal: 

731 """Allocate one source-row duration using the same fallback as ungrouped steps.""" 

732 

733 if schedule_contribution is not None: 733 ↛ 734line 733 didn't jump to line 734 because the condition on line 733 was never true

734 annual_hours = schedule_contribution["annual_hours"] 

735 if annual_hours <= 0: 

736 return Decimal("0") 

737 return schedule_contribution["amount"] / annual_hours * duration_hours 

738 if total_annual_hours <= 0: 738 ↛ 739line 738 didn't jump to line 739 because the condition on line 738 was never true

739 return Decimal("0") 

740 return (line.amount or Decimal("0")) / total_annual_hours * duration_hours 

741 

742 

743def _timeline_interval_hours(timeline_steps: tuple[ScheduleTimelineStep, ...]) -> Decimal | None: 

744 """Return a single interval only when the annualized timeline is uniform.""" 

745 

746 if not timeline_steps: 746 ↛ 747line 746 didn't jump to line 747 because the condition on line 746 was never true

747 return None 

748 first_interval = timeline_steps[0].duration_hours 

749 if all(step.duration_hours == first_interval for step in timeline_steps): 

750 return first_interval 

751 return None 

752 

753 

754def _schedule_contribution_for_row(line: EconomicsResultLine, *, row_index: int) -> dict[str, Decimal] | None: 

755 """Read per-row schedule contribution details stored on operating result lines.""" 

756 payload = line.warning_payload if isinstance(line.warning_payload, dict) else {} 

757 schedule = payload.get("schedule") 

758 if not isinstance(schedule, dict): 

759 return None 

760 contributions = schedule.get("contributions") 

761 if not isinstance(contributions, list): 761 ↛ 762line 761 didn't jump to line 762 because the condition on line 761 was never true

762 return None 

763 for contribution in contributions: 763 ↛ 771line 763 didn't jump to line 771 because the loop on line 763 didn't complete

764 if not isinstance(contribution, dict) or contribution.get("row_index") != row_index: 

765 continue 

766 amount = _to_decimal(contribution.get("amount")) 

767 annual_hours = _to_decimal(contribution.get("annual_hours")) 

768 if amount is None or annual_hours is None: 768 ↛ 769line 768 didn't jump to line 769 because the condition on line 768 was never true

769 return None 

770 return {"amount": amount, "annual_hours": annual_hours} 

771 return None 

772 

773 

774def _operating_total_timeline_points( 

775 *, 

776 timeline_steps: tuple[ScheduleTimelineStep, ...], 

777 line_amounts_by_step: dict[int, tuple[Decimal, ...]], 

778 unit: str, 

779 cumulative: bool, 

780) -> tuple[ChartDatum, ...]: 

781 """Create total operating-cost points by summing all line amounts for each timeline step.""" 

782 points = [] 

783 running_total = Decimal("0") 

784 if cumulative: 

785 points.append( 

786 _operating_timeline_datum( 

787 key="total.0", 

788 label="0 h", 

789 value=Decimal("0"), 

790 unit=unit, 

791 series_key="total", 

792 elapsed_hours=Decimal("0"), 

793 duration_hours=Decimal("0"), 

794 row_index=None, 

795 cumulative=True, 

796 ) 

797 ) 

798 for step in timeline_steps: 

799 step_amount = sum((amounts[step.index] for amounts in line_amounts_by_step.values()), Decimal("0")) 

800 if cumulative: 

801 running_total += step_amount 

802 value = running_total 

803 else: 

804 value = step_amount 

805 elapsed_hours = step.elapsed_hours + step.duration_hours 

806 points.append( 

807 _operating_timeline_datum( 

808 key=f"total.{step.index + 1}", 

809 label=f"{elapsed_hours} h", 

810 value=value, 

811 unit=unit, 

812 series_key="total", 

813 elapsed_hours=elapsed_hours, 

814 duration_hours=step.duration_hours, 

815 row_index=step.row_index, 

816 cumulative=cumulative, 

817 ) 

818 ) 

819 return tuple(points) 

820 

821 

822def _operating_line_timeline_series( 

823 *, 

824 line: EconomicsResultLine, 

825 timeline_steps: tuple[ScheduleTimelineStep, ...], 

826 amounts: tuple[Decimal, ...], 

827 unit: str, 

828 cumulative: bool, 

829) -> ChartSeries: 

830 """Create the per-line cumulative series used by the operating-cost timeline chart.""" 

831 points = [] 

832 running_total = Decimal("0") 

833 if cumulative: 833 ↛ 849line 833 didn't jump to line 849 because the condition on line 833 was always true

834 points.append( 

835 _operating_timeline_datum( 

836 key=f"line.{line.pk}.0", 

837 label="0 h", 

838 value=Decimal("0"), 

839 unit=unit, 

840 series_key=f"line_{line.pk}", 

841 elapsed_hours=Decimal("0"), 

842 duration_hours=Decimal("0"), 

843 row_index=None, 

844 cumulative=True, 

845 source_row=ChartSourceRow(id=line.pk, row_key=line.row_key, label=line.label),