Coverage for backend/django/Economics/costing/operating/stream_properties.py: 89%
265 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
4from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
6from django.core.exceptions import ObjectDoesNotExist
8from core.auxiliary.enums import ConType
9from core.auxiliary.models.PropertyInfo import PropertyInfo
10from Economics.shared.choices import (
11 DefaultRateType,
12 OperatingLineBasisQuantitySource,
13 OperatingLineCategory,
14 OperatingLineEconomicEffect,
15 OperatingLineRateSourceMode,
16 OutletStreamDisposition,
17)
18from Economics.studies.models import EconomicsStudy
19from Economics.costing.models import OperatingCostLine
20from Economics.costing.operating.line_calculation import (
21 operating_line_rate_defaults_for_category,
22)
23from Economics.settings_profiles.services.settings_profiles import get_settings_profile
24from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
27STREAM_OPERATING_LINE_SOURCE = "selected_output_stream_property"
28UNIT_POWER_WORK_SOURCE_KIND = "unit_power_work_property"
29UNIT_HEATING_DUTY_SOURCE_KIND = "unit_heating_duty_property"
30UNIT_COOLING_DUTY_SOURCE_KIND = "unit_cooling_duty_property"
31WORK_PROPERTY_KEYS = frozenset(
32 {
33 "activepower",
34 "chargingpowerin",
35 "chargingpowerout",
36 "importexport",
37 "inpower",
38 "power",
39 "powertransfer",
40 "workelectrical",
41 "workmechanical",
42 }
43)
44HEATING_DUTY_PROPERTY_KEYS = frozenset({"heatadded", "heatdemand", "sink.heat"})
45COOLING_DUTY_PROPERTY_KEYS = frozenset({"heatdutyinverted", "heatremoved", "source.heat"})
46EXTERNAL_DUTY_OBJECT_TYPES = frozenset(
47 {
48 "heatexchanger",
49 "heatexchanger1d",
50 "heatexchangerlc",
51 "heatexchangerntu",
52 "plateheatexchanger",
53 }
54)
55OPERATING_LINE_DECIMAL_QUANTUM = Decimal("0.00000001")
58@dataclass(frozen=True)
59class OperatingStreamPropertyOption:
60 property_info: int
61 stream_id: int
62 stream_name: str
63 source_object_id: int
64 source_object_name: str
65 source_kind: str
66 property_key: str
67 display_name: str
68 unit: str
69 unit_type: str
70 value_preview: str
71 has_value: bool
72 suggested_group: str
73 suggested_category: str
74 suggested_disposition: str
75 selected_operating_line: int | None = None
78def output_stream_property_options(study: EconomicsStudy) -> list[OperatingStreamPropertyOption]:
79 """Return scalar numeric properties that can seed operating lines."""
80 selected_line_by_property = {
81 line.source_property_info_id: line.pk
82 for line in study.operating_lines.filter(
83 source=STREAM_OPERATING_LINE_SOURCE,
84 source_property_info__isnull=False,
85 )
86 }
87 options: list[OperatingStreamPropertyOption] = []
88 seen_properties: set[int] = set()
89 for stream, source_kind, group, category, disposition in _suggested_streams(study):
90 for property_info in _eligible_stream_properties(stream):
91 if not _property_matches_group(property_info, group):
92 continue
93 seen_properties.add(property_info.pk)
94 value = property_info.get_value()
95 options.append(
96 OperatingStreamPropertyOption(
97 property_info=property_info.pk,
98 stream_id=stream.pk,
99 stream_name=stream.componentName or f"Stream {stream.pk}",
100 source_object_id=stream.pk,
101 source_object_name=stream.componentName or f"Stream {stream.pk}",
102 source_kind=source_kind,
103 property_key=property_info.key,
104 display_name=property_info.displayName,
105 unit=property_info.unit or "",
106 unit_type=property_info.unitType or "",
107 value_preview="" if value in (None, "") else str(value),
108 has_value=property_info.has_value(),
109 suggested_group=group,
110 suggested_category=category,
111 suggested_disposition=disposition,
112 selected_operating_line=selected_line_by_property.get(property_info.pk),
113 )
114 )
115 for unit in _unit_operations(study):
116 for property_info in _eligible_stream_properties(unit):
117 source_kind = unit_energy_source_kind(unit=unit, property_info=property_info)
118 if property_info.pk in seen_properties or source_kind is None:
119 continue
120 seen_properties.add(property_info.pk)
121 value = property_info.get_value()
122 options.append(
123 OperatingStreamPropertyOption(
124 property_info=property_info.pk,
125 stream_id=unit.pk,
126 stream_name=unit.componentName or f"Unit {unit.pk}",
127 source_object_id=unit.pk,
128 source_object_name=unit.componentName or f"Unit {unit.pk}",
129 source_kind=source_kind,
130 property_key=property_info.key,
131 display_name=property_info.displayName,
132 unit=property_info.unit or "",
133 unit_type=property_info.unitType or "",
134 value_preview="" if value in (None, "") else str(value),
135 has_value=property_info.has_value(),
136 suggested_group="energy",
137 suggested_category=OperatingLineCategory.ENERGY,
138 suggested_disposition="",
139 selected_operating_line=selected_line_by_property.get(property_info.pk),
140 )
141 )
142 return options
145def create_operating_line_from_output_property(
146 *,
147 study: EconomicsStudy,
148 property_info: PropertyInfo,
149 category: str,
150 economic_effect: str = OperatingLineEconomicEffect.COST,
151 outlet_stream_disposition: str = "",
152 rate_type: str | None = None,
153) -> OperatingCostLine:
154 """Create or update an operating line from a selected operating property."""
155 _validate_operating_property(study=study, property_info=property_info)
156 category, outlet_stream_disposition = _normalize_category(
157 category=category,
158 outlet_stream_disposition=outlet_stream_disposition,
159 )
160 economic_effect = _normalize_economic_effect(
161 category=category,
162 economic_effect=economic_effect,
163 )
164 source_value = _decimal_property_value(property_info)
165 settings_profile = get_settings_profile(study)
166 currency = settings_profile.currency if settings_profile else "NZD"
167 source_object = property_info.set.simulationObject
168 source_option = next(
169 option for option in output_stream_property_options(study) if option.property_info == property_info.pk
170 )
171 inferred_rate_type = _inferred_rate_type_for_operating_property(
172 category=category,
173 source_kind=source_option.source_kind,
174 )
175 rate_type = rate_type if rate_type is not None else inferred_rate_type
176 rate_defaults = operating_line_rate_defaults_for_category(
177 category=category,
178 study=study,
179 currency=currency,
180 property_unit=property_info.unit or "",
181 rate_type=rate_type,
182 )
183 defaults = {
184 "flowsheet": study.flowsheet,
185 "label": operating_line_result_label(
186 source_object=source_object,
187 property_info=property_info,
188 category=category,
189 economic_effect=economic_effect,
190 source_kind=source_option.source_kind,
191 ),
192 "line_type": category,
193 "category": category,
194 "economic_effect": economic_effect,
195 "currency": currency,
196 "basis_quantity": source_value,
197 "basis_unit": property_info.unit or "",
198 "basis_quantity_source": OperatingLineBasisQuantitySource.SOURCE_PROPERTY,
199 "rate_amount": rate_defaults["rate_amount"],
200 "rate_unit": rate_defaults["rate_unit"],
201 "rate_type": rate_type or "",
202 "rate_source_mode": OperatingLineRateSourceMode.PROJECT_DEFAULT if rate_type else OperatingLineRateSourceMode.CUSTOM,
203 "calculation_method": "work_to_cost" if source_option.suggested_group == "energy" else "rate_times_quantity",
204 "source_default_rate": rate_defaults["source_default_rate"],
205 "outlet_stream_disposition": outlet_stream_disposition,
206 "included": outlet_stream_disposition != OutletStreamDisposition.IGNORED,
207 "manual": False,
208 "source": STREAM_OPERATING_LINE_SOURCE,
209 "warning_payload": {
210 "source": STREAM_OPERATING_LINE_SOURCE,
211 "source_kind": source_option.source_kind,
212 "source_object_id": source_object.pk,
213 "source_object_name": source_object.componentName,
214 "source_object_type": source_object.objectType,
215 "property_info_id": property_info.pk,
216 "property_key": property_info.key,
217 "property_name": property_info.displayName,
218 },
219 }
220 line, _ = OperatingCostLine.objects.update_or_create(
221 study=study,
222 source_property_info=property_info,
223 source=STREAM_OPERATING_LINE_SOURCE,
224 defaults=defaults,
225 )
226 return line
229def operating_line_result_label(
230 *,
231 source_object: SimulationObject,
232 property_info: PropertyInfo,
233 category: str,
234 economic_effect: str = OperatingLineEconomicEffect.COST,
235 source_kind: str = "",
236) -> str:
237 object_name = source_object.componentName or "Property source"
238 if category == OperatingLineCategory.ENERGY:
239 suffix = "credit" if economic_effect == OperatingLineEconomicEffect.REVENUE else "cost"
240 if source_kind == UNIT_POWER_WORK_SOURCE_KIND:
241 return f"{object_name} annual electricity {suffix}"
242 if source_kind == UNIT_HEATING_DUTY_SOURCE_KIND:
243 return f"{object_name} annual heating {suffix}"
244 if source_kind == UNIT_COOLING_DUTY_SOURCE_KIND: 244 ↛ 246line 244 didn't jump to line 246 because the condition on line 244 was always true
245 return f"{object_name} annual cooling {suffix}"
246 property_name = property_info.displayName or "energy"
247 return f"{object_name} annual energy {suffix} from {property_name}"
248 if category == OperatingLineCategory.FEEDSTOCK:
249 return f"{object_name} annual feedstock cost"
250 if category == OperatingLineCategory.OUTPUT_REVENUE: 250 ↛ 252line 250 didn't jump to line 252 because the condition on line 250 was always true
251 return f"{object_name} annual output revenue"
252 if category == OperatingLineCategory.DISPOSAL:
253 return f"{object_name} annual disposal cost"
254 if category == OperatingLineCategory.MAINTENANCE:
255 return f"{object_name} annual maintenance cost"
256 property_name = property_info.displayName or "property"
257 return f"{object_name} annual operating cost from {property_name}"
260def display_label_for_operating_line(line: OperatingCostLine) -> str:
261 if (
262 line.source != STREAM_OPERATING_LINE_SOURCE
263 or not line.source_property_info_id
264 or line.manual
265 or line.basis_quantity_source != OperatingLineBasisQuantitySource.SOURCE_PROPERTY
266 ):
267 return line.label
268 try:
269 property_info = line.source_property_info
270 source_object = property_info.set.simulationObject
271 except (AttributeError, ObjectDoesNotExist):
272 return line.label
274 legacy_label = f"{source_object.componentName or 'Property source'} {property_info.displayName}"
275 if line.label != legacy_label: 275 ↛ 277line 275 didn't jump to line 277 because the condition on line 275 was always true
276 return line.label
277 payload = line.warning_payload if isinstance(line.warning_payload, dict) else {}
278 return operating_line_result_label(
279 source_object=source_object,
280 property_info=property_info,
281 category=line.category or line.line_type,
282 economic_effect=line.economic_effect,
283 source_kind=str(payload.get("source_kind", "")),
284 )
287def _inferred_rate_type_for_operating_property(*, category: str, source_kind: str) -> str:
288 if category == OperatingLineCategory.ENERGY and source_kind == UNIT_POWER_WORK_SOURCE_KIND:
289 return DefaultRateType.ELECTRICITY
290 if category == OperatingLineCategory.MAINTENANCE: 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true
291 return DefaultRateType.MAINTENANCE
292 return ""
295def _normalize_economic_effect(*, category: str, economic_effect: str) -> str:
296 if category == OperatingLineCategory.OUTPUT_REVENUE:
297 return OperatingLineEconomicEffect.REVENUE
298 if economic_effect == OperatingLineEconomicEffect.REVENUE:
299 return OperatingLineEconomicEffect.REVENUE
300 return OperatingLineEconomicEffect.COST
303def sync_operating_lines_for_property(property_info: PropertyInfo) -> int:
304 """Refresh property-backed operating lines after their source property changes."""
305 from Economics.results.services.lifecycle.runs import mark_result_runs_stale_for_study
307 changed_study_ids = sync_operating_line_sources_for_property(property_info)
308 stale_count = 0
309 for study in EconomicsStudy.objects.filter(pk__in=changed_study_ids).order_by("pk"):
310 stale_count += mark_result_runs_stale_for_study(
311 study=study,
312 reason="flowsheet_property_changed",
313 )
314 return stale_count
317def disconnect_operating_lines_for_deleted_property(property_info: PropertyInfo) -> int:
318 """Clear source-backed quantities that referenced a property being deleted."""
319 from Economics.results.services.lifecycle.runs import mark_result_runs_stale_for_study
321 changed_study_ids: set[int] = set()
322 lines = OperatingCostLine.objects.filter(
323 flowsheet=property_info.flowsheet,
324 source_property_info=property_info,
325 basis_quantity_source=OperatingLineBasisQuantitySource.SOURCE_PROPERTY,
326 ).select_related("study")
327 for line in lines:
328 line.source_property_info = None
329 line.basis_quantity_source = OperatingLineBasisQuantitySource.MANUAL_OVERRIDE
330 line.basis_quantity = None
331 line.basis_unit = ""
332 line.save(
333 update_fields=[
334 "source_property_info",
335 "basis_quantity_source",
336 "basis_quantity",
337 "basis_unit",
338 "updated_at",
339 ]
340 )
341 changed_study_ids.add(line.study_id)
342 stale_count = 0
343 for study in EconomicsStudy.objects.filter(pk__in=changed_study_ids).order_by("pk"):
344 stale_count += mark_result_runs_stale_for_study(
345 study=study,
346 reason="flowsheet_property_changed",
347 )
348 return stale_count
351def sync_operating_line_sources_for_property(property_info: PropertyInfo) -> set[int]:
352 """Refresh persisted operating-line quantities backed by one property."""
353 source_value = _decimal_property_value(property_info)
354 basis_unit = property_info.unit or ""
355 changed_study_ids: set[int] = set()
356 lines = OperatingCostLine.objects.filter(
357 flowsheet=property_info.flowsheet,
358 source=STREAM_OPERATING_LINE_SOURCE,
359 source_property_info=property_info,
360 basis_quantity_source=OperatingLineBasisQuantitySource.SOURCE_PROPERTY,
361 ).select_related("study")
362 for line in lines:
363 changed_fields: list[str] = []
364 if line.basis_quantity != source_value:
365 line.basis_quantity = source_value
366 changed_fields.append("basis_quantity")
367 if line.basis_unit != basis_unit:
368 line.basis_unit = basis_unit
369 changed_fields.append("basis_unit")
370 if changed_fields:
371 line.save(update_fields=[*changed_fields, "updated_at"])
372 changed_study_ids.add(line.study_id)
373 return changed_study_ids
376def sync_operating_line_sources_for_study(study: EconomicsStudy) -> set[int]:
377 """Refresh all source-property operating quantities for a study before recalculation."""
378 changed_study_ids: set[int] = set()
379 property_ids = (
380 OperatingCostLine.objects.filter(
381 flowsheet=study.flowsheet,
382 study=study,
383 source=STREAM_OPERATING_LINE_SOURCE,
384 source_property_info__isnull=False,
385 basis_quantity_source=OperatingLineBasisQuantitySource.SOURCE_PROPERTY,
386 )
387 .order_by()
388 .values_list("source_property_info_id", flat=True)
389 .distinct()
390 )
391 for property_info in PropertyInfo.objects.filter(
392 flowsheet=study.flowsheet,
393 pk__in=property_ids,
394 ):
395 changed_study_ids.update(sync_operating_line_sources_for_property(property_info))
396 return changed_study_ids
399def sync_project_default_operating_line_rates_for_study(study: EconomicsStudy, *, reason: str) -> int:
400 """Refresh operating-line rates that are linked to the project default."""
401 from Economics.results.services.lifecycle.runs import mark_result_runs_stale_for_study
403 study = EconomicsStudy.objects.get(pk=study.pk)
404 settings_profile = get_settings_profile(study)
405 currency = settings_profile.currency if settings_profile else "NZD"
406 changed = False
407 lines = OperatingCostLine.objects.filter(
408 flowsheet=study.flowsheet,
409 study=study,
410 rate_source_mode=OperatingLineRateSourceMode.PROJECT_DEFAULT,
411 rate_type__gt="",
412 ).select_related("source_default_rate")
413 for line in lines:
414 rate_defaults = operating_line_rate_defaults_for_category(
415 category=line.category,
416 study=study,
417 currency=currency or line.currency or "NZD",
418 property_unit=line.basis_unit,
419 rate_type=line.rate_type,
420 )
421 source_default_rate = rate_defaults["source_default_rate"]
423 changed_fields: list[str] = []
424 if line.source_default_rate_id != (source_default_rate.pk if source_default_rate is not None else None):
425 line.source_default_rate = source_default_rate
426 changed_fields.append("source_default_rate")
427 for field_name in ("rate_amount", "rate_unit"):
428 value = rate_defaults[field_name]
429 if getattr(line, field_name) != value:
430 setattr(line, field_name, value)
431 changed_fields.append(field_name)
432 if changed_fields: 432 ↛ 413line 432 didn't jump to line 413 because the condition on line 432 was always true
433 line.save(update_fields=[*changed_fields, "updated_at"])
434 changed = True
436 return mark_result_runs_stale_for_study(study=study, reason=reason) if changed else 0
439def _suggested_streams(study: EconomicsStudy):
440 for stream in _terminal_output_streams(study):
441 yield (
442 stream,
443 "terminal_output_stream",
444 "output",
445 OperatingLineCategory.OUTPUT_REVENUE,
446 OutletStreamDisposition.SOLD,
447 )
448 for stream in _starting_input_streams(study):
449 yield (
450 stream,
451 "starting_input_stream",
452 "feedstock",
453 OperatingLineCategory.FEEDSTOCK,
454 "",
455 )
458def _terminal_output_streams(study: EconomicsStudy):
459 return (
460 SimulationObject.objects.filter(
461 flowsheet=study.flowsheet,
462 objectType="stream",
463 connectedPorts__direction=ConType.Outlet,
464 )
465 .exclude(connectedPorts__direction=ConType.Inlet)
466 .distinct()
467 .prefetch_related("properties__ContainedProperties__values")
468 .order_by("componentName", "pk")
469 )
472def _starting_input_streams(study: EconomicsStudy):
473 return (
474 SimulationObject.objects.filter(
475 flowsheet=study.flowsheet,
476 objectType="stream",
477 connectedPorts__direction=ConType.Inlet,
478 )
479 .exclude(connectedPorts__direction=ConType.Outlet)
480 .distinct()
481 .prefetch_related("properties__ContainedProperties__values")
482 .order_by("componentName", "pk")
483 )
486def _unit_operations(study: EconomicsStudy):
487 return (
488 SimulationObject.objects.filter(flowsheet=study.flowsheet)
489 .exclude(objectType="stream")
490 .prefetch_related("properties__ContainedProperties__values")
491 .order_by("componentName", "pk")
492 )
495def _eligible_stream_properties(stream: SimulationObject):
496 property_set = getattr(stream, "properties", None)
497 if property_set is None: 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true
498 return []
499 return [
500 property_info
501 for property_info in property_set.ContainedProperties.filter(type="numeric").order_by("displayName", "key", "pk")
502 if _is_scalar_current_property(property_info)
503 ]
506def _is_scalar_current_property(property_info: PropertyInfo) -> bool:
507 values = list(property_info.values.all())
508 return len(values) == 1 and values[0].value not in (None, "")
511def _validate_operating_property(*, study: EconomicsStudy, property_info: PropertyInfo) -> None:
512 available_ids = {option.property_info for option in output_stream_property_options(study)}
513 if property_info.pk not in available_ids:
514 raise ValueError("Operating-line source must be a suggested current scalar operating property.")
517def _normalize_category(*, category: str, outlet_stream_disposition: str) -> tuple[str, str]:
518 if category == OperatingLineCategory.OUTPUT_REVENUE:
519 disposition = outlet_stream_disposition or OutletStreamDisposition.SOLD
520 if disposition != OutletStreamDisposition.SOLD: 520 ↛ 521line 520 didn't jump to line 521 because the condition on line 520 was never true
521 raise ValueError("Sold outputs must use the sold disposition.")
522 return OperatingLineCategory.OUTPUT_REVENUE, disposition
523 if category == OperatingLineCategory.DISPOSAL: 523 ↛ 524line 523 didn't jump to line 524 because the condition on line 523 was never true
524 disposition = outlet_stream_disposition or OutletStreamDisposition.DISPOSED
525 if disposition != OutletStreamDisposition.DISPOSED:
526 raise ValueError("Disposed outputs must use the disposed disposition.")
527 return OperatingLineCategory.DISPOSAL, disposition
528 if category in (OperatingLineCategory.ENERGY, OperatingLineCategory.FEEDSTOCK): 528 ↛ 530line 528 didn't jump to line 530 because the condition on line 528 was always true
529 return category, ""
530 raise ValueError("Output stream properties can be categorised as Energy, Feedstock, Sold output, or Disposed output.")
533def _property_matches_group(property_info: PropertyInfo, group: str) -> bool:
534 if group in ("output", "feedstock"): 534 ↛ 536line 534 didn't jump to line 536 because the condition on line 534 was always true
535 return _is_mass_flow_property(property_info)
536 if group == "energy":
537 return _is_power_or_work_property(property_info)
538 return False
541def unit_energy_source_kind(*, unit: SimulationObject, property_info: PropertyInfo) -> str | None:
542 object_type = _normalized_property_key(unit.objectType)
543 property_key = _normalized_property_key(property_info.key)
544 if property_key in WORK_PROPERTY_KEYS or property_key.endswith(".work"):
545 return UNIT_POWER_WORK_SOURCE_KIND
546 if object_type in EXTERNAL_DUTY_OBJECT_TYPES:
547 return None
548 if property_key == "heatduty":
549 if object_type == "cooler":
550 return UNIT_COOLING_DUTY_SOURCE_KIND
551 return UNIT_HEATING_DUTY_SOURCE_KIND
552 if property_key in HEATING_DUTY_PROPERTY_KEYS:
553 return UNIT_HEATING_DUTY_SOURCE_KIND
554 if property_key in COOLING_DUTY_PROPERTY_KEYS:
555 return UNIT_COOLING_DUTY_SOURCE_KIND
556 return None
559def _is_mass_flow_property(property_info: PropertyInfo) -> bool:
560 text = _property_search_text(property_info)
561 return "massflow" in text or ("mass" in text and "flow" in text)
564def _is_power_or_work_property(property_info: PropertyInfo) -> bool:
565 return unit_energy_source_kind(
566 unit=property_info.set.simulationObject,
567 property_info=property_info,
568 ) is not None
571def _property_search_text(property_info: PropertyInfo) -> str:
572 return "".join(
573 [
574 property_info.key or "",
575 property_info.displayName or "",
576 property_info.unitType or "",
577 ]
578 ).lower().replace("_", "").replace(" ", "").replace("-", "")
581def _normalized_property_key(key: str) -> str:
582 return (key or "").lower().replace("_", "").replace(" ", "").replace("-", "")
585def _decimal_property_value(property_info: PropertyInfo) -> Decimal | None:
586 value = property_info.get_value()
587 if value in (None, ""):
588 return None
589 try:
590 return Decimal(str(value)).quantize(OPERATING_LINE_DECIMAL_QUANTUM, rounding=ROUND_HALF_UP)
591 except (InvalidOperation, TypeError, ValueError):
592 return None