Coverage for backend/django/Economics/results/services/lifecycle/fingerprints.py: 95%

1"""Dependency fingerprint contracts for Economics presentation result runs. 

2 

3Fingerprints are the audit contract that decides whether a stored result run is 

4current, reusable, or stale. This module should only describe source inputs and 

5persist dependency rows; it must not calculate financial metrics or write result 

6lines. 

7""" 

8 

9from __future__ import annotations 

10 

11import hashlib 

12import json 

13import uuid 

14from typing import Any 

15 

16from django.core.cache import cache 

17from django.db import models, transaction 

18from django.contrib.postgres.aggregates import StringAgg 

19from django.db.models.functions import Cast, JSONObject, MD5 

20from pydantic import field_serializer 

21 

22from Economics.costing.models import ( 

23 

24 CapitalCostLine, 

25 

26 CostCurve, 

27 

28 CostDriver, 

29 

30 CostableItem, 

31 

32 EquipmentMapping, 

33 

34 OperatingCostLine, 

35 

36) 

37 

38from Economics.reference_data.models import CostIndexSeries, CostIndexValue 

39 

40from Economics.results.models import EconomicsResultDependency, EconomicsResultLine, EconomicsResultRun 

41 

42from Economics.settings_profiles.models import EconomicsSettingsProfile 

43from Economics.settings_profiles.services.settings_profiles import get_settings_profile 

44 

45from Economics.studies.models import EconomicsStudy 

46from Economics.studies.services.baseline_access import resolve_baseline_study 

47 

48from Economics.shared.choices import EconomicsBaselineMode, EconomicsScheduleMode, ResultDependencyType, ResultRunStatus 

49from Economics.formulas.engine.core import FORMULA_AUDIT_SCHEMA_VERSION 

50from Economics.results.services.lifecycle.common import EconomicsContract, get_assumptions, version 

51from Economics.results.services.financial_metrics.metric_catalog import FinancialMetricKey, required_financial_metric_spec 

52from Economics.scheduling.composite.plan_services import saved_schedule_plan_payload 

53from Economics.scheduling.services import schedule_preview 

54from Economics.shared.payloads import json_ready 

55from core.auxiliary.models import PropertyInfo 

56from core.auxiliary.models.Scenario import Scenario 

57from core.auxiliary.models.DataCell import DataCell 

58from core.auxiliary.models.DataColumn import DataColumn 

59from core.auxiliary.models.DataRow import DataRow 

60from core.auxiliary.models.Solution import Solution 

61 

62 

63FINGERPRINT_ALGORITHM = "sha256" 

64FINGERPRINT_PREFIX = "sha256:" 

65FORMULA_REGISTRY_VERSION = "2026-06-15.depreciation-tax-model" 

66FORMULA_SEMANTIC_FINGERPRINTS = { 

67 "cost_curve": "expression_text parsed by constrained AST-to-SymPy evaluator", 

68 "production_schedule": "selected schedule mode, scenario timing, schedule inputs, and solved row outputs", 

69 "generated_capital_line": "cost_curve * capital_index_factor * optional_lang_factor * contingency_factor", 

70 "generated_unit_capex_subtotal": "sum included generated unit-operation capital line formulas", 

71 "custom_capital_line": "fixed literal or custom_capex_percentage_basis * basis_percent / 100", 

72 "custom_capex_percentage_basis": "generated_unit_capex_subtotal only", 

73 "custom_capital_total": "sum included custom capital line formulas", 

74 "peak_demand_capacity": "sum included capital line peak-demand capacity in kW", 

75 "electrical_upgrade_capex": "peak_demand_capacity * electrical_upgrade_rate_amount", 

76 "operating_line": "basis_quantity * annualization_factor * rate_amount", 

77 "annual_operating_expense": "sum included non-revenue operating-line formulas", 

78 "annual_operating_revenue": "sum included output-revenue operating-line formulas", 

79 "default_rate_derived_steam": "fuel_price_nzd_per_gj * steam_energy_gj_per_t / (boiler_efficiency_percent / 100)", 

80 "process_energy_contribution": "direct annualized energy quantity or operating-line annual basis converted to MWh", 

81 "annual_profit": "target_annual_revenue - target_annual_opex", 

82 "annual_savings": "baseline_annual_opex - target_annual_opex", 

83 "annual_net_benefit": "annual_savings + target_annual_revenue", 

84 "annual_depreciation": "sum included depreciable capital-line bases less residual value over straight-line equipment life", 

85 "depreciation_tax_shield": "annual_depreciation * tax_rate", 

86 "after_tax_annual_cash_flow": "annual_net_benefit * (1 - tax_rate) + depreciation_tax_shield", 

87 "incremental_capex": "target_capex - baseline_capex", 

88 "roi_percent": "((cash_flow_basis * lifetime + residual_value - incremental_capex) / incremental_capex) * 100", 

89 "cash_flow_rows": "year 0 negative incremental capex, operating years after-tax annual cash-flow basis, final year residual uplift", 

90} 

