Coverage for backend/django/Economics/costing/operating/serializers.py: 80%
185 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1from drf_spectacular.utils import extend_schema_field
2from rest_framework import serializers
4from Economics.costing.models import OperatingCostLine
5from Economics.shared.choices import (
6 DefaultRateType,
7 OperatingLineBasisQuantitySource,
8 OperatingLineCategory,
9 OperatingLineEconomicEffect,
10 OperatingLineRateSourceMode,
11 OutletStreamDisposition,
12)
13from Economics.shared.serializer_base import FlowsheetScopedSerializer
14from Economics.shared.serializers import UnitOptionSerializer
15from Economics.shared.unit_options import MAINTENANCE_RATE_UNIT
16from Economics.costing.operating.stream_properties import (
17 OperatingStreamPropertyOption,
18 _decimal_property_value as property_decimal_value,
19 display_label_for_operating_line,
20)
21from Economics.costing.operating.unit_options import operating_basis_unit_options, operating_rate_unit_options
24class OperatingCostLineSerializer(FlowsheetScopedSerializer):
25 same_flowsheet_fields = ("study", "costable_item", "source_property_info")
26 basis_unit_options = serializers.SerializerMethodField()
27 rate_unit_options = serializers.SerializerMethodField()
29 class Meta:
30 model = OperatingCostLine
31 fields = (
32 "id",
33 "flowsheet",
34 "study",
35 "costable_item",
36 "label",
37 "line_type",
38 "category",
39 "economic_effect",
40 "currency",
41 "basis_quantity",
42 "basis_unit",
43 "basis_quantity_source",
44 "basis_unit_options",
45 "rate_amount",
46 "rate_unit",
47 "rate_unit_options",
48 "rate_type",
49 "rate_source_mode",
50 "calculation_method",
51 "source_property_info",
52 "source_default_rate",
53 "outlet_stream_disposition",
54 "included",
55 "manual",
56 "source",
57 "warning_payload",
58 "created_at",
59 "updated_at",
60 )
61 read_only_fields = (
62 "id",
63 "flowsheet",
64 "basis_unit_options",
65 "rate_unit_options",
66 "created_at",
67 "updated_at",
68 )
70 def to_representation(self, instance):
71 representation = super().to_representation(instance)
72 representation["label"] = display_label_for_operating_line(instance)
73 return representation
75 @extend_schema_field(UnitOptionSerializer(many=True))
76 def get_basis_unit_options(self, instance) -> list[dict[str, str]]:
77 return operating_basis_unit_options(instance)
79 @extend_schema_field(UnitOptionSerializer(many=True))
80 def get_rate_unit_options(self, instance) -> list[dict[str, str]]:
81 return operating_rate_unit_options(instance)
83 def to_internal_value(self, data):
84 if isinstance(data, dict) and "annual_amount" in data:
85 raise serializers.ValidationError(
86 {
87 "annual_amount": (
88 "Direct annual operating amounts are not supported. "
89 "Use basis_quantity, basis_unit, rate_amount, and rate_unit."
90 )
91 }
92 )
93 return super().to_internal_value(data)
95 def validate(self, attrs):
96 attrs = super().validate(attrs)
97 category = attrs.get("category") or getattr(self.instance, "category", OperatingLineCategory.CUSTOM)
98 line_type = attrs.get("line_type") or getattr(self.instance, "line_type", "")
99 if (not category or category == OperatingLineCategory.CUSTOM) and line_type in OperatingLineCategory.values:
100 attrs["category"] = line_type
101 category = line_type
102 economic_effect = attrs.get(
103 "economic_effect",
104 getattr(self.instance, "economic_effect", ""),
105 )
106 study = attrs.get("study", getattr(self.instance, "study", None))
107 costable_item = attrs.get("costable_item", getattr(self.instance, "costable_item", None))
108 if not economic_effect:
109 economic_effect = (
110 OperatingLineEconomicEffect.REVENUE
111 if category == OperatingLineCategory.OUTPUT_REVENUE
112 else OperatingLineEconomicEffect.COST
113 )
114 attrs["economic_effect"] = economic_effect
116 errors = {}
117 if costable_item is not None and study is not None and costable_item.study_id != study.pk:
118 errors["costable_item"] = "Operating line costable item must belong to the same economics study."
119 if category == OperatingLineCategory.OUTPUT_REVENUE and economic_effect != OperatingLineEconomicEffect.REVENUE: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 attrs["economic_effect"] = OperatingLineEconomicEffect.REVENUE
121 economic_effect = OperatingLineEconomicEffect.REVENUE
122 if _operating_line_type_changed(self.instance, category=category, economic_effect=economic_effect):
123 _reset_type_owned_fields(attrs, category=category)
124 disposition = attrs.get("outlet_stream_disposition", getattr(self.instance, "outlet_stream_disposition", ""))
125 if category == OperatingLineCategory.OUTPUT_REVENUE and disposition not in ("", OutletStreamDisposition.SOLD): 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 errors["outlet_stream_disposition"] = "Sold output lines must classify the outlet stream as sold."
127 if category == OperatingLineCategory.DISPOSAL and disposition not in ("", OutletStreamDisposition.DISPOSED): 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 errors["outlet_stream_disposition"] = "Disposal lines must classify the outlet stream as disposed."
130 calculation_method = attrs.get("calculation_method", getattr(self.instance, "calculation_method", ""))
131 if calculation_method == "manual_annual":
132 errors["calculation_method"] = "Manual annual operating lines are not supported. Use an annual basis and annual rate."
133 elif calculation_method not in {"rate_times_quantity", "work_to_cost"}:
134 errors["calculation_method"] = "Choose a supported operating-line calculation method."
135 rate_source_mode = attrs.get(
136 "rate_source_mode",
137 getattr(self.instance, "rate_source_mode", OperatingLineRateSourceMode.CUSTOM),
138 )
139 rate_type = attrs.get("rate_type", getattr(self.instance, "rate_type", ""))
140 source_default_rate = attrs.get(
141 "source_default_rate",
142 getattr(self.instance, "source_default_rate", None),
143 )
144 if "rate_source_mode" not in attrs and source_default_rate is not None:
145 attrs["rate_source_mode"] = OperatingLineRateSourceMode.SOURCE_DEFAULT
146 rate_source_mode = OperatingLineRateSourceMode.SOURCE_DEFAULT
147 if source_default_rate is not None and not rate_type: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 attrs["rate_type"] = source_default_rate.rate_type
149 rate_type = source_default_rate.rate_type
150 if rate_source_mode == OperatingLineRateSourceMode.PROJECT_DEFAULT and not rate_type: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true
151 errors["rate_type"] = "Project default rates require a rate type."
152 if rate_source_mode == OperatingLineRateSourceMode.SOURCE_DEFAULT and source_default_rate is None: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 errors["source_default_rate"] = "Source default rates require a selected source."
154 if source_default_rate is not None and rate_type and source_default_rate.rate_type != rate_type: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 errors["source_default_rate"] = "Selected source default does not match the rate type."
156 if rate_source_mode == OperatingLineRateSourceMode.CUSTOM:
157 attrs["source_default_rate"] = None
159 basis_quantity_source = attrs.get(
160 "basis_quantity_source",
161 getattr(
162 self.instance,
163 "basis_quantity_source",
164 OperatingLineBasisQuantitySource.MANUAL_OVERRIDE,
165 ),
166 )
167 manual = attrs.get("manual", getattr(self.instance, "manual", False))
168 source_property_info = attrs.get(
169 "source_property_info",
170 getattr(self.instance, "source_property_info", None),
171 )
172 if basis_quantity_source == OperatingLineBasisQuantitySource.SOURCE_PROPERTY:
173 if manual:
174 errors["basis_quantity_source"] = "Manually created lines cannot use flowsheet quantities."
175 elif source_property_info is None: 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true
176 errors["basis_quantity_source"] = "Flowsheet quantities require a source property."
177 else:
178 attrs["basis_quantity"] = property_decimal_value(source_property_info)
179 attrs["basis_unit"] = source_property_info.unit or ""
181 warning_payload = attrs.get("warning_payload", getattr(self.instance, "warning_payload", {}))
182 included = attrs.get("included", getattr(self.instance, "included", True))
183 if isinstance(warning_payload, dict) and warning_payload.get("source") == "outlet_stream_suggestion": 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 if not disposition:
185 errors["outlet_stream_disposition"] = (
186 "Outlet stream suggestions must be classified as sold, disposed, or ignored before affecting economics."
187 )
188 if disposition == OutletStreamDisposition.IGNORED and included:
189 errors["included"] = "Ignored outlet stream suggestions cannot be included in economics totals."
190 if errors:
191 raise serializers.ValidationError(errors)
192 return attrs
195def _operating_line_type_changed(instance: OperatingCostLine | None, *, category: str, economic_effect: str) -> bool:
196 if instance is None:
197 return False
198 return category != instance.category or economic_effect != instance.economic_effect
201def _reset_type_owned_fields(attrs: dict, *, category: str) -> None:
202 attrs.setdefault("basis_quantity", None)
203 attrs.setdefault("basis_unit", _default_basis_unit(category))
204 attrs.setdefault("basis_quantity_source", OperatingLineBasisQuantitySource.MANUAL_OVERRIDE)
205 attrs.setdefault("source_property_info", None)
206 attrs.setdefault("rate_amount", None)
207 attrs.setdefault("rate_unit", _default_rate_unit(category))
208 attrs.setdefault("rate_type", "")
209 attrs.setdefault("rate_source_mode", OperatingLineRateSourceMode.CUSTOM)
210 attrs.setdefault("source_default_rate", None)
211 attrs.setdefault("outlet_stream_disposition", _default_outlet_stream_disposition(category))
214def _default_basis_unit(category: str) -> str:
215 if category == OperatingLineCategory.ENERGY: 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true
216 return "kW"
217 if category in { 217 ↛ 223line 217 didn't jump to line 223 because the condition on line 217 was always true
218 OperatingLineCategory.FEEDSTOCK,
219 OperatingLineCategory.OUTPUT_REVENUE,
220 OperatingLineCategory.DISPOSAL,
221 }:
222 return "kg/year"
223 if category == OperatingLineCategory.MAINTENANCE:
224 return "% fixed capital investment"
225 if category == OperatingLineCategory.LABOUR:
226 return "FTE"
227 return ""
230def _default_rate_unit(category: str) -> str:
231 if category == OperatingLineCategory.MAINTENANCE: 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true
232 return MAINTENANCE_RATE_UNIT
233 if category == OperatingLineCategory.ENERGY: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 return "NZD/kWh"
235 if category in { 235 ↛ 241line 235 didn't jump to line 241 because the condition on line 235 was always true
236 OperatingLineCategory.FEEDSTOCK,
237 OperatingLineCategory.OUTPUT_REVENUE,
238 OperatingLineCategory.DISPOSAL,
239 }:
240 return "NZD/kg"
241 return "NZD/unit"
244def _default_outlet_stream_disposition(category: str) -> str:
245 if category == OperatingLineCategory.OUTPUT_REVENUE: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 return OutletStreamDisposition.SOLD
247 if category == OperatingLineCategory.DISPOSAL: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 return OutletStreamDisposition.DISPOSED
249 return ""
252class OperatingStreamPropertyOptionSerializer(serializers.Serializer):
253 property_info = serializers.IntegerField()
254 stream_id = serializers.IntegerField()
255 stream_name = serializers.CharField()
256 source_object_id = serializers.IntegerField()
257 source_object_name = serializers.CharField()
258 source_kind = serializers.CharField()
259 property_key = serializers.CharField()
260 display_name = serializers.CharField()
261 unit = serializers.CharField()
262 unit_type = serializers.CharField()
263 value_preview = serializers.CharField()
264 has_value = serializers.BooleanField()
265 suggested_group = serializers.CharField()
266 suggested_category = serializers.CharField()
267 suggested_disposition = serializers.CharField()
268 selected_operating_line = serializers.IntegerField(allow_null=True)
270 def to_representation(self, instance: OperatingStreamPropertyOption):
271 return super().to_representation(instance)
274class OperatingLineFromPropertyRequestSerializer(serializers.Serializer):
275 property_info = serializers.IntegerField()
276 category = serializers.ChoiceField(
277 choices=[
278 OperatingLineCategory.ENERGY,
279 OperatingLineCategory.FEEDSTOCK,
280 OperatingLineCategory.OUTPUT_REVENUE,
281 OperatingLineCategory.DISPOSAL,
282 ]
283 )
284 economic_effect = serializers.ChoiceField(
285 choices=OperatingLineEconomicEffect.choices,
286 required=False,
287 )
288 outlet_stream_disposition = serializers.ChoiceField(
289 choices=[
290 "",
291 OutletStreamDisposition.SOLD,
292 OutletStreamDisposition.DISPOSED,
293 ],
294 required=False,
295 allow_blank=True,
296 )
297 rate_type = serializers.ChoiceField(
298 choices=DefaultRateType.choices,
299 required=False,
300 allow_blank=True,
301 )
303 def validate(self, attrs):
304 attrs = super().validate(attrs)
305 category = attrs.get("category")
306 if not attrs.get("economic_effect"):
307 attrs["economic_effect"] = (
308 OperatingLineEconomicEffect.REVENUE
309 if category == OperatingLineCategory.OUTPUT_REVENUE
310 else OperatingLineEconomicEffect.COST
311 )
312 if category == OperatingLineCategory.OUTPUT_REVENUE:
313 attrs["economic_effect"] = OperatingLineEconomicEffect.REVENUE
314 if attrs.get("rate_type") and attrs.get("category") != OperatingLineCategory.ENERGY: 314 ↛ 315line 314 didn't jump to line 315 because the condition on line 314 was never true
315 raise serializers.ValidationError(
316 {"rate_type": "Rate type can only be selected for energy operating lines."}
317 )
318 if attrs.get("rate_type") == DefaultRateType.MAINTENANCE: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true
319 raise serializers.ValidationError(
320 {"rate_type": "Maintenance is not an energy rate type."}
321 )
322 return attrs
325class OperatingLinesFromPropertiesRequestSerializer(serializers.Serializer):
326 lines = OperatingLineFromPropertyRequestSerializer(many=True, allow_empty=False)
328 def validate_lines(self, lines):
329 seen_property_ids = set()
330 for line in lines:
331 property_id = line["property_info"]
332 if property_id in seen_property_ids:
333 raise serializers.ValidationError(
334 "Each suggested property can only be selected once."
335 )
336 seen_property_ids.add(property_id)
337 return lines