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