Coverage for backend/django/Economics/costing/costable_items/serializers.py: 85%
228 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1from decimal import Decimal
3from drf_spectacular.utils import extend_schema_field
4from rest_framework import serializers
6from Economics.costing.models import CostCurve, CostDriver, CostableItem, EquipmentMapping
7from Economics.shared.choices import BulkEquipmentDriverInputMode, CostDriverSource
8from Economics.shared.serializer_base import FlowsheetScopedSerializer
9from Economics.costing.cost_curves.driver_properties import (
10 apply_recommended_property_for_mapping,
11 manual_property_for_costable_item,
12 normalize_property_cost_driver,
13 validate_cost_driver_property,
14)
15from Economics.costing.capital.lang_factors import resolve_lang_factor
18class EquipmentMappingSerializer(FlowsheetScopedSerializer):
19 same_flowsheet_fields = ("costable_item", "cost_curve")
20 lang_factor_label = serializers.SerializerMethodField()
21 effective_lang_factor = serializers.SerializerMethodField()
22 lang_factor_default_value = serializers.SerializerMethodField()
23 lang_factor_source = serializers.SerializerMethodField()
24 lang_factor_is_custom = serializers.SerializerMethodField()
25 lang_factor_default_key = serializers.SerializerMethodField()
27 class Meta:
28 model = EquipmentMapping
29 fields = (
30 "id",
31 "flowsheet",
32 "costable_item",
33 "cost_curve",
34 "equipment_category",
35 "equipment_subtype",
36 "cost_basis",
37 "install_factor_profile",
38 "install_factor",
39 "use_study_lang_factor",
40 "lang_factor_label",
41 "effective_lang_factor",
42 "lang_factor_default_value",
43 "lang_factor_source",
44 "lang_factor_is_custom",
45 "lang_factor_default_key",
46 "applicability_notes",
47 "created_at",
48 "updated_at",
49 )
50 read_only_fields = (
51 "id",
52 "flowsheet",
53 "lang_factor_label",
54 "effective_lang_factor",
55 "lang_factor_default_value",
56 "lang_factor_source",
57 "lang_factor_is_custom",
58 "lang_factor_default_key",
59 "created_at",
60 "updated_at",
61 )
63 @extend_schema_field(serializers.CharField)
64 def get_lang_factor_label(self, instance) -> str:
65 return resolve_lang_factor(instance).label
67 @extend_schema_field(serializers.DecimalField(max_digits=10, decimal_places=6, allow_null=True))
68 def get_effective_lang_factor(self, instance) -> Decimal | None:
69 return resolve_lang_factor(instance).effective_value
71 @extend_schema_field(serializers.DecimalField(max_digits=10, decimal_places=6, allow_null=True))
72 def get_lang_factor_default_value(self, instance) -> Decimal | None:
73 return resolve_lang_factor(instance).active_default_value
75 @extend_schema_field(serializers.CharField)
76 def get_lang_factor_source(self, instance) -> str:
77 return resolve_lang_factor(instance).source
79 @extend_schema_field(serializers.BooleanField)
80 def get_lang_factor_is_custom(self, instance) -> bool:
81 return resolve_lang_factor(instance).is_custom
83 @extend_schema_field(serializers.CharField)
84 def get_lang_factor_default_key(self, instance) -> str:
85 return resolve_lang_factor(instance).default_key
87 def validate(self, attrs):
88 attrs = super().validate(attrs)
89 costable_item = attrs.get("costable_item") or getattr(self.instance, "costable_item", None)
90 cost_curve = attrs.get("cost_curve") if "cost_curve" in attrs else getattr(self.instance, "cost_curve", None)
91 errors = {}
92 if cost_curve is not None:
93 self._validate_cost_curve_assignment(errors=errors, costable_item=costable_item, cost_curve=cost_curve, attrs=attrs)
94 if errors:
95 raise serializers.ValidationError(errors)
96 return attrs
98 def _validate_cost_curve_assignment(self, *, errors: dict, costable_item, cost_curve: CostCurve, attrs: dict) -> None:
99 if costable_item is None: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 errors["costable_item"] = "Cost curve assignment requires a costable item."
101 return
102 driver = getattr(costable_item, "cost_driver", None)
103 if driver is None: 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true
104 errors["cost_driver"] = "Cost curve assignment requires a selected cost driver."
105 return
106 if not cost_curve.cost_basis: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 errors["cost_basis"] = "Cost curve assignment requires a purchase or installed cost basis."
108 if cost_curve.equipment_category and cost_curve.equipment_category != self._equipment_category(attrs):
109 errors["equipment_category"] = "Mapping equipment category must match the selected cost curve."
110 if cost_curve.equipment_subtype and cost_curve.equipment_subtype != self._equipment_subtype(attrs): 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true
111 errors["equipment_subtype"] = "Mapping equipment subtype must match the selected cost curve."
112 if (cost_curve.valid_min is None or cost_curve.valid_max is None) and not cost_curve.valid_range_note:
113 errors["cost_curve"] = "Cost curve assignment requires valid range metadata or a range note."
115 def _equipment_category(self, attrs: dict) -> str:
116 cost_curve = attrs.get("cost_curve")
117 if "equipment_category" in attrs and attrs["equipment_category"]:
118 return attrs["equipment_category"]
119 if cost_curve is not None and cost_curve.equipment_category:
120 return cost_curve.equipment_category
121 if "equipment_category" in attrs: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 return attrs["equipment_category"]
123 return getattr(self.instance, "equipment_category", "")
125 def _equipment_subtype(self, attrs: dict) -> str:
126 cost_curve = attrs.get("cost_curve")
127 if "equipment_subtype" in attrs and attrs["equipment_subtype"]: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 return attrs["equipment_subtype"]
129 if cost_curve is not None and cost_curve.equipment_subtype: 129 ↛ 131line 129 didn't jump to line 131 because the condition on line 129 was always true
130 return cost_curve.equipment_subtype
131 if "equipment_subtype" in attrs:
132 return attrs["equipment_subtype"]
133 return getattr(self.instance, "equipment_subtype", "")
135 def update(self, instance, validated_data):
136 cost_curve = validated_data.get("cost_curve")
137 if cost_curve is not None:
138 if not validated_data.get("equipment_category") and cost_curve.equipment_category:
139 validated_data["equipment_category"] = cost_curve.equipment_category
140 if not validated_data.get("equipment_subtype") and cost_curve.equipment_subtype:
141 validated_data["equipment_subtype"] = cost_curve.equipment_subtype
142 category_changed = (
143 "equipment_category" in validated_data
144 and validated_data["equipment_category"] != instance.equipment_category
145 )
146 instance = super().update(instance, validated_data)
147 if category_changed:
148 apply_recommended_property_for_mapping(instance)
149 return instance
152class CostDriverSerializer(FlowsheetScopedSerializer):
153 same_flowsheet_fields = ("costable_item", "property_info", "manual_property_info")
155 class Meta:
156 model = CostDriver
157 fields = (
158 "id",
159 "flowsheet",
160 "costable_item",
161 "source",
162 "property_info",
163 "manual_property_info",
164 "sizing_mode",
165 "canonical_unit",
166 "design_value",
167 "unresolved_reason_code",
168 "warning_payload",
169 "created_at",
170 "updated_at",
171 )
172 read_only_fields = ("id", "flowsheet", "created_at", "updated_at")
174 def to_representation(self, instance):
175 data = super().to_representation(instance)
176 if data.get("sizing_mode") in {"scale", "sum", "max"}: 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true
177 data["sizing_mode"] = ""
178 return data
180 def validate(self, attrs):
181 attrs = super().validate(attrs)
182 property_info = attrs.get("property_info") if "property_info" in attrs else None
183 if property_info is not None:
184 driver = self.instance
185 if driver is None:
186 costable_item = attrs.get("costable_item")
187 if costable_item is None: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true
188 raise serializers.ValidationError(
189 {"costable_item": "Cost driver property selection requires a costable item."}
190 )
191 driver = CostDriver(costable_item=costable_item)
192 driver.manual_property_info = attrs.get(
193 "manual_property_info",
194 manual_property_for_costable_item(costable_item),
195 )
196 try:
197 validate_cost_driver_property(driver, property_info)
198 except ValueError as exc:
199 raise serializers.ValidationError({"property_info": str(exc)}) from exc
200 return attrs
202 def create(self, validated_data):
203 property_info = validated_data.get("property_info") if "property_info" in validated_data else None
204 costable_item = validated_data.get("costable_item")
205 if costable_item is not None and "manual_property_info" not in validated_data: 205 ↛ 207line 205 didn't jump to line 207 because the condition on line 205 was always true
206 validated_data["manual_property_info"] = manual_property_for_costable_item(costable_item)
207 driver = CostDriver(
208 costable_item=costable_item,
209 manual_property_info=validated_data.get("manual_property_info"),
210 design_value=validated_data.get("design_value"),
211 warning_payload=validated_data.get("warning_payload") or {},
212 )
213 if property_info is not None:
214 validated_data.update(normalize_property_cost_driver(driver, property_info))
215 elif "property_info" in validated_data: 215 ↛ 217line 215 didn't jump to line 217 because the condition on line 215 was always true
216 validated_data.update(normalize_property_cost_driver(driver, None))
217 elif validated_data.get("source") == CostDriverSource.MANUAL_OVERRIDE:
218 validated_data["property_info"] = None
219 validated_data.setdefault("sizing_mode", "manual")
220 return super().create(validated_data)
222 def update(self, instance, validated_data):
223 if "property_info" in validated_data: 223 ↛ 225line 223 didn't jump to line 225 because the condition on line 223 was always true
224 validated_data.update(normalize_property_cost_driver(instance, validated_data["property_info"]))
225 elif validated_data.get("source") == CostDriverSource.MANUAL_OVERRIDE:
226 validated_data["property_info"] = None
227 validated_data.setdefault("sizing_mode", "manual")
228 return super().update(instance, validated_data)
231class CostDriverPropertyOptionSerializer(serializers.Serializer):
232 property_info = serializers.IntegerField()
233 scope = serializers.ChoiceField(choices=("unit", "input_stream", "output_stream"))
234 object_id = serializers.IntegerField()
235 object_name = serializers.CharField()
236 object_type = serializers.CharField()
237 property_key = serializers.CharField()
238 display_name = serializers.CharField()
239 unit = serializers.CharField(allow_blank=True)
240 unit_type = serializers.CharField(allow_blank=True)
241 value_preview = serializers.CharField(allow_blank=True)
242 has_value = serializers.BooleanField()
243 recommended = serializers.BooleanField()
244 recommendation_label = serializers.CharField(allow_blank=True)
247class BulkEquipmentDriverInputSetupSerializer(serializers.Serializer):
248 mode = serializers.ChoiceField(choices=BulkEquipmentDriverInputMode.choices)
249 manual_value = serializers.CharField(required=False, allow_blank=True, default="")
251 def validate_manual_value(self, value):
252 return str(value or "").strip()
255class BulkEquipmentDriverInputsField(serializers.DictField):
256 child = BulkEquipmentDriverInputSetupSerializer()
258 def to_internal_value(self, data):
259 if data in (None, ""): 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 return {}
261 if not isinstance(data, dict): 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 raise serializers.ValidationError("Driver input setup must be an object keyed by cost curve input.")
263 normalized = super().to_internal_value(data)
264 if any(not str(key or "").strip() for key in normalized): 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 raise serializers.ValidationError("Driver input keys cannot be blank.")
266 return {str(key).strip(): value for key, value in normalized.items()}
269class BulkEquipmentSetupRequestSerializer(serializers.Serializer):
270 mapping_ids = serializers.ListField(child=serializers.IntegerField(), allow_empty=False)
271 cost_curve = serializers.IntegerField(required=False, allow_null=True)
272 dry_run = serializers.BooleanField(default=True)
273 apply_cost_curve = serializers.BooleanField(default=True)
274 apply_recommended_sizing_property = serializers.BooleanField(default=True)
275 overwrite_sizing_property = serializers.BooleanField(default=True)
276 driver_inputs = BulkEquipmentDriverInputsField(required=False, default=dict)
279class BulkEquipmentDriverInputResultSerializer(serializers.Serializer):
280 key = serializers.CharField()
281 label = serializers.CharField()
282 unit = serializers.CharField(allow_blank=True)
283 mode = serializers.ChoiceField(choices=BulkEquipmentDriverInputMode.choices)
284 status = serializers.CharField()
285 property = serializers.IntegerField(allow_null=True)
286 property_label = serializers.CharField(allow_blank=True)
287 property_unit = serializers.CharField(allow_blank=True)
288 property_scope = serializers.CharField(allow_blank=True)
289 property_object_name = serializers.CharField(allow_blank=True)
290 manual_value = serializers.CharField(allow_blank=True)
291 overwrite = serializers.BooleanField()
292 message = serializers.CharField(allow_blank=True)
295class BulkEquipmentSetupResultSerializer(serializers.Serializer):
296 mapping = serializers.IntegerField()
297 costable_item = serializers.IntegerField()
298 unit_name = serializers.CharField()
299 cost_curve = serializers.IntegerField(allow_null=True)
300 cost_curve_name = serializers.CharField(allow_blank=True)
301 curve_status = serializers.CharField()
302 curve_overwrite = serializers.BooleanField()
303 cost_driver = serializers.IntegerField(allow_null=True)
304 sizing_status = serializers.CharField()
305 sizing_property = serializers.IntegerField(allow_null=True)
306 sizing_property_label = serializers.CharField(allow_blank=True)
307 sizing_property_unit = serializers.CharField(allow_blank=True)
308 sizing_property_scope = serializers.CharField(allow_blank=True)
309 sizing_property_object_name = serializers.CharField(allow_blank=True)
310 sizing_overwrite = serializers.BooleanField()
311 driver_input_results = serializers.DictField(child=BulkEquipmentDriverInputResultSerializer())
312 message = serializers.CharField(allow_blank=True)
315class CostableItemSerializer(FlowsheetScopedSerializer):
316 same_flowsheet_fields = ("study", "simulation_object")
317 cost_driver = CostDriverSerializer(read_only=True)
318 equipment_mapping = EquipmentMappingSerializer(read_only=True)
319 simulation_object_name = serializers.CharField(source="simulation_object.componentName", read_only=True)
320 simulation_object_type = serializers.CharField(source="simulation_object.objectType", read_only=True)
322 class Meta:
323 model = CostableItem
324 fields = (
325 "id",
326 "flowsheet",
327 "study",
328 "item_type",
329 "simulation_object",
330 "simulation_object_name",
331 "simulation_object_type",
332 "name",
333 "included",
334 "manual",
335 "notes",
336 "cost_driver",
337 "equipment_mapping",
338 "created_at",
339 "updated_at",
340 )
341 read_only_fields = ("id", "flowsheet", "created_at", "updated_at")
343 def validate(self, attrs):
344 attrs = super().validate(attrs)
345 simulation_object = attrs.get("simulation_object", getattr(self.instance, "simulation_object", None))
346 if simulation_object is not None and simulation_object.objectType == "group": 346 ↛ 347line 346 didn't jump to line 347 because the condition on line 346 was never true
347 raise serializers.ValidationError(
348 {"simulation_object": "Flowsheet groups cannot be created or shown as v1 costable items."}
349 )
350 return attrs