91 

92 

93class DependencyFingerprint(EconomicsContract): 

94 """Deterministic fingerprint plus source-row links for one lifecycle input.""" 

95 dependency_type: str 

96 dependency_key: str 

97 fingerprint_value: str 

98 fingerprint_basis: str 

99 source_label: str = "" 

100 source_row_key: str = "" 

101 source_version: str = "" 

102 source_settings_profile: EconomicsSettingsProfile | None = None 

103 source_costable_item: CostableItem | None = None 

104 source_cost_curve: CostCurve | None = None 

105 source_capital_line: CapitalCostLine | None = None 

106 source_operating_line: OperatingCostLine | None = None 

107 source_index_series: CostIndexSeries | None = None 

108 source_index_value: CostIndexValue | None = None 

109 source_scenario: Any | None = None 

110 source_property_info: Any | None = None 

111 

112 @field_serializer( 

113 "source_settings_profile", 

114 "source_costable_item", 

115 "source_cost_curve", 

116 "source_capital_line", 

117 "source_operating_line", 

118 "source_index_series", 

119 "source_index_value", 

120 "source_scenario", 

121 "source_property_info", 

122 when_used="json", 

123 ) 

124 def serialize_source_model(self, value): 

125 return value.pk if isinstance(value, models.Model) else value 

126 

127 @property 

128 def identity(self) -> tuple[str, str]: 

129 return (self.dependency_type, self.dependency_key) 

130 

131 

132def build_dependency_fingerprints( 

133 study: EconomicsStudy, 

134 *, 

135 persisted_schedule_fingerprint: str = "", 

136 schedule_version: str | None = None, 

137) -> list[DependencyFingerprint]: 

138 """Collect source-readable fingerprints for all v1 presentation result inputs. 

139 

140 Generated capital lines must be synchronized by the orchestrator before this 

141 runs, so persisted generated rows participate in the same audit contract as 

142 manually entered rows. 

143 """ 

144 fingerprints: list[DependencyFingerprint] = [] 

145 fingerprints.extend(_formula_fingerprints()) 

146 fingerprints.extend(_assumption_fingerprints(study)) 

147 fingerprints.extend(_baseline_fingerprints(study)) 

148 fingerprints.extend( 

149 _schedule_fingerprints( 

150 study, 

151 persisted_fingerprint_value=persisted_schedule_fingerprint, 

152 source_version=schedule_version, 

153 ) 

154 ) 

155 fingerprints.extend(_costable_item_fingerprints(study)) 

156 fingerprints.extend(_cost_driver_fingerprints(study)) 

157 fingerprints.extend(_equipment_mapping_fingerprints(study)) 

158 fingerprints.extend(_capital_line_fingerprints(study)) 

159 fingerprints.extend(_operating_line_fingerprints(study)) 

160 fingerprints.extend(_cost_curve_fingerprints(study)) 

161 fingerprints.extend(_index_data_fingerprints(study)) 

162 return sorted(fingerprints, key=lambda fingerprint: fingerprint.identity) 

163 

164 

165def _formula_fingerprints() -> list[DependencyFingerprint]: 

166 """Fingerprint formula semantics that are not represented by database rows.""" 

167 return [ 

168 _dependency( 

169 dependency_type=ResultDependencyType.ASSUMPTIONS, 

170 dependency_key="formula:registry", 

171 payload={ 

172 "formula_registry_version": FORMULA_REGISTRY_VERSION, 

173 "formula_audit_schema_version": FORMULA_AUDIT_SCHEMA_VERSION, 

174 "semantic_fingerprints": FORMULA_SEMANTIC_FINGERPRINTS, 

175 }, 

176 fingerprint_basis="economics_formula_registry.semantic_fingerprints", 

177 source_label="Economics formula registry", 

178 source_row_key="formula:registry", 

179 source_version=FORMULA_REGISTRY_VERSION, 

180 ) 

181 ] 

182 

183 

184def _assumption_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]: 

185 """Fingerprint study assumptions or an explicit missing-assumptions marker.""" 

186 assumptions = get_assumptions(study) 

187 if assumptions is None: 

