Coverage for backend/django/Economics/costing/line_properties/sync.py: 82%
167 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +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 formula_text = formula.render_property_formula(render_bindings)
114 if line.calculation_basis != CapitalLineBasis.BASE_CAPEX_PERCENT:
115 formula_text = _unit_literal(
116 formula.evaluate() or Decimal("0"),
117 formula.formula.unit,
118 )
119 return NativePropertyExpression(
120 formula_text,
121 True,
122 value=formula.evaluate(),
123 )
124 except FormulaError as exc:
125 return NativePropertyExpression("", False, f"`{line.label}` {exc.message}")
128def _unit_literal(value: Decimal, unit: str) -> str:
129 from core.auxiliary.formula_units import formula_unit_expression
131 unit_expression = formula_unit_expression(unit)
132 if not unit_expression: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 return format(value, "f")
134 if value == Decimal("0"):
135 return "0"
136 if value == Decimal("1"): 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 return f"({unit_expression})"
138 if value == Decimal("-1"): 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 return f"-({unit_expression})"
140 return f"({format(value, 'f')} * ({unit_expression}))"
143def _materialize_line_property(
144 *,
145 study: EconomicsStudy,
146 property_set: PropertySet,
147 field_key: str,
148 property_key: str,
149 display_name: str,
150 unit_type: str,
151 unit: str,
152 expression: NativePropertyExpression,
153) -> PropertyValue:
154 property_info = _line_property_info(
155 study=study,
156 property_set=property_set,
157 field_key=field_key,
158 property_key=property_key,
159 display_name=display_name,
160 unit_type=unit_type,
161 unit=unit,
162 )
163 value = _single_scalar_value(property_info)
164 display_value = None if expression.value is None else str(expression.value)
165 changed_fields = []
166 if value.value != display_value:
167 value.value = display_value
168 value.displayValue = display_value
169 changed_fields.extend(["value", "displayValue"])
170 if value.formula != expression.formula:
171 value.formula = expression.formula
172 changed_fields.append("formula")
173 if changed_fields:
174 value.save(update_fields=changed_fields)
175 apply_economics_property_state(
176 property_info,
177 editable=False,
178 formula_incomplete=not expression.solve_visible,
179 formula_incomplete_reason=expression.blocked_reason,
180 )
181 return value
184def _persist_line_formula(
185 *,
186 study: EconomicsStudy,
187 property_value: PropertyValue,
188 line_key: str,
189 formula_key: str,
190 formula: NativePropertyExpression,
191 capital_line: CapitalCostLine | None = None,
192 operating_line: OperatingCostLine | None = None,
193) -> None:
194 status = "calculated" if formula.value is not None and not formula.blocked_reason else "unavailable"
195 EconomicsLineFormula.objects.update_or_create(
196 flowsheet_state=study.flowsheet_state,
197 study=study,
198 line_key=line_key,
199 defaults={
200 "property_value": property_value,
201 "capital_line": capital_line,
202 "operating_line": operating_line,
203 "formula_key": formula_key,
204 "formula": formula.formula,
205 "property_formula": formula.formula,
206 "unit": property_value.property.unit,
207 "value": str(formula.value) if formula.value is not None else None,
208 "status": status,
209 "formula_audit": {
210 "formula_key": formula_key,
211 "formula": formula.formula,
212 "value": str(formula.value) if formula.value is not None else None,
213 },
214 "blocked_reason": formula.blocked_reason,
215 },
216 )
219def _line_property_info(
220 *,
221 study: EconomicsStudy,
222 property_set: PropertySet,
223 field_key: str,
224 property_key: str,
225 display_name: str,
226 unit_type: str,
227 unit: str,
228) -> PropertyInfo:
229 formula_record = (
230 EconomicsLineFormula.objects.filter(
231 flowsheet_state=study.flowsheet_state,
232 study=study,
233 line_key=field_key,
234 property_value__isnull=False,
235 )
236 .select_related("property_value__property")
237 .order_by("pk")
238 .first()
239 )
240 property_info = formula_record.property_value.property if formula_record is not None else None
241 defaults = {
242 "set": property_set,
243 "type": "numeric",
244 "unitType": unit_type,
245 "unit": unit,
246 "displayName": display_name,
247 "index": 0,
248 }
249 if property_info is None:
250 return PropertyInfo.objects.create(
251 flowsheet_state=study.flowsheet_state,
252 key=property_key,
253 **defaults,
254 )
255 changed_fields = []
256 for field_name, value in defaults.items():
257 if getattr(property_info, field_name) != value: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 setattr(property_info, field_name, value)
259 changed_fields.append(field_name)
260 if property_info.key != property_key: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 property_info.key = property_key
262 changed_fields.append("key")
263 if changed_fields: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 property_info.save(update_fields=changed_fields)
265 return property_info
268def _single_scalar_value(property_info: PropertyInfo) -> PropertyValue:
269 values = list(property_info.values.order_by("pk"))
270 value = values[0] if values else None
271 for duplicate in values[1:]: 271 ↛ 272line 271 didn't jump to line 272 because the loop on line 271 never started
272 duplicate.delete()
273 if value is None:
274 value = PropertyValue.objects.create(
275 flowsheet_state=property_info.flowsheet_state,
276 property=property_info,
277 value=None,
278 displayValue=None,
279 enabled=True,
280 )
281 return value
284def _capital_line_target(*, study: EconomicsStudy, line: CapitalCostLine) -> PropertySet:
285 simulation_object = None
286 if not line.manual and line.costable_item_id:
287 simulation_object = line.costable_item.simulation_object
288 if simulation_object is None:
289 return _root_property_set(study)
290 return _property_set_for_object(study, simulation_object)
293def _operating_line_target(*, study: EconomicsStudy, line: OperatingCostLine) -> PropertySet:
294 simulation_object = None
295 if not line.manual and line.source_property_info_id:
296 try:
297 simulation_object = line.source_property_info.set.simulationObject
298 except ObjectDoesNotExist:
299 simulation_object = None
300 if simulation_object is None and not line.manual and line.costable_item_id:
301 simulation_object = line.costable_item.simulation_object
302 if simulation_object is None:
303 return _root_property_set(study)
304 return _property_set_for_object(study, simulation_object)
307def _root_property_set(study: EconomicsStudy) -> PropertySet:
308 root_grouping = study.flowsheet_state.root_grouping
309 try:
310 root_object = getattr(root_grouping, "simulationObject", None)
311 except ObjectDoesNotExist:
312 root_object = None
313 if root_object is not None: 313 ↛ 315line 313 didn't jump to line 315 because the condition on line 313 was always true
314 return _property_set_for_object(study, root_object)
315 property_set = PropertySet.objects.filter(flowsheet_state=study.flowsheet_state, simulationObject__isnull=True).order_by("pk").first()
316 if property_set is not None:
317 return property_set
318 return PropertySet.objects.create(flowsheet_state=study.flowsheet_state, simulationObject=None)
321def _property_set_for_object(study: EconomicsStudy, simulation_object) -> PropertySet:
322 property_set, _ = PropertySet.objects.get_or_create(
323 flowsheet_state=study.flowsheet_state,
324 simulationObject=simulation_object,
325 )
326 return property_set
329def _delete_stale_line_formulas(study: EconomicsStudy, *, active_field_keys: set[str]) -> None:
330 stale_formulas = list(
331 study.line_formulas.exclude(line_key__in=active_field_keys)
332 .select_related("property_value__property")
333 .order_by("pk")
334 )
335 for formula in stale_formulas: 335 ↛ 336line 335 didn't jump to line 336 because the loop on line 335 never started
336 if formula.property_value_id and formula.property_value.property_id:
337 formula.property_value.property.delete()
338 study.line_formulas.exclude(line_key__in=active_field_keys).delete()
341def _line_field_key(line_kind: str, line_id: int) -> str:
342 return f"{line_kind}_line:{line_id}"
345def _study_currency(study: EconomicsStudy) -> str:
346 assumptions = get_settings_profile(study)
347 if assumptions is None:
348 return "NZD"
349 return assumptions.currency or "NZD"