Coverage for backend/django/Economics/results/services/lifecycle/result_lines.py: 89%
132 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Result-line materialization for Economics presentation runs.
3This module copies already-calculated financial meaning into persisted
4``EconomicsResultLine`` rows. It owns row keys, grouping, quantization, and
5warning payload shape; it must not decide whether a run is current or stale.
6"""
8from __future__ import annotations
10from decimal import Decimal, ROUND_HALF_UP
11from typing import Any
13from core.auxiliary.models.PropertyInfo import PropertyInfo
14from Economics.results.models import EconomicsResultLine, EconomicsResultRun
15from Economics.studies.models import EconomicsStudy
16from Economics.costing.models import OperatingCostLine
17from Economics.shared.choices import ResultLineKind
18from Economics.settings_profiles.services.depreciation import build_straight_line_depreciation_schedule
19from Economics.results.services.financial_metrics.contracts import FinancialMetric
20from Economics.results.services.financial_metrics.metric_catalog import FinancialMetricKey
21from Economics.formulas.engine.core import FormulaError, FormulaEvaluation
22from Economics.formulas.builders.capital import build_electrical_upgrade_formula
23from Economics.formulas.builders.operating import (
24 PROCESS_ENERGY_QUANTITY_UNIT,
25 build_process_energy_contribution_formula,
26)
27from Economics.costing.operating.line_calculation import AnnualBasisQuantity
28from Economics.costing.operating.schedule_calculation import OperatingLineCalculation, calculate_operating_line
29from Economics.costing.operating.resource_basis import (
30 RESOURCE_CATEGORY_ANNUAL_COOLING_DUTY,
31 RESOURCE_CATEGORY_ANNUAL_HEATING_DUTY,
32 operating_resource_breakdown_category,
33 operating_resource_source_kind,
34)
35from Economics.costing.operating.resource_classification import operating_resource_classification
36from Economics.costing.operating.resource_classification import operating_resource_metric_quantity
37from Economics.results.services.lifecycle.common import EconomicsContract
38from Economics.shared.payloads import result_amount
41RESULT_RESOURCE_QUANTITY_QUANTUM = Decimal("0.00000001")
44class OperatingResourcePayload(EconomicsContract):
45 """Source-resource fields persisted on operating result lines."""
46 source_property_info: PropertyInfo | None = None
47 resource_source_kind: str = ""
48 resource_source_object_name: str = ""
49 resource_source_object_type: str = ""
50 resource_property_name: str = ""
51 resource_breakdown_category: str = ""
52 resource_classification: str = ""
53 resource_metric_quantity: Decimal | None = None
54 resource_metric_unit: str = ""
55 annual_basis_quantity: Decimal | None = None
56 annual_basis_unit: str = ""
59def _create_metric_lines(*, result_run: EconomicsResultRun, metrics: dict[FinancialMetricKey, FinancialMetric]) -> None:
60 """Upsert financial metric rows while removing metrics no longer emitted."""
61 metric_row_keys = {metric.row_key for metric in metrics.values()}
62 result_run.lines.filter(group="financial_metrics").exclude(row_key__in=metric_row_keys).delete()
63 rows = []
64 for sort_order, metric in enumerate(sorted(metrics.values(), key=_metric_sort_key), start=10):
65 rows.append(EconomicsResultLine(
66 flowsheet_state=result_run.flowsheet_state,
67 result_run=result_run,
68 row_key=metric.row_key,
69 kind=metric.line_kind,
70 group="financial_metrics",
71 label=metric.label,
72 amount=result_amount(metric.value),
73 unit=metric.unit,
74 source_row_key=metric.row_key,
75 source_label=metric.label,
76 warning_payload={
77 "status": metric.status,
78 "raw_value": str(metric.value) if metric.value is not None else None,
79 "assumptions": metric.assumptions.to_mapping(),
80 "formula_audit": _metric_formula_payload(metric, study=result_run.study),
81 },
82 sort_order=sort_order,
83 ))
84 _bulk_upsert_result_lines(rows)
87def _create_capital_result_lines(*, result_run: EconomicsResultRun) -> None:
88 """Upsert included capital-line detail rows for a presentation result run."""
89 capital_lines = result_run.study.capital_lines.filter(included=True, amount__isnull=False).select_related(
90 "costable_item",
91 "cost_curve",
92 )
93 rows = []
94 for sort_order, line in enumerate(capital_lines, start=100):
95 rows.append(EconomicsResultLine(
96 flowsheet_state=result_run.flowsheet_state,
97 result_run=result_run,
98 row_key=f"capital_line.{line.pk}",
99 kind=ResultLineKind.CAPITAL,
100 group="capital_lines",
101 label=line.label,
102 amount=result_amount(line.amount),
103 unit=line.currency,
104 source_costable_item=line.costable_item,
105 source_cost_curve=line.cost_curve,
106 source_capital_line=line,
107 source_row_key=f"capital_line:{line.pk}",
108 source_label=line.label,
109 source_note=line.source,
110 warning_payload={
111 "line_type": line.line_type,
112 "calculation_basis": line.calculation_basis,
113 "basis_percent": str(line.basis_percent) if line.basis_percent is not None else None,
114 "peak_demand_kw": str(line.peak_demand_kw) if line.peak_demand_kw is not None else None,
115 "minimum_peak_demand_kw": str(line.minimum_peak_demand_kw)
116 if line.minimum_peak_demand_kw is not None
117 else None,
118 "manual": line.manual,
119 "confidence": line.confidence,
120 "capital_factors": _capital_factor_list(line.warning_payload),
121 "warnings": _line_warning_list(line.warning_payload),
122 },
123 sort_order=sort_order,
124 ))
125 _bulk_upsert_result_lines(rows)
128def _create_depreciation_result_lines(*, result_run: EconomicsResultRun) -> None:
129 """Replace annual straight-line depreciation rows for this result run."""
130 result_run.lines.filter(group="depreciation_lines").delete()
131 schedule = build_straight_line_depreciation_schedule(result_run.study)
132 for sort_order, line in enumerate(schedule.lines, start=400):
133 EconomicsResultLine.objects.create(
134 flowsheet_state=result_run.flowsheet_state,
135 result_run=result_run,
136 kind=ResultLineKind.DEPRECIATION,
137 group="depreciation_lines",
138 label=line.label,
139 row_key=f"depreciation_line.{line.capital_line_id}",
140 amount=result_amount(line.annual_depreciation),
141 unit=f"{result_run.result_currency}/year",
142 source_capital_line_id=line.capital_line_id,
143 source_row_key=f"capital_line:{line.capital_line_id}",
144 source_label=line.label,
145 warning_payload={
146 "depreciation_mode": line.depreciation_mode,
147 "depreciable_basis": str(line.depreciable_basis),
148 "life_years": line.life_years,
149 "salvage_percent": str(line.salvage_percent),
150 "annual_depreciation": str(line.annual_depreciation),
151 },
152 sort_order=sort_order,
153 )
156def _create_operating_result_lines(*, result_run: EconomicsResultRun) -> None:
157 """Upsert included operating-line detail rows with resource breakdown fields."""
158 operating_lines = result_run.study.operating_lines.filter(included=True).select_related(
159 "costable_item",
160 "source_default_rate",
161 "source_property_info__set__simulationObject",
162 )
163 rows = []
164 for sort_order, line in enumerate(operating_lines, start=500):
165 calculation = calculate_operating_line(line, study=result_run.study)
166 if calculation.annual_amount is None: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true
167 continue
168 resource_payload = _operating_resource_payload(
169 line=line,
170 study=result_run.study,
171 annual_basis=calculation.annual_basis,
172 schedule_controlled=calculation.schedule_controlled,
173 )
174 resource_defaults = resource_payload.model_dump()
175 rows.append(EconomicsResultLine(
176 flowsheet_state=result_run.flowsheet_state,
177 result_run=result_run,
178 row_key=f"operating_line.{line.pk}",
179 kind=ResultLineKind.OPERATING,
180 group="operating_lines",
181 label=line.label,
182 amount=result_amount(calculation.annual_amount),
183 unit=f"{line.currency}/year",
184 source_costable_item=line.costable_item,
185 source_operating_line=line,
186 source_row_key=f"operating_line:{line.pk}",
187 source_label=line.label,
188 source_note=line.source,
189 warning_payload={
190 "line_type": line.line_type,
191 "category": line.category,
192 "economic_effect": line.economic_effect,
193 "manual": line.manual,
194 "basis_quantity": str(line.basis_quantity) if line.basis_quantity is not None else None,
195 "basis_unit": line.basis_unit,
196 "rate_amount": str(line.rate_amount) if line.rate_amount is not None else None,
197 "rate_unit": line.rate_unit,
198 "calculation_method": line.calculation_method,
199 "formula_audit": _operating_formula_audit(calculation),
200 "schedule_controlled": calculation.schedule_controlled,
201 "schedule": _operating_schedule_payload(calculation),
202 "source_default_rate": line.source_default_rate.key if line.source_default_rate_id else "",
203 "outlet_stream_disposition": line.outlet_stream_disposition,
204 "warnings": _line_warning_list(line.warning_payload),
205 },
206 sort_order=sort_order,
207 **resource_defaults,
208 ))
209 _bulk_upsert_result_lines(rows)
212def _bulk_upsert_result_lines(rows: list[EconomicsResultLine]) -> None:
213 """Persist result materialization in one conflict-aware statement."""
214 if not rows:
215 return
216 EconomicsResultLine.objects.bulk_create(
217 rows,
218 update_conflicts=True,
219 unique_fields=("result_run", "row_key"),
220 update_fields=(
221 "kind", "group", "label", "amount", "unit",
222 "source_costable_item", "source_cost_curve", "source_capital_line",
223 "source_operating_line", "source_index_value", "source_property_info",
224 "source_row_key", "source_label", "source_note", "resource_source_kind",
225 "resource_source_object_name", "resource_source_object_type",
226 "resource_property_name", "resource_breakdown_category",
227 "resource_classification", "resource_metric_quantity", "resource_metric_unit",
228 "annual_basis_quantity", "annual_basis_unit", "warning_payload", "sort_order",
229 ),
230 )
233def _operating_resource_payload(
234 *,
235 line: OperatingCostLine,
236 study: EconomicsStudy,
237 annual_basis: AnnualBasisQuantity | None,
238 schedule_controlled: bool = False,
239) -> OperatingResourcePayload:
240 """Build the operating resource audit fields shared by tables and chart drilldowns."""
241 source_object = None
242 property_info = line.source_property_info if line.source_property_info_id else None
243 if line.source_property_info_id:
244 source_object = line.source_property_info.set.simulationObject
245 source_kind = operating_resource_source_kind(line=line, source_object=source_object, property_info=property_info)
246 resource_category = operating_resource_breakdown_category(
247 line=line,
248 source_kind=source_kind,
249 )
250 display_annual_basis = _resource_display_annual_basis(
251 line=line,
252 study=study,
253 resource_category=resource_category,
254 fallback=annual_basis,
255 schedule_controlled=schedule_controlled,
256 )
257 classification = operating_resource_classification(
258 line=line,
259 resource_category=resource_category,
260 )
261 resource_metric_basis = operating_resource_metric_quantity(
262 line=line,
263 classification=classification,
264 annual_basis=display_annual_basis,
265 )
266 payload = OperatingResourcePayload(
267 source_property_info=property_info,
268 resource_source_kind=source_kind,
269 resource_source_object_name=getattr(source_object, "componentName", "") or "",
270 resource_source_object_type=getattr(source_object, "objectType", "") or "",
271 resource_property_name=getattr(property_info, "displayName", "") or "",
272 resource_breakdown_category=resource_category,
273 resource_classification=classification,
274 )
275 if resource_metric_basis is not None: 275 ↛ 285line 275 didn't jump to line 285 because the condition on line 275 was always true
276 payload = payload.model_copy(
277 update={
278 "resource_metric_quantity": resource_metric_basis.quantity.quantize(
279 RESULT_RESOURCE_QUANTITY_QUANTUM,
280 rounding=ROUND_HALF_UP,
281 ),
282 "resource_metric_unit": resource_metric_basis.unit,
283 }
284 )
285 if display_annual_basis is not None: 285 ↛ 295line 285 didn't jump to line 295 because the condition on line 285 was always true
286 payload = payload.model_copy(
287 update={
288 "annual_basis_quantity": display_annual_basis.quantity.quantize(
289 RESULT_RESOURCE_QUANTITY_QUANTUM,
290 rounding=ROUND_HALF_UP,
291 ),
292 "annual_basis_unit": display_annual_basis.unit,
293 }
294 )
295 return payload
298def _resource_display_annual_basis(
299 *,
300 line: OperatingCostLine,
301 study: EconomicsStudy,
302 resource_category: str,
303 fallback: AnnualBasisQuantity | None,
304 schedule_controlled: bool,
305) -> AnnualBasisQuantity | None:
306 """Prefer process-energy totals for heat duties unless a production schedule already controls them."""
307 if resource_category not in {
308 RESOURCE_CATEGORY_ANNUAL_COOLING_DUTY,
309 RESOURCE_CATEGORY_ANNUAL_HEATING_DUTY,
310 } or schedule_controlled:
311 return fallback
312 try:
313 contribution_formula = build_process_energy_contribution_formula(line, study=study)
314 annual_energy = contribution_formula.evaluate()
315 except FormulaError:
316 annual_energy = None
317 if annual_energy is None: 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true
318 return fallback
319 return AnnualBasisQuantity(quantity=annual_energy, unit=PROCESS_ENERGY_QUANTITY_UNIT)
322def _operating_formula_audit(calculation: OperatingLineCalculation) -> dict[str, Any] | None:
323 """Persist enough calculation detail for result-line explanations and chart tooltips."""
324 if calculation.annual_amount is None: 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true
325 return None
326 if calculation.schedule_controlled:
327 return {
328 "kind": "production_schedule",
329 "annual_amount": str(calculation.annual_amount),
330 "annual_basis_quantity": str(calculation.annual_basis.quantity)
331 if calculation.annual_basis is not None
332 else None,
333 "annual_basis_unit": calculation.annual_basis.unit if calculation.annual_basis is not None else "",
334 "rate_amount": str(calculation.rate_amount) if calculation.rate_amount is not None else None,
335 "rate_unit": calculation.rate_unit,
336 }
337 operating_formula = calculation.steady_formula
338 if operating_formula is None: 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 return None
340 return operating_formula.formula.audit_payload(
341 FormulaEvaluation(
342 value=calculation.annual_amount,
343 bindings={
344 operating_formula.formula.inputs[0].key: operating_formula.basis_quantity,
345 },
346 conversion_diagnostics=(
347 {
348 "input": operating_formula.formula.inputs[0].key,
349 "source_value": str(operating_formula.basis_quantity),
350 "source_unit": operating_formula.basis_unit,
351 "target_value": str(operating_formula.annual_basis.quantity),
352 "target_unit": operating_formula.annual_basis.unit,
353 },
354 ),
355 )
356 )
359def _operating_schedule_payload(calculation: OperatingLineCalculation) -> dict[str, Any] | None:
360 """Store per-schedule-row contribution data needed to reconstruct operating timelines."""
361 if not calculation.schedule_controlled or calculation.schedule_resolution is None:
362 return None
363 resolution = calculation.schedule_resolution
364 return {
365 "scenario": resolution.scenario_id,
366 "source": "Production schedule",
367 "property": resolution.property_name,
368 "unit": resolution.unit,
369 "row_count": resolution.row_count,
370 "contributions": [
371 {
372 "row_index": contribution.row_index,
373 "annual_hours": str(contribution.annual_hours),
374 "basis_quantity": str(contribution.basis_quantity),
375 "basis_unit": contribution.basis_unit,
376 "amount": str(contribution.amount),
377 "source_scenario": contribution.source_scenario_id,
378 "source_scenario_name": contribution.source_scenario_name,
379 "source_row_index": contribution.source_row_index,
380 "source_rule": contribution.source_rule_id,
381 "source_rule_label": contribution.source_rule_label,
382 }
383 for contribution in calculation.contributions
384 ],
385 }
389def _create_cash_flow_lines(*, result_run: EconomicsResultRun, rows) -> None:
390 """Replace discounted cash-flow rows because the row set is calculation-owned."""
391 result_run.lines.filter(group="discounted_cash_flow").delete()
392 for index, row in enumerate(rows, start=1000):
393 EconomicsResultLine.objects.create(
394 flowsheet_state=result_run.flowsheet_state,
395 result_run=result_run,
396 kind=ResultLineKind.CASH_FLOW,
397 group="discounted_cash_flow",
398 label=f"Year {row.year} discounted cash flow",
399 row_key=f"cash_flow.year_{row.year}",
400 amount=result_amount(row.present_value),
401 unit=result_run.result_currency,
402 source_row_key=f"cash_flow.year_{row.year}",
403 source_label=f"Year {row.year}",
404 warning_payload={
405 "cash_flow": str(row.cash_flow),
406 "discount_factor": str(row.discount_factor),
407 "present_value": str(row.present_value),
408 "cumulative_cash_flow": str(row.cumulative_cash_flow),
409 "cumulative_present_value": str(row.cumulative_present_value),
410 },
411 sort_order=index,
412 )
415def _line_warning_list(payload) -> list[dict[str, Any]]:
416 """Read normalized line warnings from source-line payloads defensively."""
417 if not isinstance(payload, dict): 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 return []
419 warnings = payload.get("warnings", [])
420 return warnings if isinstance(warnings, list) else []
423def _metric_formula_payload(metric: FinancialMetric, *, study) -> dict[str, Any] | None:
424 if metric.key == FinancialMetricKey.ELECTRICAL_UPGRADE:
425 try:
426 formula = build_electrical_upgrade_formula(study)
427 except FormulaError:
428 return None
429 return formula.formula.audit_payload(
430 FormulaEvaluation(value=metric.value, bindings=formula.bindings)
431 )
432 return metric.formula_audit
435def _capital_factor_list(payload) -> list[dict[str, Any]]:
436 """Read generated capital factor rows from source-line payloads defensively."""
437 if not isinstance(payload, dict): 437 ↛ 438line 437 didn't jump to line 438 because the condition on line 437 was never true
438 return []
439 factors = payload.get("capital_factors", [])
440 return factors if isinstance(factors, list) else []
442def _metric_sort_key(metric: FinancialMetric) -> tuple[int, str]:
443 """Return the stable presentation order for financial metric rows."""
444 return metric.sort_order, metric.key.value