Coverage for backend/django/Economics/costing/capital/serializers.py: 78%
171 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from decimal import Decimal
3from django.core.exceptions import ObjectDoesNotExist
4from drf_spectacular.utils import extend_schema_field
5from rest_framework import serializers
7from Economics.costing.models import CapitalCostLine, CostCurve, CostableItem
8from Economics.costing.cost_curves.driver_properties import validate_cost_driver_property
9from Economics.scheduling.series import study_property_is_schedule_varying
10from Economics.shared.choices import CapitalLineBasis, CapitalLineDepreciationMode, EconomicsScheduleMode
11from Economics.shared.serializer_base import FlowsheetScopedSerializer, _current_flowsheet_id
12from Economics.costing.cost_curves.evaluation import normalize_economics_unit_notation
13from Economics.costing.cost_curves.driver_specs import (
14 CapitalCostDriverInput,
15 CapitalCostDriverInputsPayload,
16 CostCurveDriverSpec,
17 CostCurveDriverSpecPayload,
18 normalize_capital_cost_driver_inputs,
19 normalize_required_driver_specs,
20)
21from idaes_factory.unit_conversion.unit_conversion import can_convert
24@extend_schema_field(
25 CapitalCostDriverInput.model_json_schema(),
26 component_name="CapitalCostDriverInput",
27)
28class CapitalCostDriverInputField(serializers.JSONField):
29 """JSON transport field whose OpenAPI component comes from Pydantic."""
31 def to_representation(self, value):
32 if isinstance(value, CapitalCostDriverInput): 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true
33 return value.model_dump(mode="json")
34 return super().to_representation(value)
37class CapitalCostLineSerializer(FlowsheetScopedSerializer):
38 same_flowsheet_fields = ("study", "costable_item")
39 driver_inputs = serializers.DictField(
40 child=CapitalCostDriverInputField(),
41 required=False,
42 )
44 class Meta:
45 model = CapitalCostLine
46 fields = (
47 "id",
48 "flowsheet",
49 "study",
50 "costable_item",
51 "cost_curve",
52 "label",
53 "line_type",
54 "calculation_basis",
55 "amount",
56 "basis_percent",
57 "depreciation_mode",
58 "depreciation_life_years",
59 "depreciation_salvage_percent",
60 "peak_demand_kw",
61 "minimum_peak_demand_kw",
62 "currency",
63 "included",
64 "manual",
65 "source",
66 "confidence",
67 "warning_payload",
68 "driver_inputs",
69 "created_at",
70 "updated_at",
71 )
72 read_only_fields = ("id", "flowsheet", "created_at", "updated_at")
74 def validate(self, attrs):
75 attrs = super().validate(attrs)
76 calculation_basis = attrs.get(
77 "calculation_basis",
78 getattr(self.instance, "calculation_basis", CapitalLineBasis.FIXED),
79 )
80 if calculation_basis == CapitalLineBasis.FIXED and "basis_percent" not in attrs:
81 basis_percent = None
82 else:
83 basis_percent = attrs.get(
84 "basis_percent",
85 getattr(self.instance, "basis_percent", None),
86 )
87 amount = attrs.get("amount", getattr(self.instance, "amount", None))
88 depreciation_mode = attrs.get(
89 "depreciation_mode",
90 getattr(self.instance, "depreciation_mode", CapitalLineDepreciationMode.STUDY_DEFAULT),
91 )
92 if depreciation_mode == CapitalLineDepreciationMode.CUSTOM:
93 depreciation_life_years = attrs.get(
94 "depreciation_life_years",
95 getattr(self.instance, "depreciation_life_years", None),
96 )
97 depreciation_salvage_percent = attrs.get(
98 "depreciation_salvage_percent",
99 getattr(self.instance, "depreciation_salvage_percent", None),
100 )
101 else:
102 depreciation_life_years = attrs.get("depreciation_life_years")
103 depreciation_salvage_percent = attrs.get("depreciation_salvage_percent")
104 peak_demand_kw = attrs.get("peak_demand_kw", getattr(self.instance, "peak_demand_kw", None))
105 minimum_peak_demand_kw = attrs.get(
106 "minimum_peak_demand_kw",
107 getattr(self.instance, "minimum_peak_demand_kw", None),
108 )
109 manual = attrs.get("manual", getattr(self.instance, "manual", False))
110 errors = {}
111 if calculation_basis == CapitalLineBasis.BASE_CAPEX_PERCENT:
112 if basis_percent is None:
113 errors["basis_percent"] = "Percentage capital lines require a percentage."
114 elif basis_percent < 0: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 errors["basis_percent"] = "Percentage capital lines cannot be negative."
116 elif basis_percent is not None: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 errors["basis_percent"] = "Fixed capital lines do not use a percentage basis."
118 if depreciation_mode == CapitalLineDepreciationMode.CUSTOM:
119 if depreciation_life_years in (None, 0):
120 errors["depreciation_life_years"] = "Custom depreciation requires an equipment life."
121 elif depreciation_life_years is not None:
122 errors["depreciation_life_years"] = "Only custom depreciation uses a line equipment life."
123 if depreciation_mode != CapitalLineDepreciationMode.CUSTOM and depreciation_salvage_percent is not None:
124 errors["depreciation_salvage_percent"] = "Only custom depreciation uses a line residual value."
125 if (
126 depreciation_salvage_percent is not None
127 and not Decimal("0") <= depreciation_salvage_percent <= Decimal("100")
128 ):
129 errors["depreciation_salvage_percent"] = "Residual value must be between 0 and 100 percent."
130 if manual and calculation_basis == CapitalLineBasis.FIXED and amount is not None and amount < 0: 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true
131 errors["amount"] = "Fixed capital lines cannot be negative."
132 if peak_demand_kw is not None and peak_demand_kw < 0: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 errors["peak_demand_kw"] = "Peak demand cannot be negative."
134 if minimum_peak_demand_kw is not None and minimum_peak_demand_kw < 0: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 errors["minimum_peak_demand_kw"] = "Minimum peak demand cannot be negative."
136 if peak_demand_kw is not None and minimum_peak_demand_kw is not None and peak_demand_kw < minimum_peak_demand_kw:
137 errors["peak_demand_kw"] = "Peak demand cannot be below the current flowsheet work."
138 if errors:
139 raise serializers.ValidationError(errors)
140 cost_curve = attrs.get("cost_curve", getattr(self.instance, "cost_curve", None))
141 if cost_curve is not None:
142 costable_item = attrs.get("costable_item", getattr(self.instance, "costable_item", None))
143 if "driver_inputs" in attrs:
144 attrs["driver_inputs"] = _normalized_capital_cost_driver_inputs(
145 attrs["driver_inputs"]
146 )
147 elif "cost_curve" in attrs or self.instance is None:
148 attrs["driver_inputs"] = _reconciled_capital_line_driver_inputs_for_curve(
149 getattr(self.instance, "driver_inputs", {}) if self.instance else {},
150 cost_curve,
151 )
152 if "driver_inputs" in attrs or "costable_item" in attrs: 152 ↛ 164line 152 didn't jump to line 164 because the condition on line 152 was always true
153 driver_inputs = attrs.get(
154 "driver_inputs",
155 getattr(self.instance, "driver_inputs", {}) if self.instance else {},
156 )
157 study = attrs.get("study", getattr(self.instance, "study", None))
158 _validate_capital_line_driver_inputs_for_curve(
159 driver_inputs,
160 cost_curve,
161 costable_item=costable_item,
162 study=study,
163 )
164 return attrs
166 def create(self, validated_data):
167 _normalize_capital_line_values(validated_data)
168 return super().create(validated_data)
170 def update(self, instance, validated_data):
171 _normalize_capital_line_values(validated_data)
172 return super().update(instance, validated_data)
175def _normalize_capital_line_values(validated_data: dict) -> None:
176 calculation_basis = validated_data.get("calculation_basis")
177 if calculation_basis == CapitalLineBasis.BASE_CAPEX_PERCENT:
178 validated_data["amount"] = None
179 elif calculation_basis == CapitalLineBasis.FIXED:
180 validated_data["basis_percent"] = None
183def _normalized_capital_cost_driver_inputs(
184 inputs,
185) -> CapitalCostDriverInputsPayload:
186 """Route DRF writes through the Pydantic capital-driver-input contract."""
187 try:
188 return normalize_capital_cost_driver_inputs(inputs)
189 except ValueError as exc:
190 raise serializers.ValidationError({"driver_inputs": str(exc)}) from exc
193def _validate_capital_line_driver_inputs_for_curve(
194 inputs: CapitalCostDriverInputsPayload,
195 curve: CostCurve,
196 *,
197 costable_item: CostableItem | None,
198 study,
199) -> None:
200 """Reject driver-input payloads that do not match the selected curve contract."""
201 try:
202 specs = normalize_required_driver_specs(curve.required_driver_specs)
203 except ValueError as exc:
204 raise serializers.ValidationError({"cost_curve": str(exc)}) from exc
205 spec_keys = {spec["key"] for spec in specs}
206 unknown_keys = sorted(set(inputs) - spec_keys)
207 if unknown_keys: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 raise serializers.ValidationError(
209 {"driver_inputs": f"Unknown driver input keys for selected cost curve: {', '.join(unknown_keys)}"}
210 )
211 missing_keys = sorted(spec["key"] for spec in specs if spec.get("required", True) and spec["key"] not in inputs)
212 if missing_keys: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 raise serializers.ValidationError(
214 {"driver_inputs": f"Missing required driver input keys for selected cost curve: {', '.join(missing_keys)}"}
215 )
216 flowsheet_id = _current_flowsheet_id() or study.flowsheet_state.flowsheet_id
217 specs_by_key = {spec["key"]: spec for spec in specs}
218 for key, driver_input in inputs.items():
219 spec = specs_by_key[key]
220 source = driver_input.get("source") or ""
221 if source and source not in (spec.get("source_options") or []): 221 ↛ 222line 221 didn't jump to line 222 because the condition on line 221 was never true
222 raise serializers.ValidationError(
223 {"driver_inputs": f"Driver input `{key}` source `{source}` is not allowed for the selected cost curve."}
224 )
225 if not _units_are_compatible(driver_input["unit"], spec["unit"]): 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true
226 raise serializers.ValidationError(
227 {"driver_inputs": f"Driver input `{key}` unit `{driver_input['unit']}` is incompatible with `{spec['unit']}`."}
228 )
229 property_info_id = driver_input.get("property_info")
230 if driver_input.get("source") == "property" and property_info_id is not None:
231 property_info = _validate_driver_input_property(
232 key=key,
233 property_info_id=property_info_id,
234 flowsheet_id=flowsheet_id,
235 target_unit=spec["unit"],
236 costable_item=costable_item,
237 )
238 if driver_input.get("aggregate_function"):
239 _validate_schedule_aggregate_property(
240 key=key,
241 study=study,
242 property_info=property_info,
243 )
244 elif driver_input.get("aggregate_function"): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 raise serializers.ValidationError(
246 {
247 "driver_inputs": (
248 f"Driver input `{key}` can use aggregate sizing only after a sizing property has been selected."
249 )
250 }
251 )
254def _reconciled_capital_line_driver_inputs_for_curve(inputs, curve: CostCurve) -> CapitalCostDriverInputsPayload:
255 """Keep matching driver-input keys and replace stale keys for a selected curve."""
256 try:
257 existing = normalize_capital_cost_driver_inputs(inputs or {})
258 except ValueError:
259 existing = {}
260 try:
261 specs = normalize_required_driver_specs(curve.required_driver_specs)
262 except ValueError as exc:
263 raise serializers.ValidationError({"cost_curve": str(exc)}) from exc
264 reconciled: CapitalCostDriverInputsPayload = {}
265 for spec_payload in specs:
266 spec = CostCurveDriverSpec.model_validate(spec_payload)
267 input_payload = existing.get(spec.key)
268 if input_payload is None: 268 ↛ 271line 268 didn't jump to line 271 because the condition on line 268 was always true
269 input_payload = CapitalCostDriverInput.from_spec_default(spec).model_dump(mode="json")
270 else:
271 input_payload = {**input_payload, "unit": spec.unit}
272 reconciled[spec.key] = input_payload
273 return reconciled
276def _units_are_compatible(source_unit: str | None, target_unit: str | None) -> bool:
277 try:
278 return can_convert(
279 normalize_economics_unit_notation(str(source_unit or "").strip()),
280 normalize_economics_unit_notation(str(target_unit or "").strip()),
281 )
282 except Exception:
283 return False
286def _validate_driver_input_property(
287 *,
288 key: str,
289 property_info_id: int,
290 flowsheet_id: int,
291 target_unit: str,
292 costable_item: CostableItem | None,
293):
294 from core.auxiliary.models import PropertyInfo
296 property_info = PropertyInfo.objects.filter(
297 pk=property_info_id,
298 flowsheet_state__flowsheet_id=flowsheet_id,
299 ).first()
300 if property_info is None: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 raise serializers.ValidationError(
302 {"driver_inputs": f"Driver input `{key}` references a property outside this flowsheet."}
303 )
304 if not _units_are_compatible(property_info.unit, target_unit):
305 raise serializers.ValidationError(
306 {"driver_inputs": f"Driver input `{key}` property unit `{property_info.unit}` is incompatible with `{target_unit}`."}
307 )
308 if costable_item is None: 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 return property_info
310 try:
311 driver = costable_item.cost_driver
312 except (AttributeError, ObjectDoesNotExist):
313 return property_info
314 try:
315 validate_cost_driver_property(driver, property_info)
316 except ValueError as exc:
317 raise serializers.ValidationError({"driver_inputs": str(exc)}) from exc
318 return property_info
321def _validate_schedule_aggregate_property(*, key: str, study, property_info) -> None:
322 if (
323 study is None
324 or not (
325 (study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id)
326 or study.schedule_mode == EconomicsScheduleMode.COMPOSITE
327 )
328 ):
329 raise serializers.ValidationError(
330 {
331 "driver_inputs": (
332 f"Driver input `{key}` can use aggregate sizing only when the study uses a production schedule."
333 )
334 }
335 )
336 if not study_property_is_schedule_varying(study=study, property_info=property_info):
337 raise serializers.ValidationError(
338 {
339 "driver_inputs": (
340 f"Driver input `{key}` can use aggregate sizing only for a property that varies across the production schedule."
341 )
342 }
343 )