188 return [ 

189 _dependency( 

190 dependency_type=ResultDependencyType.ASSUMPTIONS, 

191 dependency_key=f"assumptions:missing:{study.pk}", 

192 payload={"study_id": study.pk, "assumptions": None}, 

193 fingerprint_basis="economics_assumptions.missing", 

194 source_label="Missing economics assumptions", 

195 source_row_key=f"study:{study.pk}:assumptions", 

196 ) 

197 ] 

198 return [ 

199 _dependency( 

200 dependency_type=ResultDependencyType.ASSUMPTIONS, 

201 dependency_key=f"assumptions:{assumptions.pk}", 

202 payload={ 

203 "id": assumptions.pk, 

204 "currency": assumptions.currency, 

205 "location": assumptions.location, 

206 "basis_date": assumptions.basis_date, 

207 "discount_rate_percent": assumptions.discount_rate_percent, 

208 "project_lifetime_years": assumptions.project_lifetime_years, 

209 "inflation_method": assumptions.inflation_method, 

210 "annual_operating_hours": assumptions.annual_operating_hours, 

211 "tax_rate_percent": assumptions.tax_rate_percent, 

212 "depreciation_enabled": assumptions.depreciation_enabled, 

213 "default_depreciation_life_years": assumptions.default_depreciation_life_years, 

214 "default_depreciation_salvage_percent": assumptions.default_depreciation_salvage_percent, 

215 "contingency_percent": assumptions.contingency_percent, 

216 "electrical_upgrade_rate_amount": assumptions.electrical_upgrade_rate_amount, 

217 "electrical_upgrade_rate_unit": assumptions.electrical_upgrade_rate_unit, 

218 "default_lang_factor": assumptions.default_lang_factor, 

219 "capital_index_series_id": assumptions.capital_index_series_id, 

220 "operating_index_series_id": assumptions.operating_index_series_id, 

221 "default_rate_overrides": assumptions.default_rate_overrides, 

222 "notes": assumptions.notes, 

223 "updated_at": assumptions.updated_at, 

224 }, 

225 fingerprint_basis="economics_assumptions.fields", 

226 source_label=f"Assumptions for {study.name}", 

227 source_row_key=f"assumptions:{assumptions.pk}", 

228 source_version=version(assumptions.updated_at), 

229 source_settings_profile=assumptions, 

230 ) 

231 ] 

232 

233 

234def _baseline_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]: 

235 """Fingerprint the selected baseline contract for the target study.""" 

236 if study.baseline_mode == EconomicsBaselineMode.STUDY: 

237 return [_study_baseline_fingerprint(study)] 

238 

239 baseline = get_assumptions(study) 

240 if baseline is None: 

241 return [ 

242 _dependency( 

243 dependency_type=ResultDependencyType.BASELINE, 

244 dependency_key=f"baseline:missing:{study.pk}", 

245 payload={"study_id": study.pk, "mode": EconomicsBaselineMode.MANUAL, "baseline": None}, 

246 fingerprint_basis="economics_baseline.missing", 

247 source_label="Missing economics baseline", 

248 source_row_key=f"study:{study.pk}:baseline", 

249 ) 

250 ] 

251 return [ 

252 _dependency( 

253 dependency_type=ResultDependencyType.BASELINE, 

254 dependency_key=f"baseline:{baseline.pk}", 

255 payload=_baseline_payload(study=study, baseline=baseline), 

256 fingerprint_basis="economics_baseline.manual_fields_and_inherited_study_assumptions", 

257 source_label=f"Baseline for {study.name}", 

258 source_row_key=f"baseline:{baseline.pk}", 

259 source_version=version(baseline.updated_at), 

260 source_settings_profile=baseline, 

261 ) 

262 ] 

263 

264 

265def _baseline_payload(*, study: EconomicsStudy, baseline: EconomicsSettingsProfile) -> dict[str, Any]: 

266 """Include manual baseline fields and inherited assumptions that affect calculations.""" 

267 return { 

268 "id": baseline.pk, 

269 "mode": EconomicsBaselineMode.MANUAL, 

270 "manual_capex": baseline.manual_capex, 

271 "manual_annual_opex": baseline.manual_annual_opex, 

272 "annual_heat_basis_mode": baseline.annual_heat_basis_mode, 

273 "manual_annual_heat_basis": baseline.manual_annual_heat_basis, 

274 "manual_annual_heat_basis_unit": baseline.manual_annual_heat_basis_unit, 

275 "average_power_input": baseline.average_power_input, 

276 "average_power_unit": baseline.average_power_unit, 

277 "residual_value": baseline.residual_value, 

278 "notes": baseline.baseline_notes, 

279 "inherited_study_assumptions": _manual_baseline_inherited_assumptions(study), 

280 "updated_at": baseline.updated_at, 

281 } 

