Coverage for backend/django/Economics/results/services/lifecycle/result_lines.py: 90%
117 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +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 import FinancialMetric
20from Economics.formulas.engine.core import FormulaError, FormulaEvaluation
21from Economics.formulas.builders.capital import build_electrical_upgrade_formula
22from Economics.formulas.builders.operating import (
23 PROCESS_ENERGY_QUANTITY_UNIT,
24 build_operating_line_formula,
25 build_process_energy_contribution_formula,
26)
27from Economics.costing.operating.line_calculation import AnnualBasisQuantity
28from Economics.costing.operating.resource_basis import (
29 RESOURCE_CATEGORY_ANNUAL_COOLING_DUTY,
30 RESOURCE_CATEGORY_ANNUAL_HEATING_DUTY,
31 operating_resource_breakdown_category,
32 operating_resource_source_kind,
33)
34from Economics.results.services.lifecycle.common import EconomicsContract
35from Economics.shared.payloads import result_amount
38RESULT_RESOURCE_QUANTITY_QUANTUM = Decimal("0.00000001")
41class OperatingResourcePayload(EconomicsContract):
42 """Source-resource fields persisted on operating result lines."""
43 source_property_info: PropertyInfo | None = None
44 resource_source_kind: str = ""
45 resource_source_object_name: str = ""
46 resource_source_object_type: str = ""
47 resource_property_name: str = ""
48 resource_breakdown_category: str = ""
49 annual_basis_quantity: Decimal | None = None
50 annual_basis_unit: str = ""
53def _create_metric_lines(*, result_run: EconomicsResultRun, metrics: dict[str, FinancialMetric]) -> None:
54 """Upsert financial metric rows while removing metrics no longer emitted."""
55 metric_row_keys = {f"metric.{metric.key}" for metric in metrics.values()}
56 result_run.lines.filter(group="financial_metrics").exclude(row_key__in=metric_row_keys).delete()
57 for sort_order, metric in enumerate(sorted(metrics.values(), key=_metric_sort_key), start=10):
58 EconomicsResultLine.objects.update_or_create(
59 flowsheet=result_run.flowsheet,
60 result_run=result_run,
61 row_key=f"metric.{metric.key}",
62 defaults={
63 "kind": _metric_line_kind(metric.key),
64 "group": "financial_metrics",
65 "label": _metric_label(metric.key),
66 "amount": result_amount(metric.value),
67 "unit": metric.unit,
68 "source_row_key": f"metric.{metric.key}",
69 "source_label": _metric_label(metric.key),
70 "warning_payload": {
71 "status": metric.status,
72 "raw_value": str(metric.value) if metric.value is not None else None,
73 "assumptions": metric.assumptions.to_mapping(),
74 "formula_audit": _metric_formula_payload(metric, study=result_run.study),
75 },
76 "sort_order": sort_order,
77 },
78 )
81def _create_capital_result_lines(*, result_run: EconomicsResultRun) -> None:
82 """Upsert included capital-line detail rows for a presentation result run."""
83 capital_lines = result_run.study.capital_lines.filter(included=True, amount__isnull=False).select_related(
84 "costable_item",
85 "cost_curve",
86 )
87 for sort_order, line in enumerate(capital_lines, start=100):
88 EconomicsResultLine.objects.update_or_create(
89 flowsheet=result_run.flowsheet,
90 result_run=result_run,
91 row_key=f"capital_line.{line.pk}",
92 defaults={
93 "kind": ResultLineKind.CAPITAL,
94 "group": "capital_lines",
95 "label": line.label,
96 "amount": result_amount(line.amount),
97 "unit": line.currency,
98 "source_costable_item": line.costable_item,
99 "source_cost_curve": line.cost_curve,
100 "source_capital_line": line,
101 "source_row_key": f"capital_line:{line.pk}",
102 "source_label": line.label,
103 "source_note": line.source,
104 "warning_payload": {
105 "line_type": line.line_type,
106 "calculation_basis": line.calculation_basis,
107 "basis_percent": str(line.basis_percent) if line.basis_percent is not None else None,
108 "peak_demand_kw": str(line.peak_demand_kw) if line.peak_demand_kw is not None else None,
109 "minimum_peak_demand_kw": str(line.minimum_peak_demand_kw)
110 if line.minimum_peak_demand_kw is not None
111 else None,
112 "manual": line.manual,
113 "confidence": line.confidence,
114 "capital_factors": _capital_factor_list(line.warning_payload),
115 "warnings": _line_warning_list(line.warning_payload),
116 },
117 "sort_order": sort_order,
118 },
119 )
122def _create_depreciation_result_lines(*, result_run: EconomicsResultRun) -> None:
123 """Replace annual straight-line depreciation rows for this result run."""
124 result_run.lines.filter(group="depreciation_lines").delete()
125 schedule = build_straight_line_depreciation_schedule(result_run.study)
126 for sort_order, line in enumerate(schedule.lines, start=400):
127 EconomicsResultLine.objects.create(
128 flowsheet=result_run.flowsheet,
129 result_run=result_run,
130 kind=ResultLineKind.DEPRECIATION,
131 group="depreciation_lines",
132 label=line.label,
133 row_key=f"depreciation_line.{line.capital_line_id}",
134 amount=result_amount(line.annual_depreciation),
135 unit=f"{result_run.result_currency}/year",
136 source_capital_line_id=line.capital_line_id,
137 source_row_key=f"capital_line:{line.capital_line_id}",
138 source_label=line.label,
139 warning_payload={
140 "depreciation_mode": line.depreciation_mode,
141 "depreciable_basis": str(line.depreciable_basis),
142 "life_years": line.life_years,
143 "salvage_percent": str(line.salvage_percent),
144 "annual_depreciation": str(line.annual_depreciation),
145 },
146 sort_order=sort_order,
147 )
150def _create_operating_result_lines(*, result_run: EconomicsResultRun) -> None:
151 """Upsert included operating-line detail rows with resource breakdown fields."""
152 operating_lines = result_run.study.operating_lines.filter(included=True).select_related(
153 "costable_item",
154 "source_default_rate",
155 "source_property_info__set__simulationObject",
156 )
157 for sort_order, line in enumerate(operating_lines, start=500):
158 try:
159 operating_formula = build_operating_line_formula(line, study=result_run.study)
160 annual_amount = operating_formula.evaluate()
161 except FormulaError:
162 annual_amount = None
163 if annual_amount is None: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 continue
165 resource_payload = _operating_resource_payload(
166 line=line,
167 study=result_run.study,
168 annual_basis=operating_formula.annual_basis,
169 )
170 resource_defaults = resource_payload.model_dump()
171 EconomicsResultLine.objects.update_or_create(
172 flowsheet=result_run.flowsheet,
173 result_run=result_run,
174 row_key=f"operating_line.{line.pk}",
175 defaults={
176 "kind": ResultLineKind.OPERATING,
177 "group": "operating_lines",
178 "label": line.label,
179 "amount": result_amount(annual_amount),
180 "unit": f"{line.currency}/year",
181 "source_costable_item": line.costable_item,
182 "source_operating_line": line,
183 "source_row_key": f"operating_line:{line.pk}",
184 "source_label": line.label,
185 "source_note": line.source,
186 "warning_payload": {
187 "line_type": line.line_type,
188 "category": line.category,
189 "economic_effect": line.economic_effect,
190 "manual": line.manual,
191 "basis_quantity": str(line.basis_quantity) if line.basis_quantity is not None else None,
192 "basis_unit": line.basis_unit,
193 "rate_amount": str(line.rate_amount) if line.rate_amount is not None else None,
194 "rate_unit": line.rate_unit,
195 "calculation_method": line.calculation_method,
196 "formula_audit": operating_formula.formula.audit_payload(
197 FormulaEvaluation(
198 value=annual_amount,
199 bindings={
200 operating_formula.formula.inputs[0].key: operating_formula.basis_quantity,
201 },
202 conversion_diagnostics=(
203 {
204 "input": operating_formula.formula.inputs[0].key,
205 "source_value": str(operating_formula.basis_quantity),
206 "source_unit": operating_formula.basis_unit,
207 "target_value": str(operating_formula.annual_basis.quantity),
208 "target_unit": operating_formula.annual_basis.unit,
209 },
210 ),
211 )
212 ),
213 "source_default_rate": line.source_default_rate.key if line.source_default_rate_id else "",
214 "outlet_stream_disposition": line.outlet_stream_disposition,
215 "warnings": _line_warning_list(line.warning_payload),
216 },
217 **resource_defaults,
218 "sort_order": sort_order,
219 },
220 )
223def _operating_resource_payload(
224 *,
225 line: OperatingCostLine,
226 study: EconomicsStudy,
227 annual_basis: AnnualBasisQuantity | None,
228) -> OperatingResourcePayload:
229 """Build the operating resource audit fields shared by tables and chart drilldowns."""
230 source_object = None
231 property_info = line.source_property_info if line.source_property_info_id else None
232 if line.source_property_info_id:
233 source_object = line.source_property_info.set.simulationObject
234 source_kind = operating_resource_source_kind(line=line, source_object=source_object, property_info=property_info)
235 resource_category = operating_resource_breakdown_category(
236 line=line,
237 source_kind=source_kind,
238 )
239 display_annual_basis = _resource_display_annual_basis(
240 line=line,
241 study=study,
242 resource_category=resource_category,
243 fallback=annual_basis,
244 )
245 payload = OperatingResourcePayload(
246 source_property_info=property_info,
247 resource_source_kind=source_kind,
248 resource_source_object_name=getattr(source_object, "componentName", "") or "",
249 resource_source_object_type=getattr(source_object, "objectType", "") or "",
250 resource_property_name=getattr(property_info, "displayName", "") or "",
251 resource_breakdown_category=resource_category,
252 )
253 if display_annual_basis is not None: 253 ↛ 263line 253 didn't jump to line 263 because the condition on line 253 was always true
254 payload = payload.model_copy(
255 update={
256 "annual_basis_quantity": display_annual_basis.quantity.quantize(
257 RESULT_RESOURCE_QUANTITY_QUANTUM,
258 rounding=ROUND_HALF_UP,
259 ),
260 "annual_basis_unit": display_annual_basis.unit,
261 }
262 )
263 return payload
266def _resource_display_annual_basis(
267 *,
268 line: OperatingCostLine,
269 study: EconomicsStudy,
270 resource_category: str,
271 fallback: AnnualBasisQuantity | None,
272) -> AnnualBasisQuantity | None:
273 if resource_category not in {
274 RESOURCE_CATEGORY_ANNUAL_COOLING_DUTY,
275 RESOURCE_CATEGORY_ANNUAL_HEATING_DUTY,
276 }:
277 return fallback
278 try:
279 contribution_formula = build_process_energy_contribution_formula(line, study=study)
280 annual_energy = contribution_formula.evaluate()
281 except FormulaError:
282 annual_energy = None
283 if annual_energy is None: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 return fallback
285 return AnnualBasisQuantity(quantity=annual_energy, unit=PROCESS_ENERGY_QUANTITY_UNIT)
289def _create_cash_flow_lines(*, result_run: EconomicsResultRun, rows) -> None:
290 """Replace discounted cash-flow rows because the row set is calculation-owned."""
291 result_run.lines.filter(group="discounted_cash_flow").delete()
292 for index, row in enumerate(rows, start=1000):
293 EconomicsResultLine.objects.create(
294 flowsheet=result_run.flowsheet,
295 result_run=result_run,
296 kind=ResultLineKind.CASH_FLOW,
297 group="discounted_cash_flow",
298 label=f"Year {row.year} discounted cash flow",
299 row_key=f"cash_flow.year_{row.year}",
300 amount=result_amount(row.present_value),
301 unit=result_run.result_currency,
302 source_row_key=f"cash_flow.year_{row.year}",
303 source_label=f"Year {row.year}",
304 warning_payload={
305 "cash_flow": str(row.cash_flow),
306 "discount_factor": str(row.discount_factor),
307 "present_value": str(row.present_value),
308 "cumulative_cash_flow": str(row.cumulative_cash_flow),
309 "cumulative_present_value": str(row.cumulative_present_value),
310 },
311 sort_order=index,
312 )
315def _line_warning_list(payload) -> list[dict[str, Any]]:
316 """Read normalized line warnings from source-line payloads defensively."""
317 if not isinstance(payload, dict): 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true
318 return []
319 warnings = payload.get("warnings", [])
320 return warnings if isinstance(warnings, list) else []
323def _metric_formula_payload(metric: FinancialMetric, *, study) -> dict[str, Any] | None:
324 if metric.key == "electrical_upgrade":
325 try:
326 formula = build_electrical_upgrade_formula(study)
327 except FormulaError:
328 return None
329 return formula.formula.audit_payload(
330 FormulaEvaluation(value=metric.value, bindings=formula.bindings)
331 )
332 return metric.formula_audit
335def _capital_factor_list(payload) -> list[dict[str, Any]]:
336 """Read generated capital factor rows from source-line payloads defensively."""
337 if not isinstance(payload, dict): 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 return []
339 factors = payload.get("capital_factors", [])
340 return factors if isinstance(factors, list) else []
344def _metric_line_kind(metric_key: str) -> str:
345 """Map financial metric keys to result-line kinds used by result tables."""
346 if metric_key in {
347 "capex",
348 "purchase_basis_equipment",
349 "installed_basis_equipment",
350 "contingency",
351 "electrical_upgrade",
352 }:
353 return ResultLineKind.CAPITAL
354 if metric_key in {"annual_depreciation", "depreciation_tax_shield"}:
355 return ResultLineKind.DEPRECIATION
356 if metric_key == "peak_demand":
357 return ResultLineKind.FINANCIAL_METRIC
358 if metric_key in {"annual_opex", "annual_revenue", "annual_profit"}:
359 return ResultLineKind.OPERATING
360 return ResultLineKind.FINANCIAL_METRIC
363def _metric_sort_key(metric: FinancialMetric) -> tuple[int, str]:
364 """Return the stable presentation order for financial metric rows."""
365 order = {
366 "capex": 0,
367 "purchase_basis_equipment": 1,
368 "installed_basis_equipment": 2,
369 "contingency": 3,
370 "electrical_upgrade": 4,
371 "peak_demand": 5,
372 "annual_depreciation": 6,
373 "depreciation_tax_shield": 7,
374 "annual_opex": 10,
375 "annual_revenue": 11,
376 "annual_profit": 12,
377 "after_tax_annual_cash_flow": 13,
378 }
379 return order.get(metric.key, 100), metric.key
382def _metric_label(metric_key: str) -> str:
383 """Return user-facing labels for metrics that differ from title-cased keys."""
384 labels = {
385 "annual_opex": "Annual expenses",
386 "annual_revenue": "Annual revenue",
387 "annual_profit": "Annual profit",
388 "annual_depreciation": "Annual equipment depreciation",
389 "depreciation_tax_shield": "Depreciation tax shield",
390 "after_tax_annual_cash_flow": "After-tax annual cash flow",
391 "purchase_basis_equipment": "Purchased basis equipment",
392 "installed_basis_equipment": "Installed basis equipment",
393 "electrical_upgrade": "Electrical upgrade",
394 "peak_demand": "Peak demand",
395 }
396 if metric_key in labels:
397 return labels[metric_key]
398 return metric_key.replace("_", " ").title()