Coverage for backend/django/Economics/results/services/resource_metrics.py: 86%
107 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Shared resource-use summaries for Economics result runs."""
3from __future__ import annotations
5from decimal import Decimal, ROUND_HALF_UP
6from enum import StrEnum
7from typing import Iterable
9from pydantic import ConfigDict
11from Economics.costing.operating.resource_classification import (
12 RESOURCE_CLASSIFICATION_ELECTRICITY,
13 RESOURCE_CLASSIFICATION_NATURAL_GAS,
14 RESOURCE_CLASSIFICATION_STEAM_ELECTRICITY,
15 RESOURCE_CLASSIFICATION_STEAM_NATURAL_GAS,
16)
17from Economics.costing.operating.resource_basis import converted_annual_resource_quantity
18from Economics.results.models import EconomicsResultLine
19from Economics.results.services.comparison.contracts import (
20 ComparisonMetricValue,
21 ComparisonStatus,
22 EconomicsComparisonContract,
23)
24from Economics.results.services.financial_metrics.metric_catalog import MetricComparisonDirection
27NZD_PER_YEAR = "NZD/year"
28RESOURCE_QUANTUM = Decimal("0.00000001")
29ZERO = Decimal("0")
32class ResourceMetricStatus(StrEnum):
33 """Availability status for a single-result high-level resource metric."""
35 AVAILABLE = "available"
36 UNAVAILABLE = "unavailable"
37 UNIT_MISMATCH = "unit_mismatch"
40class ResourceMetricSpec(EconomicsComparisonContract):
41 """Display and comparison metadata for a high-level resource metric."""
43 model_config = ConfigDict(frozen=True)
45 row_key: str
46 label: str
47 unit: str
48 maximum_fraction_digits: int = 2
49 comparison_direction: MetricComparisonDirection = MetricComparisonDirection.LOWER_IS_BETTER
52class ResourceMetricPayload(EconomicsComparisonContract):
53 """JSON-facing high-level resource metric for one result run."""
55 row_key: str
56 label: str
57 amount: str | None
58 unit: str
59 status: ResourceMetricStatus
60 maximum_fraction_digits: int
61 comparison_direction: MetricComparisonDirection
64RESOURCE_ELECTRICITY_QUANTITY = "resource.electricity.quantity"
65RESOURCE_ELECTRICITY_COST = "resource.electricity.cost"
66RESOURCE_NATURAL_GAS_QUANTITY = "resource.natural_gas.quantity"
67RESOURCE_NATURAL_GAS_COST = "resource.natural_gas.cost"
69RESOURCE_METRIC_SPECS: tuple[ResourceMetricSpec, ...] = (
70 ResourceMetricSpec(
71 row_key=RESOURCE_ELECTRICITY_QUANTITY,
72 label="Electricity use",
73 unit="MWh/year",
74 ),
75 ResourceMetricSpec(
76 row_key=RESOURCE_ELECTRICITY_COST,
77 label="Electricity cost",
78 unit=NZD_PER_YEAR,
79 maximum_fraction_digits=0,
80 ),
81 ResourceMetricSpec(
82 row_key=RESOURCE_NATURAL_GAS_QUANTITY,
83 label="Natural gas use",
84 unit="GJ/year",
85 ),
86 ResourceMetricSpec(
87 row_key=RESOURCE_NATURAL_GAS_COST,
88 label="Natural gas cost",
89 unit=NZD_PER_YEAR,
90 maximum_fraction_digits=0,
91 ),
92)
95def resource_metric_specs() -> tuple[ResourceMetricSpec, ...]:
96 """Return the stable high-level resource metric display order."""
97 return RESOURCE_METRIC_SPECS
100def resource_metric_values_for_lines(
101 lines: Iterable[EconomicsResultLine],
102) -> dict[str, ComparisonMetricValue]:
103 """Aggregate electricity and natural-gas quantities/costs from operating result lines."""
104 totals = {
105 RESOURCE_ELECTRICITY_QUANTITY: ZERO,
106 RESOURCE_ELECTRICITY_COST: ZERO,
107 RESOURCE_NATURAL_GAS_QUANTITY: ZERO,
108 RESOURCE_NATURAL_GAS_COST: ZERO,
109 }
110 mismatched_cost_rows: set[str] = set()
111 mismatched_quantity_rows: set[str] = set()
112 cost_units: dict[str, str] = {}
114 for line in lines:
115 classification = line.resource_classification
116 if classification == RESOURCE_CLASSIFICATION_ELECTRICITY:
117 _add_quantity(
118 totals=totals,
119 mismatched_quantity_rows=mismatched_quantity_rows,
120 row_key=RESOURCE_ELECTRICITY_QUANTITY,
121 quantity=_converted_resource_metric_quantity(line, "MWh/year"),
122 )
123 _add_cost(
124 totals=totals,
125 cost_units=cost_units,
126 mismatched_cost_rows=mismatched_cost_rows,
127 row_key=RESOURCE_ELECTRICITY_COST,
128 line=line,
129 )
130 continue
131 if classification == RESOURCE_CLASSIFICATION_NATURAL_GAS:
132 _add_quantity(
133 totals=totals,
134 mismatched_quantity_rows=mismatched_quantity_rows,
135 row_key=RESOURCE_NATURAL_GAS_QUANTITY,
136 quantity=_converted_resource_metric_quantity(line, "GJ/year"),
137 )
138 _add_cost(
139 totals=totals,
140 cost_units=cost_units,
141 mismatched_cost_rows=mismatched_cost_rows,
142 row_key=RESOURCE_NATURAL_GAS_COST,
143 line=line,
144 )
145 continue
146 if classification == RESOURCE_CLASSIFICATION_STEAM_ELECTRICITY:
147 _add_quantity(
148 totals=totals,
149 mismatched_quantity_rows=mismatched_quantity_rows,
150 row_key=RESOURCE_ELECTRICITY_QUANTITY,
151 quantity=_converted_resource_metric_quantity(line, "MWh/year"),
152 )
153 _add_cost(
154 totals=totals,
155 cost_units=cost_units,
156 mismatched_cost_rows=mismatched_cost_rows,
157 row_key=RESOURCE_ELECTRICITY_COST,
158 line=line,
159 )
160 continue
161 if classification == RESOURCE_CLASSIFICATION_STEAM_NATURAL_GAS:
162 _add_quantity(
163 totals=totals,
164 mismatched_quantity_rows=mismatched_quantity_rows,
165 row_key=RESOURCE_NATURAL_GAS_QUANTITY,
166 quantity=_converted_resource_metric_quantity(line, "GJ/year"),
167 )
168 _add_cost(
169 totals=totals,
170 cost_units=cost_units,
171 mismatched_cost_rows=mismatched_cost_rows,
172 row_key=RESOURCE_NATURAL_GAS_COST,
173 line=line,
174 )
176 specs_by_key = {spec.row_key: spec for spec in resource_metric_specs()}
177 values: dict[str, ComparisonMetricValue] = {}
178 for row_key, total in totals.items():
179 spec = specs_by_key[row_key]
180 comparison_status = (
181 ComparisonStatus.UNIT_MISMATCH
182 if row_key in mismatched_cost_rows or row_key in mismatched_quantity_rows
183 else None
184 )
185 values[row_key] = ComparisonMetricValue(
186 amount=(
187 None
188 if comparison_status
189 in {ComparisonStatus.UNIT_MISMATCH, ComparisonStatus.UNAVAILABLE}
190 else total.quantize(RESOURCE_QUANTUM, rounding=ROUND_HALF_UP)
191 ),
192 unit=cost_units.get(row_key, spec.unit),
193 comparison_status=comparison_status,
194 )
195 return values
198def resource_metric_payloads_for_lines(
199 lines: Iterable[EconomicsResultLine],
200) -> list[ResourceMetricPayload]:
201 """Return JSON-friendly resource metrics for a single result run."""
202 values = resource_metric_values_for_lines(lines)
203 payloads: list[ResourceMetricPayload] = []
204 for spec in resource_metric_specs():
205 value = values[spec.row_key]
206 payloads.append(
207 ResourceMetricPayload(
208 row_key=spec.row_key,
209 label=spec.label,
210 amount=_format_resource_amount(value.amount),
211 unit=value.unit,
212 status=(
213 ResourceMetricStatus.UNIT_MISMATCH
214 if value.comparison_status == ComparisonStatus.UNIT_MISMATCH
215 else ResourceMetricStatus.UNAVAILABLE
216 if value.comparison_status == ComparisonStatus.UNAVAILABLE
217 else ResourceMetricStatus.AVAILABLE
218 ),
219 maximum_fraction_digits=spec.maximum_fraction_digits,
220 comparison_direction=spec.comparison_direction,
221 )
222 )
223 return payloads
226def _add_quantity(
227 *,
228 totals: dict[str, Decimal],
229 mismatched_quantity_rows: set[str],
230 row_key: str,
231 quantity: Decimal | None,
232) -> None:
233 if quantity is None: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 mismatched_quantity_rows.add(row_key)
235 return
236 totals[row_key] += quantity
239def _add_cost(
240 *,
241 totals: dict[str, Decimal],
242 cost_units: dict[str, str],
243 mismatched_cost_rows: set[str],
244 row_key: str,
245 line: EconomicsResultLine,
246) -> None:
247 expected_unit = _annual_currency_unit_for_line(line)
248 if line.amount is None or expected_unit is None or line.unit != expected_unit: 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true
249 mismatched_cost_rows.add(row_key)
250 return
251 existing_unit = cost_units.setdefault(row_key, line.unit)
252 if existing_unit != line.unit: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 mismatched_cost_rows.add(row_key)
254 return
255 totals[row_key] += line.amount
258def _annual_currency_unit_for_line(line: EconomicsResultLine) -> str | None:
259 """Return the annual cost unit snapshot persisted on the result line."""
260 if not line.unit: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 return None
262 if "/" not in line.unit: 262 ↛ 263line 262 didn't jump to line 263 because the condition on line 262 was never true
263 return None
264 denominator = line.unit.rsplit("/", 1)[1].strip()
265 if denominator not in {"y", "yr", "year", "years"}: 265 ↛ 266line 265 didn't jump to line 266 because the condition on line 265 was never true
266 return None
267 return line.unit
270def _converted_resource_metric_quantity(line: EconomicsResultLine, target_unit: str) -> Decimal | None:
271 if line.resource_metric_quantity is None or not line.resource_metric_unit: 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true
272 return None
273 return converted_annual_resource_quantity(
274 annual_basis_quantity=line.resource_metric_quantity,
275 annual_basis_unit=line.resource_metric_unit,
276 target_unit=target_unit,
277 )
280def _format_resource_amount(amount: Decimal | None) -> str | None:
281 if amount is None: 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true
282 return None
283 return format(amount, "f")