282 

283 

284def _study_baseline_fingerprint(study: EconomicsStudy) -> DependencyFingerprint: 

285 """Fingerprint the selected baseline study result values used by study mode.""" 

286 baseline_study = resolve_baseline_study(study) 

287 run = None if baseline_study is None else _current_baseline_result_run(baseline_study) 

288 payload = { 

289 "study_id": study.pk, 

290 "mode": EconomicsBaselineMode.STUDY, 

291 "baseline_study_lineage_id": ( 

292 str(study.baseline_study_lineage_id) 

293 if study.baseline_study_lineage_id is not None 

294 else None 

295 ), 

296 "baseline_flowsheet_id": study.baseline_flowsheet_id, 

297 "baseline_result_run_id": run.pk if run is not None else None, 

298 "baseline_result_completed_at": run.completed_at if run is not None else None, 

299 "baseline_result_classification": _baseline_result_run_classification(run) if run is not None else None, 

300 "baseline_result_lines": _baseline_result_line_payload(run) if run is not None else {}, 

301 } 

302 return _dependency( 

303 dependency_type=ResultDependencyType.BASELINE, 

304 dependency_key=f"baseline:study:{study.pk}", 

305 payload=payload, 

306 fingerprint_basis="economics_baseline.study_mode_current_result_metrics", 

307 source_label=f"Study baseline for {study.name}", 

308 source_row_key=f"study:{study.pk}:baseline-study", 

309 source_version=version(run.completed_at) if run is not None and run.completed_at is not None else "", 

310 ) 

311 

312 

313def _current_baseline_result_run(study: EconomicsStudy) -> EconomicsResultRun | None: 

314 """Find the selected baseline study's current result, even when it is cross-flowsheet. 

315 

316 API selection validates read access before a cross-flowsheet baseline study 

317 can be stored. Fingerprinting runs under the target study context, so the 

318 request-scoped manager would otherwise hide the baseline study's own result 

319 rows. The explicit filters keep the read pinned to that selected baseline. 

320 """ 

321 return ( 

322 # Use the unscoped manager because target-study recalculation runs 

323 # under the target flowsheet context; explicit filters pin the baseline. 

324 EconomicsResultRun._base_manager.filter( 

325 flowsheet_state=study.flowsheet_state, 

326 study=study, 

327 status=ResultRunStatus.CURRENT, 

328 ) 

329 .order_by("-created_at", "-pk") 

330 .first() 

331 ) 

332 

333 

334def _baseline_result_line_payload(run: EconomicsResultRun) -> dict[str, Any]: 

335 """Fingerprint only the result metrics consumed by study-baseline financial calculations.""" 

336 rows = {} 

337 # Use the unscoped manager for cross-flowsheet baseline runs, then pin the 

338 # payload to the exact run, owning flowsheet, and consumed metric row keys. 

339 for line in EconomicsResultLine._base_manager.filter( 

340 flowsheet_state=run.flowsheet_state, 

341 result_run=run, 

342 group="financial_metrics", 

343 row_key__in=( 

344 required_financial_metric_spec(FinancialMetricKey.CAPEX).row_key, 

345 required_financial_metric_spec(FinancialMetricKey.ANNUAL_OPEX).row_key, 

346 ), 

347 ): 

348 rows[line.row_key] = { 

349 "amount": line.amount, 

350 "unit": line.unit, 

351 "warning_payload": line.warning_payload, 

352 } 

353 return rows 

354 

355 

356def _baseline_result_run_classification(run: EconomicsResultRun) -> str: 

357 """Mirror the financial-metrics gate that rejects stale study-baseline runs.""" 

358 from Economics.results.services.lifecycle.runs import classify_result_run 

359 

360 return classify_result_run(run) 

361 

362 

363def _schedule_fingerprints( 

364 study: EconomicsStudy, 

365 *, 

366 persisted_fingerprint_value: str = "", 

367 source_version: str | None = None, 

368) -> list[DependencyFingerprint]: 

369 """Fingerprint the study production-schedule selection and source data.""" 

370 

371 scenario = study.schedule_scenario 

372 compact_source_version = ( 

373 schedule_source_version(study) if source_version is None else source_version 

374 ) 

375 if persisted_fingerprint_value and compact_source_version: 

376 return [ 

377 _schedule_dependency( 

378 study=study, 

379 fingerprint_value=persisted_fingerprint_value, 

380 source_version=compact_source_version, 

381 ) 

382 ] 

