Coverage for backend/django/Economics/costing/capital/electrical_upgrade.py: 94%
126 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Electrical-upgrade peak-demand helpers.
3The electrical upgrade is a capital allowance sized from peak electrical
4demand, not annual operating energy. Generated equipment capital lines carry a
5work-derived demand floor plus an editable selected peak demand; the aggregate
6project upgrade uses the selected peak demand across included capital lines.
7"""
9from __future__ import annotations
11from dataclasses import dataclass
12from decimal import Decimal, InvalidOperation
14from core.auxiliary.models.PropertyInfo import PropertyInfo
15from Economics.costing.capital.capital_line_sources import GENERATED_CAPITAL_LINE_SOURCE
16from Economics.costing.models import CapitalCostLine
17from Economics.studies.models import EconomicsStudy
18from Economics.shared.choices import EconomicsScheduleMode
19from Economics.shared.unit_conversion import convert_quantity
20from Economics.costing.operating.stream_properties import UNIT_POWER_WORK_SOURCE_KIND, unit_energy_source_kind
21from Economics.scheduling.series import study_schedule_property_resolution
24PEAK_DEMAND_UNIT = "kW"
25PEAK_DEMAND_QUANTUM = Decimal("0.00000001")
28@dataclass(frozen=True)
29class PeakDemandContribution:
30 capital_line_id: int
31 label: str
32 selected_peak_demand_kw: Decimal
33 minimum_peak_demand_kw: Decimal | None
36@dataclass(frozen=True)
37class PeakDemandBasis:
38 quantity_kw: Decimal | None
39 unit: str
40 contributions: tuple[PeakDemandContribution, ...]
41 blocked_reason: str = ""
44@dataclass(frozen=True)
45class ScheduledPeakDemand:
46 schedule_applies: bool
47 quantity_kw: Decimal | None = None
48 blocked_reason: str = ""
51class PeakDemandScheduleError(ValueError):
52 """Raised when schedule-controlled peak demand cannot be calculated."""
54 code = "missing_schedule_peak_demand"
56 def __init__(self, message: str):
57 super().__init__(message)
58 self.message = message
61def peak_demand_for_unit_capital_line(line: CapitalCostLine) -> Decimal | None:
62 """Return the editable peak demand basis for one generated capital line."""
63 return line.peak_demand_kw.quantize(PEAK_DEMAND_QUANTUM) if line.peak_demand_kw is not None else None
66def derive_peak_demand_basis(study: EconomicsStudy) -> PeakDemandBasis:
67 """Sum included capital-line peak demand values for upgrade sizing."""
68 contributions: list[PeakDemandContribution] = []
69 blocked_reason = _included_peak_demand_blocked_reason(study)
70 if blocked_reason:
71 return PeakDemandBasis(
72 quantity_kw=None,
73 unit=PEAK_DEMAND_UNIT,
74 contributions=(),
75 blocked_reason=blocked_reason,
76 )
77 lines = list(
78 study.capital_lines.filter(
79 included=True,
80 peak_demand_kw__isnull=False,
81 )
82 )
83 for line in lines:
84 demand = peak_demand_for_unit_capital_line(line)
85 if demand is None or demand <= 0:
86 continue
87 contributions.append(
88 PeakDemandContribution(
89 capital_line_id=line.pk,
90 label=line.label,
91 selected_peak_demand_kw=demand,
92 minimum_peak_demand_kw=line.minimum_peak_demand_kw,
93 )
94 )
95 total = sum((row.selected_peak_demand_kw for row in contributions), Decimal("0")).quantize(
96 PEAK_DEMAND_QUANTUM
97 )
98 return PeakDemandBasis(
99 quantity_kw=total if total > 0 or lines else None,
100 unit=PEAK_DEMAND_UNIT,
101 contributions=tuple(contributions),
102 )
105def unit_work_peak_demand_kw(
106 costable_item,
107 *,
108 study: EconomicsStudy | None = None,
109 bulk_property_values: bool = False,
110) -> Decimal | None:
111 """Return positive work-property capacity for a costable unit in kW.
113 Multiple work properties on the unit are additive. This mirrors the result
114 resource classification for work properties but deliberately uses the raw
115 flowsheet capacity value rather than annualized operating hours. A
116 work-capable unit with no solved work value returns a zero floor so the UI
117 can still expose an editable peak-demand field for that unit.
118 """
119 simulation_object = getattr(costable_item, "simulation_object", None)
120 property_set = getattr(simulation_object, "properties", None)
121 if simulation_object is None or property_set is None:
122 return None
123 total = Decimal("0")
124 has_work_property = False
125 property_infos = getattr(property_set, "_economics_contained_properties", None)
126 if property_infos is None:
127 property_infos = property_set.containedProperties.all()
128 for property_info in property_infos:
129 if not _is_peak_demand_work_property(property_info):
130 continue
131 has_work_property = True
132 peak_kw = _work_property_peak_kw(
133 property_info,
134 study=study,
135 bulk_property_values=bulk_property_values,
136 )
137 if peak_kw is None:
138 continue
139 total += peak_kw
140 if total > 0:
141 return total.quantize(PEAK_DEMAND_QUANTUM)
142 return Decimal("0").quantize(PEAK_DEMAND_QUANTUM) if has_work_property else None
145def _is_peak_demand_work_property(property_info: PropertyInfo) -> bool:
146 unit = (property_info.unit or "").strip()
147 if not unit: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 return False
149 source_kind = unit_energy_source_kind(
150 unit=property_info.set.simulationObject,
151 property_info=property_info,
152 )
153 return source_kind == UNIT_POWER_WORK_SOURCE_KIND
156def _work_property_peak_kw(
157 property_info: PropertyInfo,
158 *,
159 study: EconomicsStudy | None = None,
160 bulk_property_values: bool = False,
161) -> Decimal | None:
162 if not _is_peak_demand_work_property(property_info): 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true
163 return None
164 unit = (property_info.unit or "").strip()
165 scheduled_peak = _scheduled_work_property_peak_kw(property_info, study=study, unit=unit)
166 if scheduled_peak.schedule_applies:
167 if scheduled_peak.blocked_reason:
168 raise PeakDemandScheduleError(scheduled_peak.blocked_reason)
169 return scheduled_peak.quantity_kw
170 try:
171 value = property_info.get_value_bulk() if bulk_property_values else property_info.get_value()
172 raw_value = Decimal(str(value))
173 except (InvalidOperation, TypeError, ValueError):
174 return None
175 if raw_value == 0:
176 return None
177 converted = convert_quantity(
178 value=abs(raw_value),
179 source_unit=unit,
180 target_unit=PEAK_DEMAND_UNIT,
181 )
182 if converted is None or converted <= 0: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 return None
184 return converted.quantize(PEAK_DEMAND_QUANTUM)
187def _scheduled_work_property_peak_kw(
188 property_info: PropertyInfo,
189 *,
190 study: EconomicsStudy | None,
191 unit: str,
192) -> ScheduledPeakDemand:
193 if (
194 study is None
195 or not (
196 (study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id)
197 or study.schedule_mode == EconomicsScheduleMode.COMPOSITE
198 )
199 or not unit
200 ):
201 return ScheduledPeakDemand(schedule_applies=False)
202 resolution = study_schedule_property_resolution(study=study, property_info=property_info)
203 if not resolution.schedule_varying:
204 return ScheduledPeakDemand(schedule_applies=False)
205 if not resolution.points: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 return ScheduledPeakDemand(
207 schedule_applies=True,
208 blocked_reason=resolution.message
209 or "Scheduled work values are required before calculating peak demand.",
210 )
211 converted_values = []
212 for point in resolution.points:
213 if point.value is None:
214 return ScheduledPeakDemand(
215 schedule_applies=True,
216 blocked_reason=resolution.message
217 or "Scheduled work values are required before calculating peak demand.",
218 )
219 converted = convert_quantity(
220 value=abs(point.value),
221 source_unit=resolution.unit or unit,
222 target_unit=PEAK_DEMAND_UNIT,
223 )
224 if converted is None: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 return ScheduledPeakDemand(
226 schedule_applies=True,
227 blocked_reason="Scheduled work values cannot be converted for peak-demand sizing.",
228 )
229 if converted <= 0:
230 continue
231 converted_values.append(converted)
232 if not converted_values:
233 return ScheduledPeakDemand(
234 schedule_applies=True,
235 quantity_kw=Decimal("0").quantize(PEAK_DEMAND_QUANTUM),
236 )
237 return ScheduledPeakDemand(
238 schedule_applies=True,
239 quantity_kw=max(converted_values).quantize(PEAK_DEMAND_QUANTUM),
240 )
243def _included_peak_demand_blocked_reason(study: EconomicsStudy) -> str:
244 lines = study.capital_lines.filter(
245 included=True,
246 source=GENERATED_CAPITAL_LINE_SOURCE,
247 ).only("warning_payload", "costable_item_id", "cost_curve_id")
248 for line in lines:
249 payload = line.warning_payload if isinstance(line.warning_payload, dict) else {}
250 reason = payload.get("peak_demand_blocked_reason")
251 if reason:
252 return str(reason)
253 return ""