Coverage for backend/django/Economics/costing/costable_items/items.py: 88%
155 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1from __future__ import annotations
3from dataclasses import dataclass
5from django.db import transaction
7from core.auxiliary.enums import ConType
8from core.auxiliary.models.PropertyInfo import PropertyInfo
9from core.auxiliary.models.PropertySet import PropertySet
10from core.auxiliary.models.PropertyValue import PropertyValue
11from Economics.shared.choices import (
12 CostBasis,
13 CostDriverSource,
14 CostableItemType,
15)
16from Economics.costing.models import CostDriver, CostableItem, EquipmentMapping
17from Economics.formulas.property_state import apply_economics_property_state
18from Economics.studies.models import EconomicsStudy
19from Economics.costing.cost_curves.registry import (
20 COST_DRIVER_RULES,
21 CostDriverRule,
22 ManualPropertySpec,
23 PreferredProperty,
24)
25from Economics.costing.cost_curves.catalog import CUSTOM_EQUIPMENT_CATEGORY
26from Economics.costing.cost_curves.driver_properties import preferred_property_matches_property_info
27from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
30@dataclass(frozen=True)
31class EnableCostingResult:
32 costable_item: CostableItem
33 cost_driver: CostDriver
34 supported: bool
37@dataclass(frozen=True)
38class DriverPropertyMatch:
39 property_info: PropertyInfo
40 preferred_property: PreferredProperty
43def enable_costing_for_simulation_object(
44 *,
45 study: EconomicsStudy,
46 simulation_object: SimulationObject,
47) -> EnableCostingResult:
48 """
49 Create or update the Economics rows that make a unit visible to costing.
51 The v1 registry is exact and source-controlled: it only considers objectType,
52 configured property keys, and unitType. It does not inspect display names or
53 connected streams looking for plausible drivers.
54 """
55 if study.flowsheet_id != simulation_object.flowsheet_id: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 raise ValueError("Cannot enable costing for a simulation object from another flowsheet.")
57 if simulation_object.objectType == "group":
58 raise ValueError("Flowsheet groups cannot be enabled as v1 economics costable items.")
60 with transaction.atomic():
61 rule = _get_rule_for_object_type(simulation_object.objectType)
62 costable_item = _get_or_create_costable_item(study, simulation_object, supported=rule is not None)
64 if rule is None:
65 cost_driver = _set_unsupported_cost_driver(costable_item, simulation_object)
66 return EnableCostingResult(costable_item=costable_item, cost_driver=cost_driver, supported=False)
68 property_set = _get_or_create_property_set(simulation_object)
69 manual_property = _get_or_create_manual_property(property_set, rule.manual_property)
70 for alternate_manual_property in rule.alternate_manual_properties:
71 _get_or_create_manual_property(property_set, alternate_manual_property)
72 equipment_mapping = _get_or_create_equipment_mapping(costable_item, rule)
73 driver_property = _find_matching_driver_property(
74 simulation_object,
75 property_set,
76 rule,
77 equipment_category=equipment_mapping.equipment_category,
78 )
80 if rule.manual_only:
81 cost_driver = _set_manual_cost_driver(costable_item, manual_property, rule)
82 elif driver_property is not None:
83 cost_driver = _set_property_cost_driver(
84 costable_item,
85 driver_property.property_info,
86 manual_property,
87 rule,
88 preferred_property=driver_property.preferred_property,
89 )
90 elif rule.unresolved_reason_code == "manual_override_required":
91 cost_driver = _set_manual_cost_driver(costable_item, manual_property, rule)
92 else:
93 cost_driver = _set_unresolved_cost_driver(costable_item, manual_property, rule)
95 return EnableCostingResult(costable_item=costable_item, cost_driver=cost_driver, supported=True)
98def _get_rule_for_object_type(object_type: str) -> CostDriverRule | None:
99 for rule in COST_DRIVER_RULES:
100 if object_type in rule.compatible_object_types:
101 return rule
102 return None
105def _get_or_create_costable_item(
106 study: EconomicsStudy,
107 simulation_object: SimulationObject,
108 *,
109 supported: bool,
110) -> CostableItem:
111 costable_item, created = CostableItem.objects.get_or_create(
112 study=study,
113 simulation_object=simulation_object,
114 defaults={
115 "flowsheet": study.flowsheet,
116 "item_type": CostableItemType.SIMULATION_OBJECT,
117 "name": simulation_object.componentName,
118 "included": True,
119 "manual": not supported,
120 },
121 )
122 update_fields = []
123 if costable_item.flowsheet_id != study.flowsheet_id: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 raise ValueError("Existing costable item is not in the study flowsheet.")
125 if costable_item.name != simulation_object.componentName: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 costable_item.name = simulation_object.componentName
127 update_fields.append("name")
128 if costable_item.item_type != CostableItemType.SIMULATION_OBJECT: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 costable_item.item_type = CostableItemType.SIMULATION_OBJECT
130 update_fields.append("item_type")
131 was_manual = costable_item.manual
132 if supported and costable_item.manual is True:
133 costable_item.manual = False
134 update_fields.append("manual")
135 if supported and was_manual and costable_item.included is False:
136 costable_item.included = True
137 update_fields.append("included")
138 if not supported and created is False and costable_item.manual is False: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 costable_item.manual = True
140 update_fields.append("manual")
141 if update_fields:
142 update_fields.append("updated_at")
143 costable_item.save(update_fields=update_fields)
144 return costable_item
147def _get_or_create_equipment_mapping(costable_item: CostableItem, rule: CostDriverRule) -> EquipmentMapping:
148 equipment_mapping, _ = EquipmentMapping.objects.get_or_create(
149 costable_item=costable_item,
150 defaults={
151 "flowsheet": costable_item.flowsheet,
152 "equipment_category": _default_equipment_category(rule),
153 "equipment_subtype": "",
154 "cost_basis": CostBasis.PURCHASE,
155 "applicability_notes": "Created from economics capital-line configuration.",
156 },
157 )
158 return equipment_mapping
161def _default_equipment_category(rule: CostDriverRule) -> str:
162 return rule.equipment_category or (
163 rule.allowed_equipment_categories[0] if rule.allowed_equipment_categories else "equipment"
164 )
167def _get_or_create_property_set(simulation_object: SimulationObject) -> PropertySet:
168 property_set, _ = PropertySet.objects.get_or_create(
169 simulationObject=simulation_object,
170 defaults={"flowsheet": simulation_object.flowsheet},
171 )
172 return property_set
175def _get_or_create_manual_property(property_set: PropertySet, spec: ManualPropertySpec) -> PropertyInfo:
176 property_info = PropertyInfo.objects.filter(set=property_set, key=spec.key, index=0).first()
177 if property_info is None:
178 property_info = PropertyInfo.objects.create(
179 flowsheet=property_set.flowsheet,
180 set=property_set,
181 type="numeric",
182 unitType=spec.unit_type,
183 unit=spec.canonical_unit,
184 key=spec.key,
185 displayName=spec.display_name,
186 index=0,
187 )
188 PropertyValue.objects.create(
189 flowsheet=property_set.flowsheet,
190 property=property_info,
191 value=None,
192 displayValue=None,
193 enabled=True,
194 )
195 else:
196 updates = []
197 for field_name, expected_value in (
198 ("type", "numeric"),
199 ("unitType", spec.unit_type),
200 ("unit", spec.canonical_unit),
201 ("displayName", spec.display_name),
202 ):
203 if getattr(property_info, field_name) != expected_value: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 setattr(property_info, field_name, expected_value)
205 updates.append(field_name)
206 if updates: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 property_info.save(update_fields=updates)
208 if not property_info.values.exists(): 208 ↛ 209line 208 didn't jump to line 209 because the condition on line 208 was never true
209 PropertyValue.objects.create(
210 flowsheet=property_set.flowsheet,
211 property=property_info,
212 value=None,
213 displayValue=None,
214 enabled=True,
215 )
217 apply_economics_property_state(property_info, editable=True)
218 return property_info
221def _find_matching_driver_property(
222 simulation_object: SimulationObject,
223 property_set: PropertySet,
224 rule: CostDriverRule,
225 *,
226 equipment_category: str,
227) -> DriverPropertyMatch | None:
228 for preferred_property in rule.preferred_properties:
229 property_info = (
230 property_set.containedProperties.filter(
231 key=preferred_property.key,
232 unitType=preferred_property.unit_type,
233 index=0,
234 )
235 .prefetch_related("values")
236 .first()
237 )
238 if (
239 property_info is not None
240 and property_info.has_value()
241 and preferred_property_matches_property_info(
242 preferred_property=preferred_property,
243 property_info=property_info,
244 rule=rule,
245 equipment_category=equipment_category,
246 )
247 ):
248 return DriverPropertyMatch(property_info=property_info, preferred_property=preferred_property)
249 return _find_unambiguous_input_stream_driver_property(
250 simulation_object,
251 rule,
252 equipment_category=equipment_category,
253 )
256def _find_unambiguous_input_stream_driver_property(
257 simulation_object: SimulationObject,
258 rule: CostDriverRule,
259 *,
260 equipment_category: str,
261) -> DriverPropertyMatch | None:
262 if not rule.preferred_input_stream_properties:
263 return None
264 matching_properties: list[DriverPropertyMatch] = []
265 inlet_ports = simulation_object.ports.filter(direction=ConType.Inlet).select_related("stream")
266 for port in inlet_ports:
267 stream = port.stream
268 if stream is None: 268 ↛ 269line 268 didn't jump to line 269 because the condition on line 268 was never true
269 continue
270 property_set = getattr(stream, "properties", None)
271 if property_set is None: 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true
272 continue
273 for preferred_property in rule.preferred_input_stream_properties:
274 property_info = (
275 property_set.containedProperties.filter(
276 key=preferred_property.key,
277 unitType=preferred_property.unit_type,
278 index=0,
279 )
280 .prefetch_related("values")
281 .first()
282 )
283 if (
284 property_info is not None
285 and property_info.has_value()
286 and preferred_property_matches_property_info(
287 preferred_property=preferred_property,
288 property_info=property_info,
289 rule=rule,
290 equipment_category=equipment_category,
291 )
292 ):
293 matching_properties.append(
294 DriverPropertyMatch(property_info=property_info, preferred_property=preferred_property)
295 )
296 if len(matching_properties) == 1:
297 return matching_properties[0]
298 return None
301def _set_property_cost_driver(
302 costable_item: CostableItem,
303 driver_property: PropertyInfo,
304 manual_property: PropertyInfo,
305 rule: CostDriverRule,
306 *,
307 preferred_property: PreferredProperty | None = None,
308) -> CostDriver:
309 canonical_unit = preferred_property.canonical_unit if preferred_property else None
310 curve_input_variable = preferred_property.curve_input_variable if preferred_property else None
311 cost_driver, _ = CostDriver.objects.update_or_create(
312 costable_item=costable_item,
313 defaults={
314 "flowsheet": costable_item.flowsheet,
315 "source": CostDriverSource.PROPERTY,
316 "property_info": driver_property,
317 "manual_property_info": manual_property,
318 "sizing_mode": "current",
319 "canonical_unit": canonical_unit or rule.canonical_unit,
320 "design_value": None,
321 "unresolved_reason_code": "",
322 "warning_payload": _warning_payload(
323 rule,
324 warnings=[],
325 design_value_basis="auto_selected_unit_or_unambiguous_input_stream_property",
326 curve_input_variable=curve_input_variable,
327 ),
328 },
329 )
330 return cost_driver
333def _set_manual_cost_driver(
334 costable_item: CostableItem,
335 manual_property: PropertyInfo,
336 rule: CostDriverRule,
337) -> CostDriver:
338 cost_driver, _ = CostDriver.objects.update_or_create(
339 costable_item=costable_item,
340 defaults={
341 "flowsheet": costable_item.flowsheet,
342 "source": CostDriverSource.MANUAL_OVERRIDE,
343 "property_info": None,
344 "manual_property_info": manual_property,
345 "sizing_mode": "manual",
346 "canonical_unit": rule.canonical_unit,
347 "design_value": None,
348 "unresolved_reason_code": rule.unresolved_reason_code,
349 "warning_payload": _warning_payload(rule, warnings=[rule.warning_text]),
350 },
351 )
352 return cost_driver
355def _set_unresolved_cost_driver(
356 costable_item: CostableItem,
357 manual_property: PropertyInfo,
358 rule: CostDriverRule,
359) -> CostDriver:
360 cost_driver, _ = CostDriver.objects.update_or_create(
361 costable_item=costable_item,
362 defaults={
363 "flowsheet": costable_item.flowsheet,
364 "source": CostDriverSource.UNRESOLVED,
365 "property_info": None,
366 "manual_property_info": manual_property,
367 "sizing_mode": "",
368 "canonical_unit": rule.canonical_unit,
369 "design_value": None,
370 "unresolved_reason_code": rule.unresolved_reason_code,
371 "warning_payload": _warning_payload(rule, warnings=[rule.warning_text]),
372 },
373 )
374 return cost_driver
377def _warning_payload(
378 rule: CostDriverRule,
379 *,
380 warnings: list[str],
381 design_value_basis: str | None = None,
382 curve_input_variable: str | None = None,
383) -> dict:
384 payload = {
385 "equipment_category": rule.equipment_category,
386 "curve_input_variable": curve_input_variable or rule.curve_input_variable,
387 "warnings": warnings,
388 }
389 if design_value_basis:
390 payload["design_value_basis"] = design_value_basis
391 if rule.allowed_equipment_categories:
392 payload["allowed_equipment_categories"] = list(rule.allowed_equipment_categories)
393 if rule.requires_equipment_category_selection:
394 payload["requires_equipment_category_selection"] = True
395 return payload
398def _set_unsupported_cost_driver(
399 costable_item: CostableItem,
400 simulation_object: SimulationObject,
401) -> CostDriver:
402 equipment_mapping, _ = EquipmentMapping.objects.get_or_create(
403 costable_item=costable_item,
404 defaults={
405 "flowsheet": costable_item.flowsheet,
406 "equipment_category": CUSTOM_EQUIPMENT_CATEGORY,
407 "equipment_subtype": "",
408 "cost_basis": CostBasis.PURCHASE,
409 "applicability_notes": "Created from partially supported economics capital-line configuration.",
410 },
411 )
412 if not equipment_mapping.equipment_category: 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 equipment_mapping.equipment_category = CUSTOM_EQUIPMENT_CATEGORY
414 equipment_mapping.save(update_fields=["equipment_category", "updated_at"])
415 cost_driver, _ = CostDriver.objects.update_or_create(
416 costable_item=costable_item,
417 defaults={
418 "flowsheet": costable_item.flowsheet,
419 "source": CostDriverSource.UNRESOLVED,
420 "property_info": None,
421 "manual_property_info": None,
422 "sizing_mode": "",
423 "canonical_unit": "",
424 "design_value": None,
425 "unresolved_reason_code": "unit_unsupported",
426 "warning_payload": {
427 "supported": False,
428 "partially_supported": True,
429 "equipment_category": CUSTOM_EQUIPMENT_CATEGORY,
430 "unit_object_type": simulation_object.objectType,
431 "warnings": [
432 "No default sizing template is available for this unit operation yet. "
433 "Select a custom cost curve and enter the sizing value manually."
434 ],
435 },
436 },
437 )
438 return cost_driver