Coverage for backend/django/Economics/costing/operating/schedule_calculation.py: 86%
87 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from __future__ import annotations
3from dataclasses import dataclass
4from decimal import Decimal
6from idaes_factory.unit_conversion.unit_conversion import pint_registry
8from Economics.costing.models import OperatingCostLine
9from Economics.costing.operating.line_calculation import AnnualBasisQuantity
10from Economics.formulas.builders.operating import (
11 OperatingLineFormula,
12 SUPPORTED_OPERATING_METHODS,
13 build_operating_line_formula,
14 resolve_operating_line_rate_basis,
15)
16from Economics.formulas.engine.core import FormulaError
17from Economics.scheduling.series import (
18 ScheduleSeriesResolution,
19 study_schedule_property_resolution,
20 study_schedule_row_durations,
21)
22from Economics.shared.choices import EconomicsScheduleMode, OperatingLineBasisQuantitySource
23from Economics.shared.unit_conversion import convert_quantity
24from Economics.studies.models import EconomicsStudy
27@dataclass(frozen=True)
28class OperatingScheduleContribution:
29 """One annualized contribution from a production-schedule row."""
31 row_index: int
32 annual_hours: Decimal
33 basis_quantity: Decimal
34 basis_unit: str
35 amount: Decimal
36 source_scenario_id: int | None = None
37 source_scenario_name: str = ""
38 source_row_index: int | None = None
39 source_rule_id: int | None = None
40 source_rule_label: str = ""
43@dataclass(frozen=True)
44class OperatingLineCalculation:
45 """Resolved annual amount and audit data for one operating line."""
47 annual_amount: Decimal | None
48 annual_basis: AnnualBasisQuantity | None
49 schedule_controlled: bool
50 schedule_resolution: ScheduleSeriesResolution | None
51 contributions: tuple[OperatingScheduleContribution, ...]
52 steady_formula: OperatingLineFormula | None
53 rate_amount: Decimal | None = None
54 rate_unit: str = ""
55 target_basis_unit: str = ""
56 blocked_reason: str = ""
59def calculate_operating_line(line: OperatingCostLine, *, study: EconomicsStudy) -> OperatingLineCalculation:
60 """Calculate one operating line, using production schedules when applicable."""
62 if _should_use_schedule(line=line, study=study):
63 scheduled = _calculate_scheduled_operating_line(line, study=study)
64 if scheduled is not None: 64 ↛ 67line 64 didn't jump to line 67 because the condition on line 64 was always true
65 return scheduled
67 try:
68 steady_formula = build_operating_line_formula(line, study=study)
69 except FormulaError as exc:
70 return OperatingLineCalculation(
71 annual_amount=None,
72 annual_basis=None,
73 schedule_controlled=False,
74 schedule_resolution=None,
75 contributions=(),
76 steady_formula=None,
77 blocked_reason=exc.message,
78 )
80 return OperatingLineCalculation(
81 annual_amount=steady_formula.evaluate(),
82 annual_basis=steady_formula.annual_basis,
83 schedule_controlled=False,
84 schedule_resolution=None,
85 contributions=(),
86 steady_formula=steady_formula,
87 rate_amount=steady_formula.rate_amount,
88 rate_unit=steady_formula.rate_unit,
89 target_basis_unit=steady_formula.target_basis_unit,
90 )
93def _calculate_scheduled_operating_line(
94 line: OperatingCostLine,
95 *,
96 study: EconomicsStudy,
97) -> OperatingLineCalculation | None:
98 resolution = study_schedule_property_resolution(study=study, property_info=line.source_property_info)
99 if not resolution.schedule_varying: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 return None
101 try:
102 rate_basis = resolve_operating_line_rate_basis(
103 line,
104 source_basis_unit=resolution.unit or line.basis_unit,
105 )
106 except FormulaError as exc:
107 return OperatingLineCalculation(
108 annual_amount=None,
109 annual_basis=None,
110 schedule_controlled=True,
111 schedule_resolution=resolution,
112 contributions=(),
113 steady_formula=None,
114 blocked_reason=exc.message,
115 )
116 if not resolution.points or any(point.value is None for point in resolution.points): 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 return OperatingLineCalculation(
118 annual_amount=None,
119 annual_basis=None,
120 schedule_controlled=True,
121 schedule_resolution=resolution,
122 contributions=(),
123 steady_formula=_steady_formula_or_none(line, study=study),
124 rate_amount=rate_basis.rate_amount,
125 rate_unit=rate_basis.rate_unit,
126 target_basis_unit=rate_basis.target_basis_unit,
127 blocked_reason=resolution.message
128 or "This operating line is not available for every operating state in the selected production schedule.",
129 )
131 annual_durations = {
132 duration.row_index: duration.annual_hours
133 for duration in study_schedule_row_durations(study=study)
134 }
135 if not annual_durations: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true
136 return OperatingLineCalculation(
137 annual_amount=None,
138 annual_basis=None,
139 schedule_controlled=True,
140 schedule_resolution=resolution,
141 contributions=(),
142 steady_formula=_steady_formula_or_none(line, study=study),
143 rate_amount=rate_basis.rate_amount,
144 rate_unit=rate_basis.rate_unit,
145 target_basis_unit=rate_basis.target_basis_unit,
146 blocked_reason="Annual operating hours are required to use a production schedule for operating costs.",
147 )
149 contributions: list[OperatingScheduleContribution] = []
150 annual_basis_quantity = Decimal("0")
151 annual_amount = Decimal("0")
152 for point in resolution.points:
153 annual_hours = annual_durations.get(point.row_index, Decimal("0"))
154 converted_basis = _scheduled_basis_quantity(
155 point.value,
156 source_unit=rate_basis.source_basis_unit,
157 target_unit=rate_basis.target_basis_unit,
158 annual_hours=annual_hours,
159 )
160 if converted_basis is None: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 return OperatingLineCalculation(
162 annual_amount=None,
163 annual_basis=None,
164 schedule_controlled=True,
165 schedule_resolution=resolution,
166 contributions=tuple(contributions),
167 steady_formula=_steady_formula_or_none(line, study=study),
168 rate_amount=rate_basis.rate_amount,
169 rate_unit=rate_basis.rate_unit,
170 target_basis_unit=rate_basis.target_basis_unit,
171 blocked_reason="This operating line cannot be converted for production-schedule costing.",
172 )
173 row_amount = converted_basis * rate_basis.rate_amount
174 annual_basis_quantity += converted_basis
175 annual_amount += row_amount
176 contributions.append(
177 OperatingScheduleContribution(
178 row_index=point.row_index,
179 annual_hours=annual_hours,
180 basis_quantity=converted_basis,
181 basis_unit=rate_basis.target_basis_unit,
182 amount=row_amount,
183 source_scenario_id=point.source_scenario_id,
184 source_scenario_name=point.source_scenario_name,
185 source_row_index=point.source_row_index,
186 source_rule_id=point.source_rule_id,
187 source_rule_label=point.source_rule_label,
188 )
189 )
191 return OperatingLineCalculation(
192 annual_amount=annual_amount,
193 annual_basis=AnnualBasisQuantity(
194 quantity=annual_basis_quantity,
195 unit=rate_basis.target_basis_unit,
196 ),
197 schedule_controlled=True,
198 schedule_resolution=resolution,
199 contributions=tuple(contributions),
200 steady_formula=_steady_formula_or_none(line, study=study),
201 rate_amount=rate_basis.rate_amount,
202 rate_unit=rate_basis.rate_unit,
203 target_basis_unit=rate_basis.target_basis_unit,
204 )
207def _should_use_schedule(*, line: OperatingCostLine, study: EconomicsStudy) -> bool:
208 schedule_selected = (
209 (study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id)
210 or study.schedule_mode == EconomicsScheduleMode.COMPOSITE
211 )
212 return bool(
213 schedule_selected
214 and line.source_property_info_id
215 and line.basis_quantity_source == OperatingLineBasisQuantitySource.SOURCE_PROPERTY
216 and line.calculation_method in SUPPORTED_OPERATING_METHODS
217 )
220def _steady_formula_or_none(line: OperatingCostLine, *, study: EconomicsStudy) -> OperatingLineFormula | None:
221 try:
222 return build_operating_line_formula(line, study=study)
223 except FormulaError:
224 return None
227def _scheduled_basis_quantity(
228 value: Decimal,
229 *,
230 source_unit: str,
231 target_unit: str,
232 annual_hours: Decimal,
233) -> Decimal | None:
234 if not source_unit or not target_unit: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 return value
236 integrated = convert_quantity(
237 value=value,
238 source_unit=source_unit,
239 target_unit=target_unit,
240 multiplier=pint_registry.Quantity(annual_hours, "hour"),
241 )
242 if integrated is not None: 242 ↛ 244line 242 didn't jump to line 244 because the condition on line 242 was always true
243 return integrated
244 return convert_quantity(
245 value=value,
246 source_unit=source_unit,
247 target_unit=target_unit,
248 )