Coverage for backend/django/Economics/costing/operating/resource_basis.py: 96%
83 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1"""Shared operating-resource classification and process energy basis helpers."""
3from __future__ import annotations
5from dataclasses import dataclass
6from decimal import Decimal
8from core.auxiliary.models.PropertyInfo import PropertyInfo
9from flowsheetInternals.unitops.models.SimulationObject import SimulationObject
10from Economics.studies.models import EconomicsStudy
11from Economics.costing.models import OperatingCostLine
12from Economics.shared.choices import OperatingLineCategory, OutletStreamDisposition
13from Economics.formulas.engine.core import FormulaError
14from Economics.formulas.builders.operating import (
15 PROCESS_ENERGY_QUANTITY_UNIT,
16 build_process_energy_contribution_formula,
17)
18from Economics.costing.operating.stream_properties import (
19 UNIT_COOLING_DUTY_SOURCE_KIND,
20 UNIT_HEATING_DUTY_SOURCE_KIND,
21 UNIT_POWER_WORK_SOURCE_KIND,
22 unit_energy_source_kind,
23)
26RESOURCE_CATEGORY_ANNUAL_COOLING_DUTY = "annual_cooling_duty"
27RESOURCE_CATEGORY_ANNUAL_HEATING_DUTY = "annual_heating_duty"
28RESOURCE_CATEGORY_ANNUAL_WORK = "annual_work"
29RESOURCE_CATEGORY_PRODUCT_OUTPUT = "product_output"
30RESOURCE_CATEGORY_RAW_MATERIAL_CONSUMPTION = "raw_material_consumption"
31RESOURCE_CATEGORY_WASTE_DISPOSAL = "waste_disposal"
32TARGET_PROCESS_ENERGY_BASIS_UNIT = "MWh/year"
33TARGET_PROCESS_ENERGY_QUANTITY_UNIT = PROCESS_ENERGY_QUANTITY_UNIT
34TARGET_PROCESS_ENERGY_BASIS_QUANTUM = Decimal("0.00000001")
35_LCOH_PROCESS_ENERGY_RESOURCE_CATEGORIES = frozenset(
36 {
37 RESOURCE_CATEGORY_ANNUAL_HEATING_DUTY,
38 RESOURCE_CATEGORY_ANNUAL_WORK,
39 }
40)
43@dataclass(frozen=True)
44class ProcessEnergyBasisContribution:
45 operating_line_id: int
46 label: str
47 resource_breakdown_category: str
48 source_quantity: Decimal
49 source_unit: str
50 target_quantity: Decimal
51 formula_key: str
54@dataclass(frozen=True)
55class TargetProcessEnergyBasis:
56 quantity: Decimal | None
57 unit: str
58 contributions: tuple[ProcessEnergyBasisContribution, ...]
61def operating_resource_source_kind(
62 *,
63 line: OperatingCostLine,
64 source_object: SimulationObject | None = None,
65 property_info: PropertyInfo | None = None,
66) -> str:
67 """Classify the process source behind an operating line."""
68 if property_info is None and line.source_property_info_id:
69 property_info = line.source_property_info
70 if source_object is None and property_info is not None:
71 source_object = property_info.set.simulationObject
72 if line.category == OperatingLineCategory.FEEDSTOCK and property_info is not None:
73 return "starting_input_stream"
74 if (
75 line.category in (OperatingLineCategory.OUTPUT_REVENUE, OperatingLineCategory.DISPOSAL)
76 and property_info is not None
77 ):
78 return "terminal_output_stream"
79 if line.category != OperatingLineCategory.ENERGY or source_object is None or property_info is None:
80 return ""
81 return unit_energy_source_kind(unit=source_object, property_info=property_info) or ""
84def operating_resource_breakdown_category(
85 *,
86 line: OperatingCostLine,
87 source_kind: str | None = None,
88) -> str:
89 """Return the stable resource breakdown bucket for an operating line."""
90 if line.category == OperatingLineCategory.FEEDSTOCK:
91 return RESOURCE_CATEGORY_RAW_MATERIAL_CONSUMPTION
92 if (
93 line.category == OperatingLineCategory.OUTPUT_REVENUE
94 or line.outlet_stream_disposition == OutletStreamDisposition.SOLD
95 ):
96 return RESOURCE_CATEGORY_PRODUCT_OUTPUT
97 if (
98 line.category == OperatingLineCategory.DISPOSAL
99 or line.outlet_stream_disposition == OutletStreamDisposition.DISPOSED
100 ):
101 return RESOURCE_CATEGORY_WASTE_DISPOSAL
102 if line.category != OperatingLineCategory.ENERGY:
103 return ""
104 source_kind = source_kind if source_kind is not None else operating_resource_source_kind(line=line)
105 if source_kind == UNIT_COOLING_DUTY_SOURCE_KIND:
106 return RESOURCE_CATEGORY_ANNUAL_COOLING_DUTY
107 if source_kind == UNIT_HEATING_DUTY_SOURCE_KIND:
108 return RESOURCE_CATEGORY_ANNUAL_HEATING_DUTY
109 if source_kind == UNIT_POWER_WORK_SOURCE_KIND:
110 return RESOURCE_CATEGORY_ANNUAL_WORK
111 return ""
114def derive_target_process_energy_basis(study: EconomicsStudy) -> TargetProcessEnergyBasis:
115 """Sum included target heating-duty and work energy as the LCOH denominator."""
116 contributions: list[ProcessEnergyBasisContribution] = []
117 lines = study.operating_lines.filter(included=True).select_related(
118 "source_property_info__set__simulationObject",
119 )
120 for line in lines:
121 source_kind = operating_resource_source_kind(line=line)
122 category = operating_resource_breakdown_category(line=line, source_kind=source_kind)
123 if category not in _LCOH_PROCESS_ENERGY_RESOURCE_CATEGORIES:
124 continue
125 try:
126 contribution_formula = build_process_energy_contribution_formula(line, study=study)
127 target_quantity = contribution_formula.evaluate()
128 except FormulaError:
129 target_quantity = None
130 if target_quantity is None: 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true
131 continue
132 target_quantity = target_quantity.quantize(TARGET_PROCESS_ENERGY_BASIS_QUANTUM)
133 contributions.append(
134 ProcessEnergyBasisContribution(
135 operating_line_id=line.pk,
136 label=line.label,
137 resource_breakdown_category=category,
138 source_quantity=contribution_formula.source_quantity,
139 source_unit=contribution_formula.source_unit,
140 target_quantity=target_quantity,
141 formula_key=contribution_formula.formula.key,
142 )
143 )
144 total = sum((contribution.target_quantity for contribution in contributions), Decimal("0")).quantize(
145 TARGET_PROCESS_ENERGY_BASIS_QUANTUM
146 )
147 return TargetProcessEnergyBasis(
148 quantity=total if total > 0 else None,
149 unit=TARGET_PROCESS_ENERGY_BASIS_UNIT,
150 contributions=tuple(contributions),
151 )