Coverage for backend/django/Economics/costing/operating/serializers.py: 82%
215 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +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 EconomicsScheduleMode,
8 OperatingLineBasisQuantitySource,
9 OperatingLineCategory,
10 OperatingLineEconomicEffect,
11 OperatingLineRateSourceMode,
12 OutletStreamDisposition,
13)
14from Economics.shared.serializer_base import FlowsheetScopedSerializer
15from Economics.shared.serializers import UnitOptionSerializer
16from Economics.shared.unit_options import MAINTENANCE_RATE_UNIT
17from Economics.scheduling.series import study_schedule_varying_property_ids
18from Economics.costing.operating.stream_properties import (
19 OperatingStreamPropertyOption,
20 _decimal_property_value as property_decimal_value,
21 display_label_for_operating_line,
22)
23from Economics.costing.operating.unit_options import operating_basis_unit_options, operating_rate_unit_options
26class OperatingCostLineListSerializer(serializers.ListSerializer):
27 """Seed schedule classifications for every line before rendering the list."""
29 def to_representation(self, data):
30 instances = list(data.all() if hasattr(data, "all") else data)
31 properties_by_study: dict[tuple[int, str, int | None], set[int]] = {}
32 studies = {}
33 for instance in instances:
34 study = instance.study
35 study_key = (study.pk, study.schedule_mode, study.schedule_scenario_id)
36 studies[study_key] = study
37 if instance.source_property_info_id:
38 properties_by_study.setdefault(study_key, set()).add(instance.source_property_info_id)
39 cache = self.context.setdefault("_economics_schedule_varying_property_ids", {})
40 for study_key, property_ids in properties_by_study.items():
41 if study_key not in cache: 41 ↛ 40line 41 didn't jump to line 40 because the condition on line 41 was always true
42 cache[study_key] = study_schedule_varying_property_ids(
43 study=studies[study_key],
44 property_ids=property_ids,
45 )
46 return super().to_representation(instances)
49class OperatingCostLineSerializer(FlowsheetScopedSerializer):
50 same_flowsheet_fields = ("study", "costable_item", "source_property_info")
51 basis_unit_options = serializers.SerializerMethodField()
52 rate_unit_options = serializers.SerializerMethodField()
53 schedule_basis_locked = serializers.SerializerMethodField()
55 class Meta:
56 model = OperatingCostLine
57 list_serializer_class = OperatingCostLineListSerializer
58 fields = (
59 "id",
60 "flowsheet",
61 "study",
62 "costable_item",
63 "label",
64 "line_type",
65 "category",
66 "economic_effect",
67 "currency",
68 "basis_quantity",
69 "basis_unit",
70 "basis_quantity_source",
71 "schedule_basis_locked",
72 "basis_unit_options",
73 "rate_amount",
74 "rate_unit",
75 "rate_unit_options",
76 "rate_type",
77 "rate_source_mode",
78 "calculation_method",
79 "source_property_info",
80 "source_default_rate",
81 "outlet_stream_disposition",
82 "included",
83 "manual",
84 "source",
85 "warning_payload",
86 "created_at",
87 "updated_at",
88 )
89 read_only_fields = (
90 "id",
91 "flowsheet",
92 "basis_unit_options",
93 "rate_unit_options",
94 "schedule_basis_locked",
95 "created_at",
96 "updated_at",
97 )
99 def to_representation(self, instance):
100 representation = super().to_representation(instance)
101 representation["label"] = display_label_for_operating_line(instance)
102 return representation
104 @extend_schema_field(UnitOptionSerializer(many=True))
105 def get_basis_unit_options(self, instance) -> list[dict[str, str]]:
106 return operating_basis_unit_options(instance)
108 @extend_schema_field(UnitOptionSerializer(many=True))
109 def get_rate_unit_options(self, instance) -> list[dict[str, str]]:
110 return operating_rate_unit_options(instance)
112 @extend_schema_field(serializers.BooleanField)
113 def get_schedule_basis_locked(self, instance) -> bool:
114 study = instance.study
115 uses_schedule = (
116 (study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id)
117 or study.schedule_mode == EconomicsScheduleMode.COMPOSITE
118 )
119 if (
120 not uses_schedule
121 or not instance.source_property_info_id
122 or instance.basis_quantity_source != OperatingLineBasisQuantitySource.SOURCE_PROPERTY
123 ):
124 return False
125 cache = self.context.setdefault("_economics_schedule_varying_property_ids", {})
126 study_key = (study.pk, study.schedule_mode, study.schedule_scenario_id)
127 if study_key not in cache: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 cache[study_key] = study_schedule_varying_property_ids(
129 study=study,
130 property_ids={instance.source_property_info_id},
131 )
132 return instance.source_property_info_id in cache[study_key]
134 def to_internal_value(self, data):
135 if isinstance(data, dict) and "annual_amount" in data:
136 raise serializers.ValidationError(
137 {
138 "annual_amount": (
139 "Direct annual operating amounts are not supported. "
140 "Use basis_quantity, basis_unit, rate_amount, and rate_unit."
141 )
142 }
143 )
144 return super().to_internal_value(data)
146 def validate(self, attrs):
147 attrs = super().validate(attrs)
148 category = attrs.get("category") or getattr(self.instance, "category", OperatingLineCategory.CUSTOM)
149 line_type = attrs.get("line_type") or getattr(self.instance, "line_type", "")
150 if (not category or category == OperatingLineCategory.CUSTOM) and line_type in OperatingLineCategory.values:
151 attrs["category"] = line_type
152 category = line_type
153 economic_effect = attrs.get(
154 "economic_effect",
155 getattr(self.instance, "economic_effect", ""),
156 )
157 study = attrs.get("study", getattr(self.instance, "study", None))
158 costable_item = attrs.get("costable_item", getattr(self.instance, "costable_item", None))
159 if not economic_effect:
160 economic_effect = (
161 OperatingLineEconomicEffect.REVENUE
162 if category == OperatingLineCategory.OUTPUT_REVENUE
163 else OperatingLineEconomicEffect.COST
164 )
165 attrs["economic_effect"] = economic_effect
167 errors = {}
168 if costable_item is not None and study is not None and costable_item.study_id != study.pk:
169 errors["costable_item"] = "Operating line costable item must belong to the same economics study."
170 if category == OperatingLineCategory.OUTPUT_REVENUE and economic_effect != OperatingLineEconomicEffect.REVENUE: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 attrs["economic_effect"] = OperatingLineEconomicEffect.REVENUE
172 economic_effect = OperatingLineEconomicEffect.REVENUE
173 if _operating_line_type_changed(self.instance, category=category, economic_effect=economic_effect):
174 _reset_type_owned_fields(attrs, category=category)
175 disposition = attrs.get("outlet_stream_disposition", getattr(self.instance, "outlet_stream_disposition", ""))
176 if category == OperatingLineCategory.OUTPUT_REVENUE and disposition not in ("", OutletStreamDisposition.SOLD): 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true
177 errors["outlet_stream_disposition"] = "Sold output lines must classify the outlet stream as sold."
178 if category == OperatingLineCategory.DISPOSAL and disposition not in ("", OutletStreamDisposition.DISPOSED): 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true
179 errors["outlet_stream_disposition"] = "Disposal lines must classify the outlet stream as disposed."
181 calculation_method = attrs.get("calculation_method", getattr(self.instance, "calculation_method", ""))
182 if calculation_method == "manual_annual":
183 errors["calculation_method"] = "Manual annual operating lines are not supported. Use an annual basis and annual rate."
184 elif calculation_method not in {"rate_times_quantity", "work_to_cost"}:
185 errors["calculation_method"] = "Choose a supported operating-line calculation method."
186 rate_source_mode = attrs.get(
187 "rate_source_mode",
188 getattr(self.instance, "rate_source_mode", OperatingLineRateSourceMode.CUSTOM),
189 )
190 rate_type = attrs.get("rate_type", getattr(self.instance, "rate_type", ""))
191 source_default_rate = attrs.get(
192 "source_default_rate",
193 getattr(self.instance, "source_default_rate", None),
194 )
195 if "rate_source_mode" not in attrs and source_default_rate is not None:
196 attrs["rate_source_mode"] = OperatingLineRateSourceMode.SOURCE_DEFAULT
197 rate_source_mode = OperatingLineRateSourceMode.SOURCE_DEFAULT
198 if source_default_rate is not None and not rate_type: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 attrs["rate_type"] = source_default_rate.rate_type
200 rate_type = source_default_rate.rate_type
201 if rate_source_mode == OperatingLineRateSourceMode.PROJECT_DEFAULT and not rate_type: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 errors["rate_type"] = "Project default rates require a rate type."
203 if rate_source_mode == OperatingLineRateSourceMode.SOURCE_DEFAULT and source_default_rate is None: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true
204 errors["source_default_rate"] = "Source default rates require a selected source."
205 if source_default_rate is not None and rate_type and source_default_rate.rate_type != rate_type: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 errors["source_default_rate"] = "Selected source default does not match the rate type."
207 if rate_source_mode == OperatingLineRateSourceMode.CUSTOM:
208 attrs["source_default_rate"] = None
210 basis_quantity_source = attrs.get(
211 "basis_quantity_source",
212 getattr(
213 self.instance,
214 "basis_quantity_source",
215 OperatingLineBasisQuantitySource.MANUAL_OVERRIDE,
216 ),
217 )
218 manual = attrs.get("manual", getattr(self.instance, "manual", False))
219 source_property_info = attrs.get(
220 "source_property_info",
221 getattr(self.instance, "source_property_info", None),
222 )
223 if basis_quantity_source == OperatingLineBasisQuantitySource.SOURCE_PROPERTY:
224 if manual:
225 errors["basis_quantity_source"] = "Manually created lines cannot use flowsheet quantities."
226 elif source_property_info is None: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 errors["basis_quantity_source"] = "Flowsheet quantities require a source property."
228 else:
229 attrs["basis_quantity"] = property_decimal_value(source_property_info)
230 attrs["basis_unit"] = source_property_info.unit or ""
232 warning_payload = attrs.get("warning_payload", getattr(self.instance, "warning_payload", {}))
233 included = attrs.get("included", getattr(self.instance, "included", True))
234 if isinstance(warning_payload, dict) and warning_payload.get("source") == "outlet_stream_suggestion": 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 if not disposition:
236 errors["outlet_stream_disposition"] = (
237 "Outlet stream suggestions must be classified as sold, disposed, or ignored before affecting economics."
238 )
239 if disposition == OutletStreamDisposition.IGNORED and included:
240 errors["included"] = "Ignored outlet stream suggestions cannot be included in economics totals."
241 if errors:
242 raise serializers.ValidationError(errors)
243 return attrs
246def _operating_line_type_changed(instance: OperatingCostLine | None, *, category: str, economic_effect: str) -> bool:
247 if instance is None:
248 return False
249 return category != instance.category or economic_effect != instance.economic_effect
252def _reset_type_owned_fields(attrs: dict, *, category: str) -> None:
253 attrs.setdefault("basis_quantity", None)
254 attrs.setdefault("basis_unit", _default_basis_unit(category))
255 attrs.setdefault("basis_quantity_source", OperatingLineBasisQuantitySource.MANUAL_OVERRIDE)
256 attrs.setdefault("source_property_info", None)
257 attrs.setdefault("rate_amount", None)
258 attrs.setdefault("rate_unit", _default_rate_unit(category))
259 attrs.setdefault("rate_type", "")
260 attrs.setdefault("rate_source_mode", OperatingLineRateSourceMode.CUSTOM)
261 attrs.setdefault("source_default_rate", None)
262 attrs.setdefault("outlet_stream_disposition", _default_outlet_stream_disposition(category))
265def _default_basis_unit(category: str) -> str:
266 if category == OperatingLineCategory.ENERGY: 266 ↛ 267line 266 didn't jump to line 267 because the condition on line 266 was never true
267 return "kW"
268 if category in { 268 ↛ 274line 268 didn't jump to line 274 because the condition on line 268 was always true
269 OperatingLineCategory.FEEDSTOCK,
270 OperatingLineCategory.OUTPUT_REVENUE,
271 OperatingLineCategory.DISPOSAL,
272 }:
273 return "kg/year"
274 if category == OperatingLineCategory.MAINTENANCE:
275 return "% fixed capital investment"
276 if category == OperatingLineCategory.LABOUR:
277 return "FTE"
278 return ""
281def _default_rate_unit(category: str) -> str:
282 if category == OperatingLineCategory.MAINTENANCE: 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true
283 return MAINTENANCE_RATE_UNIT
284 if category == OperatingLineCategory.ENERGY: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true
285 return "NZD/kWh"
286 if category in { 286 ↛ 292line 286 didn't jump to line 292 because the condition on line 286 was always true
287 OperatingLineCategory.FEEDSTOCK,
288 OperatingLineCategory.OUTPUT_REVENUE,
289 OperatingLineCategory.DISPOSAL,
290 }:
291 return "NZD/kg"
292 return "NZD/unit"
295def _default_outlet_stream_disposition(category: str) -> str:
296 if category == OperatingLineCategory.OUTPUT_REVENUE: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true
297 return OutletStreamDisposition.SOLD
298 if category == OperatingLineCategory.DISPOSAL: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 return OutletStreamDisposition.DISPOSED
300 return ""
303class OperatingStreamPropertyOptionSerializer(serializers.Serializer):
304 property_info = serializers.IntegerField()
305 stream_id = serializers.IntegerField()
306 stream_name = serializers.CharField()
307 source_object_id = serializers.IntegerField()
308 source_object_name = serializers.CharField()
309 source_kind = serializers.CharField()
310 property_key = serializers.CharField()
311 display_name = serializers.CharField()
312 unit = serializers.CharField()
313 unit_type = serializers.CharField()
314 value_preview = serializers.CharField()
315 has_value = serializers.BooleanField()
316 suggested_group = serializers.CharField()
317 suggested_category = serializers.CharField()
318 suggested_disposition = serializers.CharField()
319 selected_operating_line = serializers.IntegerField(allow_null=True)
321 def to_representation(self, instance: OperatingStreamPropertyOption):
322 return super().to_representation(instance)
325class OperatingLineFromPropertyRequestSerializer(serializers.Serializer):
326 property_info = serializers.IntegerField()
327 category = serializers.ChoiceField(
328 choices=[
329 OperatingLineCategory.ENERGY,
330 OperatingLineCategory.FEEDSTOCK,
331 OperatingLineCategory.OUTPUT_REVENUE,
332 OperatingLineCategory.DISPOSAL,
333 ]
334 )
335 economic_effect = serializers.ChoiceField(
336 choices=OperatingLineEconomicEffect.choices,
337 required=False,
338 )
339 outlet_stream_disposition = serializers.ChoiceField(
340 choices=[
341 "",
342 OutletStreamDisposition.SOLD,
343 OutletStreamDisposition.DISPOSED,
344 ],
345 required=False,
346 allow_blank=True,
347 )
348 rate_type = serializers.ChoiceField(
349 choices=DefaultRateType.choices,
350 required=False,
351 allow_blank=True,
352 )
354 def validate(self, attrs):
355 attrs = super().validate(attrs)
356 category = attrs.get("category")
357 if not attrs.get("economic_effect"):
358 attrs["economic_effect"] = (
359 OperatingLineEconomicEffect.REVENUE
360 if category == OperatingLineCategory.OUTPUT_REVENUE
361 else OperatingLineEconomicEffect.COST
362 )
363 if category == OperatingLineCategory.OUTPUT_REVENUE:
364 attrs["economic_effect"] = OperatingLineEconomicEffect.REVENUE
365 if attrs.get("rate_type") and attrs.get("category") != OperatingLineCategory.ENERGY: 365 ↛ 366line 365 didn't jump to line 366 because the condition on line 365 was never true
366 raise serializers.ValidationError(
367 {"rate_type": "Rate type can only be selected for energy operating lines."}
368 )
369 if attrs.get("rate_type") == DefaultRateType.MAINTENANCE: 369 ↛ 370line 369 didn't jump to line 370 because the condition on line 369 was never true
370 raise serializers.ValidationError(
371 {"rate_type": "Maintenance is not an energy rate type."}
372 )
373 return attrs
376class OperatingLinesFromPropertiesRequestSerializer(serializers.Serializer):
377 lines = OperatingLineFromPropertyRequestSerializer(many=True, allow_empty=False)
379 def validate_lines(self, lines):
380 seen_property_ids = set()
381 for line in lines:
382 property_id = line["property_info"]
383 if property_id in seen_property_ids:
384 raise serializers.ValidationError(
385 "Each suggested property can only be selected once."
386 )
387 seen_property_ids.add(property_id)
388 return lines