383 

384 cache_key = _schedule_fingerprint_cache_key( 

385 study=study, 

386 source_version=compact_source_version, 

387 ) 

388 if cache_key: 

389 cached_fingerprint = cache.get(cache_key) 

390 if cached_fingerprint: 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true

391 return [ 

392 _schedule_dependency( 

393 study=study, 

394 fingerprint_value=str(cached_fingerprint), 

395 source_version=compact_source_version, 

396 ) 

397 ] 

398 

399 payload = { 

400 "study_id": study.pk, 

401 "schedule_mode": study.schedule_mode, 

402 "schedule_scenario_id": study.schedule_scenario_id, 

403 "preview": schedule_preview(study).__dict__, 

404 } 

405 source_label = "Steady-state schedule" 

406 source_row_key = f"study:{study.pk}:schedule" 

407 source_version = compact_source_version 

408 sources = {} 

409 if study.schedule_mode == EconomicsScheduleMode.COMPOSITE: 

410 plan_payload = saved_schedule_plan_payload(study) 

411 plan = getattr(study, "schedule_plan", None) 

412 payload["composite_schedule"] = plan_payload.model_dump(mode="json") 

413 payload["source_scenarios"] = _composite_schedule_source_payload(plan) 

414 source_label = "Composite production schedule" 

415 source_row_key = f"schedule_plan:{plan.pk if plan is not None else 'missing'}" 

416 source_version = version(getattr(plan, "updated_at", None)) 

417 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and scenario is not None: 

418 payload["scenario"] = _schedule_scenario_payload(scenario) 

419 source_label = scenario.displayName or "Production schedule" 

420 source_row_key = f"schedule_scenario:{scenario.pk}" 

421 source_version = compact_source_version or version(scenario.created_at) 

422 sources["source_scenario"] = scenario 

423 

424 dependency = _dependency( 

425 dependency_type=ResultDependencyType.SCHEDULE, 

426 dependency_key=f"schedule:{study.pk}", 

427 payload=payload, 

428 fingerprint_basis="economics_study.production_schedule", 

429 source_label=source_label, 

430 source_row_key=source_row_key, 

431 source_version=source_version, 

432 **sources, 

433 ) 

434 if cache_key: 

435 cache.set(cache_key, dependency.fingerprint_value, timeout=86400) 

436 return [dependency] 

437 

438 

439def schedule_source_version(study: EconomicsStudy) -> str: 

440 """Return a compact version for scenario inputs and immutable solved rows.""" 

441 

442 if study.schedule_mode != EconomicsScheduleMode.SCENARIO or study.schedule_scenario_id is None: 

443 return "" 

444 scenario = study.schedule_scenario 

445 if scenario.economics_schedule_revision is not None: 

446 return _persisted_schedule_source_version(scenario) 

447 rows = list( 

448 DataRow.objects.filter(flowsheet_state=scenario.flowsheet_state, scenario=scenario) 

449 .order_by("index", "pk") 

450 .values_list("pk", "index", "created_at") 

451 ) 

452 row_ids = [row_id for row_id, _index, _created_at in rows] 

453 columns = list( 

454 DataColumn.objects.filter(flowsheet_state=scenario.flowsheet_state, scenario=scenario) 

455 .order_by("created_at", "pk") 

456 .values_list("pk", "name", "value", "property_value_id", "created_at") 

457 ) 

458 cells = ( 

459 list( 

460 DataCell.objects.filter(data_row_id__in=row_ids) 

461 .order_by("data_row__index", "data_column__created_at", "pk") 

462 .values_list("pk", "data_row_id", "data_column_id", "value", "created_at") 

463 ) 

464 if row_ids 

465 else [] 

466 ) 

467 solution_version = Solution.objects.filter( 

468 flowsheet_state=scenario.flowsheet_state, 

469 scenario=scenario, 

470 solve_index__isnull=False, 

471 ).aggregate( 

472 signature=MD5( 

473 StringAgg( 

474 Cast( 

475 JSONObject( 

476 id="pk", 

477 property_value_id="property_id", 

478 property_info_id="property__property_id", 

479 solve_index="solve_index", 

480 values="values", 

481 ), 

482 output_field=models.TextField(), 

483 ), 

484 delimiter="\x1e", 

485 ordering=("solve_index", "property_id", "pk"), 

486 default="", 

487 ) 

488 ), 

489 ) 

490 profile = get_settings_profile(study) 

