Coverage for backend/django/Economics/results/services/lifecycle/fingerprints.py: 95%
156 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1"""Dependency fingerprint contracts for Economics presentation result runs.
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"""
9from __future__ import annotations
11import hashlib
12import json
13from typing import Any
15from django.db import models
16from pydantic import field_serializer
18from Economics.costing.models import (
20 CapitalCostLine,
22 CostCurve,
24 CostDriver,
26 CostableItem,
28 EquipmentMapping,
30 OperatingCostLine,
32)
34from Economics.reference_data.models import CostIndexSeries, CostIndexValue
36from Economics.results.models import EconomicsResultDependency, EconomicsResultRun
38from Economics.settings_profiles.models import EconomicsSettingsProfile
40from Economics.studies.models import EconomicsStudy
42from Economics.shared.choices import ResultDependencyType
43from Economics.formulas.engine.core import FORMULA_AUDIT_SCHEMA_VERSION
44from Economics.results.services.lifecycle.common import EconomicsContract, get_assumptions, version
45from Economics.shared.payloads import json_ready
46from core.auxiliary.models import PropertyInfo
49FINGERPRINT_ALGORITHM = "sha256"
50FINGERPRINT_PREFIX = "sha256:"
51FORMULA_REGISTRY_VERSION = "2026-06-15.depreciation-tax-model"
52FORMULA_SEMANTIC_FINGERPRINTS = {
53 "cost_curve": "expression_text parsed by constrained AST-to-SymPy evaluator",
54 "generated_capital_line": "cost_curve * capital_index_factor * optional_lang_factor * contingency_factor",
55 "generated_unit_capex_subtotal": "sum included generated unit-operation capital line formulas",
56 "custom_capital_line": "fixed literal or custom_capex_percentage_basis * basis_percent / 100",
57 "custom_capex_percentage_basis": "generated_unit_capex_subtotal only",
58 "custom_capital_total": "sum included custom capital line formulas",
59 "peak_demand_capacity": "sum included capital line peak-demand capacity in kW",
60 "electrical_upgrade_capex": "peak_demand_capacity * electrical_upgrade_rate_amount",
61 "operating_line": "basis_quantity * annualization_factor * rate_amount",
62 "annual_operating_expense": "sum included non-revenue operating-line formulas",
63 "annual_operating_revenue": "sum included output-revenue operating-line formulas",
64 "default_rate_derived_steam": "fuel_price_nzd_per_gj * steam_energy_gj_per_t / (boiler_efficiency_percent / 100)",
65 "process_energy_contribution": "direct annualized energy quantity or operating-line annual basis converted to MWh",
66 "annual_profit": "target_annual_revenue - target_annual_opex",
67 "annual_savings": "baseline_annual_opex - target_annual_opex + target_annual_revenue",
68 "annual_depreciation": "sum included depreciable capital-line bases less residual value over straight-line equipment life",
69 "depreciation_tax_shield": "annual_depreciation * tax_rate",
70 "after_tax_annual_cash_flow": "annual_savings * (1 - tax_rate) + depreciation_tax_shield",
71 "incremental_capex": "target_capex - baseline_capex",
72 "roi_percent": "((cash_flow_basis * lifetime + residual_value - incremental_capex) / incremental_capex) * 100",
73 "lcoh": "discounted target capex and opex less discounted residual value divided by discounted process energy",
74 "cash_flow_rows": "year 0 negative incremental capex, operating years after-tax annual cash-flow basis, final year residual uplift",
75}
78class DependencyFingerprint(EconomicsContract):
79 """Deterministic fingerprint plus source-row links for one lifecycle input."""
80 dependency_type: str
81 dependency_key: str
82 fingerprint_value: str
83 fingerprint_basis: str
84 source_label: str = ""
85 source_row_key: str = ""
86 source_version: str = ""
87 source_settings_profile: EconomicsSettingsProfile | None = None
88 source_costable_item: CostableItem | None = None
89 source_cost_curve: CostCurve | None = None
90 source_capital_line: CapitalCostLine | None = None
91 source_operating_line: OperatingCostLine | None = None
92 source_index_series: CostIndexSeries | None = None
93 source_index_value: CostIndexValue | None = None
94 source_scenario: Any | None = None
95 source_property_info: Any | None = None
97 @field_serializer(
98 "source_settings_profile",
99 "source_costable_item",
100 "source_cost_curve",
101 "source_capital_line",
102 "source_operating_line",
103 "source_index_series",
104 "source_index_value",
105 "source_scenario",
106 "source_property_info",
107 when_used="json",
108 )
109 def serialize_source_model(self, value):
110 return value.pk if isinstance(value, models.Model) else value
112 @property
113 def identity(self) -> tuple[str, str]:
114 return (self.dependency_type, self.dependency_key)
117def build_dependency_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
118 """Collect source-readable fingerprints for all v1 presentation result inputs.
120 Generated capital lines must be synchronized by the orchestrator before this
121 runs, so persisted generated rows participate in the same audit contract as
122 manually entered rows.
123 """
124 fingerprints: list[DependencyFingerprint] = []
125 fingerprints.extend(_formula_fingerprints())
126 fingerprints.extend(_assumption_fingerprints(study))
127 fingerprints.extend(_baseline_fingerprints(study))
128 fingerprints.extend(_costable_item_fingerprints(study))
129 fingerprints.extend(_cost_driver_fingerprints(study))
130 fingerprints.extend(_equipment_mapping_fingerprints(study))
131 fingerprints.extend(_capital_line_fingerprints(study))
132 fingerprints.extend(_operating_line_fingerprints(study))
133 fingerprints.extend(_cost_curve_fingerprints(study))
134 fingerprints.extend(_index_data_fingerprints(study))
135 return sorted(fingerprints, key=lambda fingerprint: fingerprint.identity)
138def _formula_fingerprints() -> list[DependencyFingerprint]:
139 """Fingerprint formula semantics that are not represented by database rows."""
140 return [
141 _dependency(
142 dependency_type=ResultDependencyType.ASSUMPTIONS,
143 dependency_key="formula:registry",
144 payload={
145 "formula_registry_version": FORMULA_REGISTRY_VERSION,
146 "formula_audit_schema_version": FORMULA_AUDIT_SCHEMA_VERSION,
147 "semantic_fingerprints": FORMULA_SEMANTIC_FINGERPRINTS,
148 },
149 fingerprint_basis="economics_formula_registry.semantic_fingerprints",
150 source_label="Economics formula registry",
151 source_row_key="formula:registry",
152 source_version=FORMULA_REGISTRY_VERSION,
153 )
154 ]
157def _assumption_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
158 """Fingerprint study assumptions or an explicit missing-assumptions marker."""
159 assumptions = get_assumptions(study)
160 if assumptions is None: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 return [
162 _dependency(
163 dependency_type=ResultDependencyType.ASSUMPTIONS,
164 dependency_key=f"assumptions:missing:{study.pk}",
165 payload={"study_id": study.pk, "assumptions": None},
166 fingerprint_basis="economics_assumptions.missing",
167 source_label="Missing economics assumptions",
168 source_row_key=f"study:{study.pk}:assumptions",
169 )
170 ]
171 return [
172 _dependency(
173 dependency_type=ResultDependencyType.ASSUMPTIONS,
174 dependency_key=f"assumptions:{assumptions.pk}",
175 payload={
176 "id": assumptions.pk,
177 "currency": assumptions.currency,
178 "location": assumptions.location,
179 "basis_date": assumptions.basis_date,
180 "discount_rate_percent": assumptions.discount_rate_percent,
181 "project_lifetime_years": assumptions.project_lifetime_years,
182 "inflation_method": assumptions.inflation_method,
183 "annual_operating_hours": assumptions.annual_operating_hours,
184 "tax_rate_percent": assumptions.tax_rate_percent,
185 "depreciation_enabled": assumptions.depreciation_enabled,
186 "default_depreciation_life_years": assumptions.default_depreciation_life_years,
187 "default_depreciation_salvage_percent": assumptions.default_depreciation_salvage_percent,
188 "contingency_percent": assumptions.contingency_percent,
189 "electrical_upgrade_rate_amount": assumptions.electrical_upgrade_rate_amount,
190 "electrical_upgrade_rate_unit": assumptions.electrical_upgrade_rate_unit,
191 "default_lang_factor": assumptions.default_lang_factor,
192 "capital_index_series_id": assumptions.capital_index_series_id,
193 "operating_index_series_id": assumptions.operating_index_series_id,
194 "default_rate_overrides": assumptions.default_rate_overrides,
195 "notes": assumptions.notes,
196 "updated_at": assumptions.updated_at,
197 },
198 fingerprint_basis="economics_assumptions.fields",
199 source_label=f"Assumptions for {study.name}",
200 source_row_key=f"assumptions:{assumptions.pk}",
201 source_version=version(assumptions.updated_at),
202 source_settings_profile=assumptions,
203 )
204 ]
207def _baseline_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
208 """Fingerprint the manual baseline contract for the target study."""
209 baseline = get_assumptions(study)
210 if baseline is None: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true
211 return [
212 _dependency(
213 dependency_type=ResultDependencyType.BASELINE,
214 dependency_key=f"baseline:missing:{study.pk}",
215 payload={"study_id": study.pk, "baseline": None},
216 fingerprint_basis="economics_baseline.missing",
217 source_label="Missing economics baseline",
218 source_row_key=f"study:{study.pk}:baseline",
219 )
220 ]
221 return [
222 _dependency(
223 dependency_type=ResultDependencyType.BASELINE,
224 dependency_key=f"baseline:{baseline.pk}",
225 payload=_baseline_payload(study=study, baseline=baseline),
226 fingerprint_basis="economics_baseline.manual_fields_and_inherited_study_assumptions",
227 source_label=f"Baseline for {study.name}",
228 source_row_key=f"baseline:{baseline.pk}",
229 source_version=version(baseline.updated_at),
230 source_settings_profile=baseline,
231 )
232 ]
235def _baseline_payload(*, study: EconomicsStudy, baseline: EconomicsSettingsProfile) -> dict[str, Any]:
236 """Include manual baseline fields and inherited assumptions that affect calculations."""
237 return {
238 "id": baseline.pk,
239 "mode": "manual",
240 "manual_capex": baseline.manual_capex,
241 "manual_annual_opex": baseline.manual_annual_opex,
242 "annual_heat_basis_mode": baseline.annual_heat_basis_mode,
243 "manual_annual_heat_basis": baseline.manual_annual_heat_basis,
244 "manual_annual_heat_basis_unit": baseline.manual_annual_heat_basis_unit,
245 "average_power_input": baseline.average_power_input,
246 "average_power_unit": baseline.average_power_unit,
247 "residual_value": baseline.residual_value,
248 "notes": baseline.baseline_notes,
249 "inherited_study_assumptions": _manual_baseline_inherited_assumptions(study),
250 "updated_at": baseline.updated_at,
251 }
254def _manual_baseline_inherited_assumptions(study: EconomicsStudy) -> dict[str, Any] | None:
255 """Return study assumptions that affect manual-baseline financial metrics."""
256 assumptions = get_assumptions(study)
257 if assumptions is None: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 return None
259 return {
260 "currency": assumptions.currency,
261 "basis_date": assumptions.basis_date,
262 "discount_rate_percent": assumptions.discount_rate_percent,
263 "project_lifetime_years": assumptions.project_lifetime_years,
264 "inflation_method": assumptions.inflation_method,
265 "annual_operating_hours": assumptions.annual_operating_hours,
266 "tax_rate_percent": assumptions.tax_rate_percent,
267 "depreciation_enabled": assumptions.depreciation_enabled,
268 "default_depreciation_life_years": assumptions.default_depreciation_life_years,
269 "default_depreciation_salvage_percent": assumptions.default_depreciation_salvage_percent,
270 "contingency_percent": assumptions.contingency_percent,
271 "electrical_upgrade_rate_amount": assumptions.electrical_upgrade_rate_amount,
272 "electrical_upgrade_rate_unit": assumptions.electrical_upgrade_rate_unit,
273 "default_lang_factor": assumptions.default_lang_factor,
274 "capital_index_series_id": assumptions.capital_index_series_id,
275 "operating_index_series_id": assumptions.operating_index_series_id,
276 "default_rate_overrides": assumptions.default_rate_overrides,
277 }
280def _costable_item_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
281 """Fingerprint costable items that define the result-row equipment scope."""
282 fingerprints = []
283 for item in study.costable_items.select_related("simulation_object").order_by("pk"):
284 fingerprints.append(
285 _dependency(
286 dependency_type=ResultDependencyType.COSTABLE_ITEM,
287 dependency_key=f"costable_item:{item.pk}",
288 payload={
289 "id": item.pk,
290 "item_type": item.item_type,
291 "simulation_object_id": item.simulation_object_id,
292 "simulation_object_type": item.simulation_object.objectType if item.simulation_object_id else None,
293 "name": item.name,
294 "included": item.included,
295 "manual": item.manual,
296 "notes": item.notes,
297 "updated_at": item.updated_at,
298 },
299 fingerprint_basis="costable_item.fields",
300 source_label=item.name,
301 source_row_key=f"costable_item:{item.pk}",
302 source_version=version(item.updated_at),
303 source_costable_item=item,
304 )
305 )
306 return fingerprints
309def _cost_driver_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
310 """Fingerprint cost drivers and their resolved property values."""
311 fingerprints = []
312 drivers = (
313 CostDriver.objects.filter(flowsheet=study.flowsheet, costable_item__study=study)
314 .select_related("costable_item", "property_info", "manual_property_info")
315 .order_by("pk")
316 )
317 for driver in drivers:
318 source_property = driver.property_info or driver.manual_property_info
319 fingerprints.append(
320 _dependency(
321 dependency_type=ResultDependencyType.PROPERTY,
322 dependency_key=f"cost_driver:{driver.pk}",
323 payload={
324 "id": driver.pk,
325 "costable_item_id": driver.costable_item_id,
326 "source": driver.source,
327 "property_info_id": driver.property_info_id,
328 "manual_property_info_id": driver.manual_property_info_id,
329 "sizing_mode": driver.sizing_mode,
330 "canonical_unit": driver.canonical_unit,
331 "design_value": driver.design_value,
332 "unresolved_reason_code": driver.unresolved_reason_code,
333 "warning_payload": driver.warning_payload,
334 "property": _property_value_payload(driver.property_info),
335 "manual_property": _property_value_payload(driver.manual_property_info),
336 "updated_at": driver.updated_at,
337 },
338 fingerprint_basis="cost_driver.fields_and_property_value",
339 source_label=f"{driver.costable_item.name} cost driver",
340 source_row_key=f"cost_driver:{driver.pk}",
341 source_version=version(driver.updated_at),
342 source_costable_item=driver.costable_item,
343 source_property_info=source_property,
344 )
345 )
346 return fingerprints
349def _equipment_mapping_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
350 """Fingerprint equipment-to-curve mappings used by generated capital lines."""
351 fingerprints = []
352 mappings = (
353 EquipmentMapping.objects.filter(flowsheet=study.flowsheet, costable_item__study=study)
354 .select_related("costable_item", "cost_curve")
355 .order_by("pk")
356 )
357 for mapping in mappings:
358 fingerprints.append(
359 _dependency(
360 dependency_type=ResultDependencyType.COSTABLE_ITEM,
361 dependency_key=f"equipment_mapping:{mapping.pk}",
362 payload={
363 "id": mapping.pk,
364 "costable_item_id": mapping.costable_item_id,
365 "cost_curve_id": mapping.cost_curve_id,
366 "equipment_category": mapping.equipment_category,
367 "equipment_subtype": mapping.equipment_subtype,
368 "cost_basis": mapping.cost_basis,
369 "install_factor_profile": mapping.install_factor_profile,
370 "install_factor": mapping.install_factor,
371 "use_study_lang_factor": mapping.use_study_lang_factor,
372 "applicability_notes": mapping.applicability_notes,
373 "updated_at": mapping.updated_at,
374 },
375 fingerprint_basis="equipment_mapping.fields",
376 source_label=f"{mapping.costable_item.name} equipment mapping",
377 source_row_key=f"equipment_mapping:{mapping.pk}",
378 source_version=version(mapping.updated_at),
379 source_costable_item=mapping.costable_item,
380 source_cost_curve=mapping.cost_curve,
381 )
382 )
383 return fingerprints
386def _property_value_payload(property_info) -> dict[str, Any] | None:
387 """Serialize the first property value used by a property-backed cost driver."""
388 if property_info is None:
389 return None
390 value = property_info.values.order_by("pk").first()
391 return {
392 "id": property_info.pk,
393 "key": property_info.key,
394 "display_name": property_info.displayName,
395 "unit": property_info.unit,
396 "unit_type": property_info.unitType,
397 "value_id": value.pk if value is not None else None,
398 "value": value.value if value is not None else None,
399 "display_value": value.displayValue if value is not None else None,
400 "enabled": value.enabled if value is not None else None,
401 "formula": value.formula if value is not None else None,
402 }
405def _capital_line_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
406 """Fingerprint included and excluded capital source rows for the study."""
407 fingerprints = []
408 for line in study.capital_lines.select_related("costable_item", "cost_curve").order_by("pk"):
409 fingerprints.append(
410 _dependency(
411 dependency_type=ResultDependencyType.CAPITAL_LINE,
412 dependency_key=f"capital_line:{line.pk}",
413 payload=_capital_line_payload(line),
414 fingerprint_basis="capital_cost_line.fields",
415 source_label=line.label,
416 source_row_key=f"capital_line:{line.pk}",
417 source_version=version(line.updated_at),
418 source_costable_item=line.costable_item,
419 source_cost_curve=line.cost_curve,
420 source_capital_line=line,
421 )
422 )
423 return fingerprints
426def _operating_line_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
427 """Fingerprint operating source rows, including pricing and resource metadata."""
428 fingerprints = []
429 for line in study.operating_lines.select_related("costable_item", "source_property_info").order_by("pk"):
430 fingerprints.append(
431 _dependency(
432 dependency_type=ResultDependencyType.OPERATING_LINE,
433 dependency_key=f"operating_line:{line.pk}",
434 payload=_operating_line_payload(line),
435 fingerprint_basis="operating_cost_line.fields",
436 source_label=line.label,
437 source_row_key=f"operating_line:{line.pk}",
438 source_version=version(line.updated_at),
439 source_costable_item=line.costable_item,
440 source_property_info=line.source_property_info,
441 source_operating_line=line,
442 )
443 )
444 return fingerprints
447def _capital_line_payload(line: CapitalCostLine) -> dict[str, Any]:
448 """Serialize capital-line fields that can affect result materialization."""
449 return {
450 "id": line.pk,
451 "study_id": line.study_id,
452 "costable_item_id": line.costable_item_id,
453 "cost_curve_id": line.cost_curve_id,
454 "label": line.label,
455 "line_type": line.line_type,
456 "calculation_basis": line.calculation_basis,
457 "amount": line.amount,
458 "basis_percent": line.basis_percent,
459 "depreciation_mode": line.depreciation_mode,
460 "depreciation_life_years": line.depreciation_life_years,
461 "depreciation_salvage_percent": line.depreciation_salvage_percent,
462 "peak_demand_kw": line.peak_demand_kw,
463 "minimum_peak_demand_kw": line.minimum_peak_demand_kw,
464 "currency": line.currency,
465 "included": line.included,
466 "manual": line.manual,
467 "source": line.source,
468 "confidence": line.confidence,
469 "warning_payload": line.warning_payload,
470 "driver_inputs": line.driver_inputs,
471 "driver_input_property_values": _driver_input_property_values(line),
472 "updated_at": line.updated_at,
473 }
476def _driver_input_property_values(line: CapitalCostLine) -> list[dict[str, Any]]:
477 """Return values for properties referenced by generated-line driver inputs."""
478 if not isinstance(line.driver_inputs, dict): 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true
479 return []
480 property_ids = sorted(
481 {
482 driver_input.get("property_info")
483 for driver_input in line.driver_inputs.values()
484 if isinstance(driver_input, dict)
485 and driver_input.get("source") == "property"
486 and driver_input.get("property_info") is not None
487 }
488 )
489 if not property_ids:
490 return []
491 properties = {
492 property_info.pk: property_info
493 for property_info in PropertyInfo.objects.filter(
494 flowsheet=line.flowsheet,
495 pk__in=property_ids,
496 )
497 }
498 return [
499 {
500 "property_info_id": property_id,
501 "unit": properties[property_id].unit if property_id in properties else "",
502 "value": properties[property_id].get_value() if property_id in properties else None,
503 }
504 for property_id in property_ids
505 ]
508def _operating_line_payload(line: OperatingCostLine) -> dict[str, Any]:
509 """Serialize operating-line fields that can affect costs or resource grouping."""
510 return {
511 "id": line.pk,
512 "study_id": line.study_id,
513 "costable_item_id": line.costable_item_id,
514 "label": line.label,
515 "line_type": line.line_type,
516 "category": line.category,
517 "currency": line.currency,
518 "basis_quantity": line.basis_quantity,
519 "basis_unit": line.basis_unit,
520 "basis_quantity_source": line.basis_quantity_source,
521 "rate_amount": line.rate_amount,
522 "rate_unit": line.rate_unit,
523 "rate_type": line.rate_type,
524 "rate_source_mode": line.rate_source_mode,
525 "calculation_method": line.calculation_method,
526 "source_property_info_id": line.source_property_info_id,
527 "source_default_rate_id": line.source_default_rate_id,
528 "outlet_stream_disposition": line.outlet_stream_disposition,
529 "included": line.included,
530 "manual": line.manual,
531 "source": line.source,
532 "warning_payload": line.warning_payload,
533 "updated_at": line.updated_at,
534 }
537def _cost_curve_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
538 """Fingerprint only cost curves referenced by this study's capital setup."""
539 curve_ids = set(
540 study.capital_lines.filter(cost_curve__isnull=False).values_list("cost_curve_id", flat=True)
541 )
542 costable_item_ids = study.costable_items.values_list("pk", flat=True)
543 curve_ids.update(
544 EquipmentMapping.objects.filter(
545 flowsheet=study.flowsheet,
546 costable_item_id__in=costable_item_ids,
547 cost_curve__isnull=False,
548 ).values_list("cost_curve_id", flat=True)
549 )
550 fingerprints = []
551 for curve in CostCurve.objects.filter(flowsheet=study.flowsheet, pk__in=curve_ids).order_by("pk"):
552 fingerprints.append(
553 _dependency(
554 dependency_type=ResultDependencyType.COST_CURVE,
555 dependency_key=f"cost_curve:{curve.pk}",
556 payload={
557 "id": curve.pk,
558 "curve_key": curve.curve_key,
559 "name": curve.name,
560 "equipment_category": curve.equipment_category,
561 "equipment_subtype": curve.equipment_subtype,
562 "cost_basis": curve.cost_basis,
563 "evaluation_kind": curve.evaluation_kind,
564 "output_unit": curve.output_unit,
565 "expression_text": curve.expression_text,
566 "required_driver_specs": curve.required_driver_specs,
567 "discrete_variants": curve.discrete_variants,
568 "valid_min": curve.valid_min,
569 "valid_max": curve.valid_max,
570 "valid_range_note": curve.valid_range_note,
571 "currency": curve.currency,
572 "basis_date": curve.basis_date,
573 "basis_index_name": curve.basis_index_name,
574 "basis_index_value": curve.basis_index_value,
575 "source_document_title": curve.source_document_title,
576 "source_page": curve.source_page,
577 "source_figure": curve.source_figure,
578 "source_data_origin": curve.source_data_origin,
579 "source_range_precision": curve.source_range_precision,
580 "source_license_status": curve.source_license_status,
581 "source_reference": curve.source_reference,
582 "source_note": curve.source_note,
583 "applicability_warning": curve.applicability_warning,
584 "active": curve.active,
585 "updated_at": curve.updated_at,
586 },
587 fingerprint_basis="cost_curve.fields",
588 source_label=curve.name,
589 source_row_key=f"cost_curve:{curve.curve_key}",
590 source_version=version(curve.updated_at),
591 source_cost_curve=curve,
592 )
593 )
594 return fingerprints
597def _index_data_fingerprints(study: EconomicsStudy) -> list[DependencyFingerprint]:
598 """Fingerprint selected index series and values used for capital escalation."""
599 assumptions = get_assumptions(study)
600 if assumptions is None: 600 ↛ 601line 600 didn't jump to line 601 because the condition on line 600 was never true
601 return []
602 series_ids = {assumptions.capital_index_series_id, assumptions.operating_index_series_id}
603 series_ids.discard(None)
604 fingerprints = []
605 for series in CostIndexSeries.objects.filter(pk__in=series_ids).order_by("pk"):
606 values = list(series.values.order_by("period_date", "pk"))
607 fingerprints.append(
608 _dependency(
609 dependency_type=ResultDependencyType.INDEX_SERIES,
610 dependency_key=f"index_series:{series.pk}",
611 payload={
612 "id": series.pk,
613 "key": series.key,
614 "name": series.name,
615 "provider": series.provider,
616 "source_series_id": series.source_series_id,
617 "frequency": series.frequency,
618 "unit": series.unit,
619 "index_basis": series.index_basis,
620 "source_url": series.source_url,
621 "release_title": series.release_title,
622 "source_asset_filename": series.source_asset_filename,
623 "source_asset_file_id": series.source_asset_file_id,
624 "source_parent_id": series.source_parent_id,
625 "latest_imported_period": series.latest_imported_period,
626 "updated_at": series.updated_at,
627 "values": [_index_value_payload(value) for value in values],
628 },
629 fingerprint_basis="cost_index_series.fields_and_values",
630 source_label=series.name,
631 source_row_key=f"index_series:{series.key}",
632 source_version=version(series.updated_at),
633 source_index_series=series,
634 )
635 )
636 for value in values:
637 fingerprints.append(
638 _dependency(
639 dependency_type=ResultDependencyType.INDEX_VALUE,
640 dependency_key=f"index_value:{value.pk}",
641 payload=_index_value_payload(value),
642 fingerprint_basis="cost_index_value.fields",
643 source_label=f"{series.name} {value.period}",
644 source_row_key=f"index_value:{series.key}:{value.period}",
645 source_version=value.period,
646 source_index_series=series,
647 source_index_value=value,
648 )
649 )
650 return fingerprints
653def _create_dependencies(*, result_run: EconomicsResultRun, fingerprints: list[DependencyFingerprint]) -> None:
654 """Persist fingerprint contracts as auditable dependency rows for a result run."""
655 for fingerprint in fingerprints:
656 EconomicsResultDependency.objects.create(
657 flowsheet=result_run.flowsheet,
658 result_run=result_run,
659 dependency_type=fingerprint.dependency_type,
660 dependency_key=fingerprint.dependency_key,
661 fingerprint_value=fingerprint.fingerprint_value,
662 fingerprint_algorithm=FINGERPRINT_ALGORITHM,
663 fingerprint_basis=fingerprint.fingerprint_basis,
664 source_label=fingerprint.source_label,
665 source_row_key=fingerprint.source_row_key,
666 source_version=fingerprint.source_version,
667 source_settings_profile=fingerprint.source_settings_profile,
668 source_costable_item=fingerprint.source_costable_item,
669 source_cost_curve=fingerprint.source_cost_curve,
670 source_capital_line=fingerprint.source_capital_line,
671 source_operating_line=fingerprint.source_operating_line,
672 source_index_series=fingerprint.source_index_series,
673 source_index_value=fingerprint.source_index_value,
674 source_scenario=fingerprint.source_scenario,
675 source_property_info=fingerprint.source_property_info,
676 )
679def _dependency(
680 *,
681 dependency_type: str,
682 dependency_key: str,
683 payload: dict[str, Any],
684 fingerprint_basis: str,
685 source_label: str = "",
686 source_row_key: str = "",
687 source_version: str = "",
688 **sources,
689) -> DependencyFingerprint:
690 """Build one dependency fingerprint from an already-normalized source payload."""
691 return DependencyFingerprint(
692 dependency_type=dependency_type,
693 dependency_key=dependency_key,
694 fingerprint_value=_fingerprint_payload(payload),
695 fingerprint_basis=fingerprint_basis,
696 source_label=source_label,
697 source_row_key=source_row_key,
698 source_version=source_version,
699 **sources,
700 )
703def _fingerprint_payload(payload: dict[str, Any]) -> str:
704 """Hash a normalized dependency payload using the persisted algorithm label."""
705 return f"{FINGERPRINT_PREFIX}{hashlib.sha256(_canonical_json(payload).encode('utf-8')).hexdigest()}"
708def _canonical_json(payload: dict[str, Any]) -> str:
709 """Return canonical JSON so semantically identical payloads hash identically."""
710 return json.dumps(json_ready(payload), sort_keys=True, separators=(",", ":"))
713def _fingerprint_map(fingerprints: list[DependencyFingerprint]) -> dict[tuple[str, str], str]:
714 """Return the comparable identity-to-hash map for current fingerprints."""
715 return {fingerprint.identity: fingerprint.fingerprint_value for fingerprint in fingerprints}
718def _stored_dependency_map(result_run: EconomicsResultRun) -> dict[tuple[str, str], str]:
719 """Return the comparable identity-to-hash map persisted for a result run."""
720 return {
721 (dependency.dependency_type, dependency.dependency_key): dependency.fingerprint_value
722 for dependency in result_run.dependencies.order_by("dependency_type", "dependency_key")
723 }
726def _index_value_payload(value: CostIndexValue) -> dict[str, Any]:
727 """Serialize one index value row for both series and row-level fingerprints."""
728 return {
729 "id": value.pk,
730 "series_id": value.series_id,
731 "period": value.period,
732 "period_date": value.period_date,
733 "value": value.value,
734 "status": value.status,
735 "source_asset_filename": value.source_asset_filename,
736 "source_series_reference": value.source_series_reference,
737 "source_period": value.source_period,
738 "source_units": value.source_units,
739 "source_subject": value.source_subject,
740 "source_group": value.source_group,
741 "source_series_title_1": value.source_series_title_1,
742 }