Coverage for backend/django/Economics/costing/line_properties/sync.py: 85%
152 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1from __future__ import annotations
3from decimal import Decimal
5from core.auxiliary.models.PropertyInfo import PropertyInfo
6from core.auxiliary.models.PropertySet import PropertySet
7from core.auxiliary.models.PropertyValue import PropertyValue
8from django.core.exceptions import ObjectDoesNotExist
9from django.db import transaction
10from Economics.costing.models import CapitalCostLine, OperatingCostLine
11from Economics.shared.choices import CapitalLineBasis
12from Economics.formulas.models import EconomicsLineFormula
13from Economics.formulas.property_state import apply_economics_property_state
14from Economics.studies.models import EconomicsStudy
15from Economics.settings_profiles.services.settings_profiles import get_settings_profile
16from Economics.costing.capital.capital_line_sources import GENERATED_CAPITAL_LINE_SOURCE
17from Economics.formulas.builders.capital import build_custom_capital_line_formula, build_generated_unit_capex_subtotal_formula
18from Economics.formulas.engine.core import FormulaError
19from Economics.formulas.builders.native_property_formulas import (
20 NativePropertyExpression,
21 generated_capital_line_property_expression,
22 operating_line_property_expression,
23)
24from Economics.costing.line_properties.references import CAPITAL_LINE_KIND, OPERATING_LINE_KIND, native_property_reference
27def sync_economics_line_properties_for_study(study: EconomicsStudy) -> dict[str, int]:
28 """Materialize generated properties and formula rows for economics line items."""
30 with transaction.atomic():
31 values: dict[str, int] = {}
32 active_field_keys: set[str] = set()
33 for line in study.capital_lines.select_related("costable_item__simulation_object", "cost_curve").order_by("pk"):
34 field_key = _line_field_key(CAPITAL_LINE_KIND, line.pk)
35 active_field_keys.add(field_key)
36 values[field_key] = _sync_capital_line_property(study=study, line=line).pk
37 for line in (
38 study.operating_lines.select_related(
39 "costable_item__simulation_object",
40 "source_property_info__set__simulationObject",
41 "source_default_rate",
42 ).order_by("pk")
43 ):
44 field_key = _line_field_key(OPERATING_LINE_KIND, line.pk)
45 active_field_keys.add(field_key)
46 values[field_key] = _sync_operating_line_property(study=study, line=line).pk
47 _delete_stale_line_formulas(study, active_field_keys=active_field_keys)
48 return values
51def _sync_capital_line_property(*, study: EconomicsStudy, line: CapitalCostLine) -> PropertyValue:
52 target = _capital_line_target(study=study, line=line)
53 expression = _capital_line_expression(study=study, line=line)
54 value = _materialize_line_property(
55 study=study,
56 property_set=target,
57 field_key=_line_field_key(CAPITAL_LINE_KIND, line.pk),
58 property_key="economics.capital_line",
59 display_name=line.label,
60 unit_type="currency",
61 unit=line.currency or _study_currency(study),
62 expression=expression,
63 )
64 _persist_line_formula(
65 study=study,
66 property_value=value,
67 line_key=_line_field_key(CAPITAL_LINE_KIND, line.pk),
68 formula_key=f"capital_line:{line.pk}",
69 formula=expression,
70 capital_line=line,
71 )
72 return value
75def _sync_operating_line_property(*, study: EconomicsStudy, line: OperatingCostLine) -> PropertyValue:
76 target = _operating_line_target(study=study, line=line)
77 expression = operating_line_property_expression(line, study=study)
78 value = _materialize_line_property(
79 study=study,
80 property_set=target,
81 field_key=_line_field_key(OPERATING_LINE_KIND, line.pk),
82 property_key="economics.operating_line",
83 display_name=line.label,
84 unit_type="costRate",
85 unit=f"{line.currency or _study_currency(study)}/year",
86 expression=expression,
87 )
88 _persist_line_formula(
89 study=study,
90 property_value=value,
91 line_key=_line_field_key(OPERATING_LINE_KIND, line.pk),
92 formula_key=f"operating_line:{line.pk}",
93 formula=expression,
94 operating_line=line,
95 )
96 return value
99def _capital_line_expression(*, study: EconomicsStudy, line: CapitalCostLine) -> NativePropertyExpression:
100 if line.source == GENERATED_CAPITAL_LINE_SOURCE:
101 return generated_capital_line_property_expression(line)
102 if line.amount is None and line.calculation_basis != CapitalLineBasis.BASE_CAPEX_PERCENT: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 return NativePropertyExpression("", False, f"`{line.label}` has no amount.")
104 generated_subtotal = build_generated_unit_capex_subtotal_formula(study).evaluate()
105 if generated_subtotal is None:
106 generated_subtotal = Decimal("0")
107 try:
108 formula = build_custom_capital_line_formula(line, base_capex=generated_subtotal)
109 render_bindings = {}
110 base_capex_reference = native_property_reference(study, "base_capital_cost")
111 if line.calculation_basis == CapitalLineBasis.BASE_CAPEX_PERCENT and base_capex_reference:
112 render_bindings["custom_capex_percentage_basis"] = base_capex_reference
113 return NativePropertyExpression(
114 formula.render_property_formula(render_bindings),
115 True,
116 value=formula.evaluate(),
117 )
118 except FormulaError as exc:
119 return NativePropertyExpression("", False, f"`{line.label}` {exc.message}")
122def _materialize_line_property(
123 *,
124 study: EconomicsStudy,
125 property_set: PropertySet,
126 field_key: str,
127 property_key: str,
128 display_name: str,
129 unit_type: str,
130 unit: str,
131 expression: NativePropertyExpression,
132) -> PropertyValue:
133 property_info = _line_property_info(
134 study=study,
135 property_set=property_set,
136 field_key=field_key,
137 property_key=property_key,
138 display_name=display_name,
139 unit_type=unit_type,
140 unit=unit,
141 )
142 value = _single_scalar_value(property_info)
143 display_value = None if expression.value is None else str(expression.value)
144 changed_fields = []
145 if value.value != display_value:
146 value.value = display_value
147 value.displayValue = display_value
148 changed_fields.extend(["value", "displayValue"])
149 if value.formula != expression.formula:
150 value.formula = expression.formula
151 changed_fields.append("formula")
152 if changed_fields:
153 value.save(update_fields=changed_fields)
154 apply_economics_property_state(
155 property_info,
156 editable=False,
157 formula_incomplete=not expression.solve_visible,
158 formula_incomplete_reason=expression.blocked_reason,
159 )
160 return value
163def _persist_line_formula(
164 *,
165 study: EconomicsStudy,
166 property_value: PropertyValue,
167 line_key: str,
168 formula_key: str,
169 formula: NativePropertyExpression,
170 capital_line: CapitalCostLine | None = None,
171 operating_line: OperatingCostLine | None = None,
172) -> None:
173 status = "calculated" if formula.value is not None and not formula.blocked_reason else "unavailable"
174 EconomicsLineFormula.objects.update_or_create(
175 flowsheet=study.flowsheet,
176 study=study,
177 line_key=line_key,
178 defaults={
179 "property_value": property_value,
180 "capital_line": capital_line,
181 "operating_line": operating_line,
182 "formula_key": formula_key,
183 "formula": formula.formula,
184 "property_formula": formula.formula,
185 "unit": property_value.property.unit,
186 "value": str(formula.value) if formula.value is not None else None,
187 "status": status,
188 "formula_audit": {
189 "formula_key": formula_key,
190 "formula": formula.formula,
191 "value": str(formula.value) if formula.value is not None else None,
192 },
193 "blocked_reason": formula.blocked_reason,
194 },
195 )
198def _line_property_info(
199 *,
200 study: EconomicsStudy,
201 property_set: PropertySet,
202 field_key: str,
203 property_key: str,
204 display_name: str,
205 unit_type: str,
206 unit: str,
207) -> PropertyInfo:
208 formula_record = (
209 EconomicsLineFormula.objects.filter(
210 flowsheet=study.flowsheet,
211 study=study,
212 line_key=field_key,
213 property_value__isnull=False,
214 )
215 .select_related("property_value__property")
216 .order_by("pk")
217 .first()
218 )
219 property_info = formula_record.property_value.property if formula_record is not None else None
220 defaults = {
221 "set": property_set,
222 "type": "numeric",
223 "unitType": unit_type,
224 "unit": unit,
225 "displayName": display_name,
226 "index": 0,
227 }
228 if property_info is None:
229 return PropertyInfo.objects.create(
230 flowsheet=study.flowsheet,
231 key=property_key,
232 **defaults,
233 )
234 changed_fields = []
235 for field_name, value in defaults.items():
236 if getattr(property_info, field_name) != value: 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true
237 setattr(property_info, field_name, value)
238 changed_fields.append(field_name)
239 if property_info.key != property_key: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 property_info.key = property_key
241 changed_fields.append("key")
242 if changed_fields: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 property_info.save(update_fields=changed_fields)
244 return property_info
247def _single_scalar_value(property_info: PropertyInfo) -> PropertyValue:
248 values = list(property_info.values.order_by("pk"))
249 value = values[0] if values else None
250 for duplicate in values[1:]: 250 ↛ 251line 250 didn't jump to line 251 because the loop on line 250 never started
251 duplicate.delete()
252 if value is None:
253 value = PropertyValue.objects.create(
254 flowsheet=property_info.flowsheet,
255 property=property_info,
256 value=None,
257 displayValue=None,
258 enabled=True,
259 )
260 return value
263def _capital_line_target(*, study: EconomicsStudy, line: CapitalCostLine) -> PropertySet:
264 simulation_object = None
265 if not line.manual and line.costable_item_id:
266 simulation_object = line.costable_item.simulation_object
267 if simulation_object is None:
268 return _root_property_set(study)
269 return _property_set_for_object(study, simulation_object)
272def _operating_line_target(*, study: EconomicsStudy, line: OperatingCostLine) -> PropertySet:
273 simulation_object = None
274 if not line.manual and line.source_property_info_id:
275 try:
276 simulation_object = line.source_property_info.set.simulationObject
277 except ObjectDoesNotExist:
278 simulation_object = None
279 if simulation_object is None and not line.manual and line.costable_item_id:
280 simulation_object = line.costable_item.simulation_object
281 if simulation_object is None:
282 return _root_property_set(study)
283 return _property_set_for_object(study, simulation_object)
286def _root_property_set(study: EconomicsStudy) -> PropertySet:
287 root_grouping = getattr(study.flowsheet, "rootGrouping", None)
288 try:
289 root_object = getattr(root_grouping, "simulationObject", None)
290 except ObjectDoesNotExist:
291 root_object = None
292 if root_object is not None:
293 return _property_set_for_object(study, root_object)
294 property_set = PropertySet.objects.filter(flowsheet=study.flowsheet, simulationObject__isnull=True).order_by("pk").first()
295 if property_set is not None: 295 ↛ 297line 295 didn't jump to line 297 because the condition on line 295 was always true
296 return property_set
297 return PropertySet.objects.create(flowsheet=study.flowsheet, simulationObject=None)
300def _property_set_for_object(study: EconomicsStudy, simulation_object) -> PropertySet:
301 property_set, _ = PropertySet.objects.get_or_create(
302 flowsheet=study.flowsheet,
303 simulationObject=simulation_object,
304 )
305 return property_set
308def _delete_stale_line_formulas(study: EconomicsStudy, *, active_field_keys: set[str]) -> None:
309 stale_formulas = list(
310 study.line_formulas.exclude(line_key__in=active_field_keys)
311 .select_related("property_value__property")
312 .order_by("pk")
313 )
314 for formula in stale_formulas: 314 ↛ 315line 314 didn't jump to line 315 because the loop on line 314 never started
315 if formula.property_value_id and formula.property_value.property_id:
316 formula.property_value.property.delete()
317 study.line_formulas.exclude(line_key__in=active_field_keys).delete()
320def _line_field_key(line_kind: str, line_id: int) -> str:
321 return f"{line_kind}_line:{line_id}"
324def _study_currency(study: EconomicsStudy) -> str:
325 assumptions = get_settings_profile(study)
326 if assumptions is None:
327 return "NZD"
328 return assumptions.currency or "NZD"