491 payload = { 

492 "study_id": study.pk, 

493 "schedule_mode": study.schedule_mode, 

494 "scenario_id": scenario.pk, 

495 "scenario_fields": { 

496 "name": scenario.displayName, 

497 "state_name": scenario.state_name, 

498 "mss_time_series_enabled": scenario.mss_time_series_enabled, 

499 "mss_time_series_interval": scenario.mss_time_series_interval, 

500 "mss_time_series_unit": scenario.mss_time_series_unit, 

501 "mss_input_mode": scenario.mss_input_mode, 

502 "created_at": scenario.created_at, 

503 }, 

504 "annual_operating_hours": profile.annual_operating_hours if profile is not None else None, 

505 "rows": rows, 

506 "columns": columns, 

507 "cells": cells, 

508 "solutions": solution_version, 

509 } 

510 return f"schedule-v2:{hashlib.sha256(_canonical_json(payload).encode('utf-8')).hexdigest()}" 

511 

512 

513def promote_legacy_schedule_source_version( 

514 study: EconomicsStudy, 

515 *, 

516 verified_source_version: str, 

517) -> str: 

518 """Promote a verified legacy hash to the persisted O(1) schedule generation. 

519 

520 Only dependencies that matched the freshly computed legacy source hash are 

521 promoted. This preserves current/stale classification across deployment 

522 without trusting a migration-time snapshot of potentially changing solves. 

523 """ 

524 

525 scenario = study.schedule_scenario 

526 if scenario is None or not verified_source_version.startswith("schedule-v2:"): 

527 return verified_source_version 

528 with transaction.atomic(): 

529 locked_scenario = Scenario.objects.select_for_update().get(pk=scenario.pk) 

530 dependency_ids_to_promote: list[int] = [] 

531 if locked_scenario.economics_schedule_revision is None: 531 ↛ 562line 531 didn't jump to line 562 because the condition on line 531 was always true

532 # A scenario can back multiple studies. Verify every study's current 

533 # legacy hash before switching the shared scenario to v3; otherwise 

534 # the first study read would strand valid dependencies from the 

535 # remaining studies on an unverifiable v2 source version. 

536 legacy_studies = ( 

537 EconomicsStudy.objects.filter( 

538 result_runs__dependencies__dependency_type=ResultDependencyType.SCHEDULE, 

539 result_runs__dependencies__source_scenario=locked_scenario, 

540 result_runs__dependencies__source_version__startswith="schedule-v2:", 

541 ) 

542 .select_related("schedule_scenario", "settings_profile") 

543 .distinct() 

544 ) 

545 for legacy_study in legacy_studies: 

546 legacy_source_version = ( 

547 verified_source_version 

548 if legacy_study.pk == study.pk 

549 else schedule_source_version(legacy_study) 

550 ) 

551 dependency_ids_to_promote.extend( 

552 EconomicsResultDependency.objects.filter( 

553 result_run__study=legacy_study, 

554 dependency_type=ResultDependencyType.SCHEDULE, 

555 source_scenario=locked_scenario, 

556 source_version=legacy_source_version, 

557 ).values_list("pk", flat=True) 

558 ) 

559 locked_scenario.economics_schedule_revision = uuid.uuid4() 

560 locked_scenario.save(update_fields=["economics_schedule_revision"]) 

561 else: 

562 dependency_ids_to_promote.extend( 

563 EconomicsResultDependency.objects.filter( 

564 dependency_type=ResultDependencyType.SCHEDULE, 

565 source_scenario=locked_scenario, 

566 source_version=verified_source_version, 

567 ).values_list("pk", flat=True) 

568 ) 

569 persisted_source_version = _persisted_schedule_source_version(locked_scenario) 

570 EconomicsResultDependency.objects.filter( 

571 pk__in=dependency_ids_to_promote, 

572 ).update(source_version=persisted_source_version) 

573 scenario.economics_schedule_revision = locked_scenario.economics_schedule_revision 

574 return persisted_source_version 

575 

576 

577def _persisted_schedule_source_version(scenario: Scenario) -> str: 

578 """Format the persisted scenario generation used by schedule dependencies.""" 

579 

580 return f"schedule-v3:{scenario.pk}:{scenario.economics_schedule_revision}" 

581 

582 

583def _schedule_fingerprint_cache_key(*, study: EconomicsStudy, source_version: str) -> str: 

584 if not source_version: 

585 return "" 

586 return f"economics:schedule-fingerprint:{study.pk}:{source_version}" 

587 

588 

589def _schedule_dependency( 

590 *, 

591 study: EconomicsStudy, 

592 fingerprint_value: str, 

593 source_version: str, 

594) -> DependencyFingerprint: 

