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