595 """Recreate dependency metadata around an exact persisted schedule hash.""" 

596 

597 scenario = study.schedule_scenario 

598 return DependencyFingerprint( 

599 dependency_type=ResultDependencyType.SCHEDULE, 

600 dependency_key=f"schedule:{study.pk}", 

601 fingerprint_value=fingerprint_value, 

602 fingerprint_basis="economics_study.production_schedule", 

603 source_label=( 

604 (scenario.displayName or "Production schedule") 

605 if scenario is not None 

606 else "Steady-state schedule" 

607 ), 

608 source_row_key=f"schedule_scenario:{scenario.pk}" if scenario is not None else f"study:{study.pk}:schedule", 

609 source_version=source_version, 

610 source_scenario=scenario, 

611 ) 

612 

613 

614def _schedule_scenario_payload(scenario) -> dict[str, Any]: 

615 rows = list( 

616 DataRow.objects.filter( 

617 flowsheet_state=scenario.flowsheet_state, 

618 scenario=scenario, 

619 ).order_by("index", "pk") 

620 ) 

621 row_ids = [row.pk for row in rows] 

622 row_index_by_id = {row.pk: row.index for row in rows} 

623 return { 

624 "id": scenario.pk, 

625 "name": scenario.displayName, 

626 "state_name": scenario.state_name, 

627 "mss_time_series_enabled": scenario.mss_time_series_enabled, 

628 "mss_time_series_interval": scenario.mss_time_series_interval, 

629 "mss_time_series_unit": scenario.mss_time_series_unit, 

630 "mss_input_mode": scenario.mss_input_mode, 

631 "created_at": scenario.created_at, 

632 "rows": [ 

633 { 

634 "id": row.pk, 

635 "index": row.index, 

636 "created_at": row.created_at, 

637 } 

638 for row in rows 

639 ], 

640 "input_columns": _schedule_input_column_payload(scenario), 

641 "input_values": _schedule_input_value_payload(row_ids=row_ids, row_index_by_id=row_index_by_id), 

642 "solution_values": _schedule_solution_payload(scenario), 

643 } 

644 

645 

646def _composite_schedule_source_payload(plan) -> list[dict[str, Any]]: 

647 """Return child scenario payloads consumed by a composite schedule plan.""" 

648 

649 if plan is None: 649 ↛ 650line 649 didn't jump to line 650 because the condition on line 649 was never true

650 return [] 

651 return [ 

652 _schedule_scenario_payload(rule.source_scenario) 

653 for rule in plan.rules.select_related("source_scenario").order_by("sort_order", "pk") 

654 if rule.source_scenario_id is not None 

655 ] 

656 

657 

658def _schedule_input_column_payload(scenario) -> list[dict[str, Any]]: 

659 return [ 

660 { 

661 "id": column.pk, 

662 "name": column.name, 

663 "property_value_id": column.property_value_id, 

664 "property_info_id": column.property_value.property_id if column.property_value_id else None, 

665 "created_at": column.created_at, 

666 } 

667 for column in DataColumn.objects.filter( 

668 flowsheet_state=scenario.flowsheet_state, 

669 scenario=scenario, 

670 ) 

671 .select_related("property_value") 

672 .order_by("created_at", "pk") 

673 ] 

674 

675 

676def _schedule_input_value_payload( 

677 *, 

678 row_ids: list[int], 

679 row_index_by_id: dict[int, int], 

680) -> list[dict[str, Any]]: 

681 if not row_ids: 681 ↛ 682line 681 didn't jump to line 682 because the condition on line 681 was never true

682 return [] 

683 return [ 

684 { 

685 "id": cell.pk, 

686 "row_index": row_index_by_id.get(cell.data_row_id), 

687 "data_column_id": cell.data_column_id, 

688 "value": cell.value, 

689 "created_at": cell.created_at, 

690 } 

691 for cell in DataCell.objects.filter(data_row_id__in=row_ids) 

692 .order_by("data_row__index", "data_column__created_at", "pk") 

693 ] 

694 

695 

696def _schedule_solution_payload(scenario) -> list[dict[str, Any]]: 

697 return [ 

698 { 

699 "id": solution.pk, 

700 "property_value_id": solution.property_id, 

701 "property_info_id": solution.property.property_id if solution.property_id else None, 

702 "solve_index": solution.solve_index, 

703 "values": solution.values, 

704 "created_at": solution.created_at, 

705 } 

706 for solution in Solution.objects.filter( 

707 flowsheet_state=scenario.flowsheet_state, 

708 scenario=scenario, 

709 solve_index__isnull=False, 

710 ) 

711 .select_related("property") 

712 .order_by("solve_index", "property_id", "pk") 

713 ] 

714 

715 

716def _manual_baseline_inherited_assumptions(study: EconomicsStudy) -> dict[str, Any] | None: 

717 """Return study assumptions that affect manual-baseline financial metrics.""" 

718 assumptions = get_assumptions(study) 

719 if assumptions is None: 719 ↛ 720line 719 didn't jump to line 720 because the condition on line 719 was never true

720 return None 

721 return { 

722 "currency": assumptions.currency, 

723 "basis_date": assumptions.basis_date, 

724 "discount_rate_percent": assumptions.discount_rate_percent, 

725 "project_lifetime_years": assumptions.project_lifetime_years, 

726 "inflation_method": assumptions.inflation_method, 

727 "annual_operating_hours": assumptions.annual_operating_hours, 

728 "tax_rate_percent": assumptions.tax_rate_percent, 

729 "depreciation_enabled": assumptions.depreciation_enabled, 

730 "default_depreciation_life_years": assumptions.default_depreciation_life_years, 

731 "default_depreciation_salvage_percent": assumptions.default_depreciation_salvage_percent, 

732 "contingency_percent": assumptions.contingency_percent, 

733 "electrical_upgrade_rate_amount": assumptions.electrical_upgrade_rate_amount, 

734 "electrical_upgrade_rate_unit": assumptions.electrical_upgrade_rate_unit, 

735 "default_lang_factor": assumptions.default_lang_factor, 

736 "capital_index_series_id": assumptions.capital_index_series_id, 

737 "operating_index_series_id": assumptions.operating_index_series_id, 

738 "default_rate_overrides": assumptions.default_rate_overrides, 

739 } 

740 

741 

742def _costable_item_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]: 

743 """Fingerprint costable items that define the result-row equipment scope.""" 

744 fingerprints = [] 

745 for item in study.costable_items.select_related("simulation_object").order_by("pk"): 

746 fingerprints.append( 

747 _dependency( 

748 dependency_type=ResultDependencyType.COSTABLE_ITEM, 

749 dependency_key=f"costable_item:{item.pk}", 

750 payload={ 

751 "id": item.pk, 

752 "item_type": item.item_type, 

753 "simulation_object_id": item.simulation_object_id, 

754 "simulation_object_type": item.simulation_object.objectType if item.simulation_object_id else None, 

755 "name": item.name, 

756 "included": item.included, 

757 "manual": item.manual, 

758 "notes": item.notes, 

759 "updated_at": item.updated_at, 

760 }, 

761 fingerprint_basis="costable_item.fields", 

762 source_label=item.name, 

763 source_row_key=f"costable_item:{item.pk}", 

764 source_version=version(item.updated_at), 

765 source_costable_item=item, 

766 ) 

767 ) 

768 return fingerprints 

769 

770 

771def _cost_driver_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]: 

772 """Fingerprint cost drivers and their resolved property values.""" 

773 fingerprints = [] 

774 drivers = ( 

775 CostDriver.objects.filter(flowsheet_state=study.flowsheet_state, costable_item__study=study) 

776 .select_related("costable_item", "property_info", "manual_property_info") 

777 .prefetch_related("property_info__values", "manual_property_info__values") 

778 .order_by("pk") 

779 ) 

780 for driver in drivers: 

781 source_property = driver.property_info or driver.manual_property_info 

782 fingerprints.append( 

783 _dependency( 

784 dependency_type=ResultDependencyType.PROPERTY, 

785 dependency_key=f"cost_driver:{driver.pk}", 

786 payload={ 

787 "id": driver.pk, 

788 "costable_item_id": driver.costable_item_id, 

789 "source": driver.source, 

790 "property_info_id": driver.property_info_id, 

791 "manual_property_info_id": driver.manual_property_info_id, 

792 "sizing_mode": driver.sizing_mode, 

793 "canonical_unit": driver.canonical_unit, 

794 "design_value": driver.design_value, 

795 "unresolved_reason_code": driver.unresolved_reason_code, 

796 "warning_payload": driver.warning_payload, 

797 "property": _property_value_payload(driver.property_info), 

798 "manual_property": _property_value_payload(driver.manual_property_info), 

799 "updated_at": driver.updated_at, 

800 }, 

801 fingerprint_basis="cost_driver.fields_and_property_value", 

802 source_label=f"{driver.costable_item.name} cost driver", 

803 source_row_key=f"cost_driver:{driver.pk}",