Coverage for backend/django/Economics/settings_profiles/serializers.py: 82%
220 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.reference_data.models import CostIndexSeries, EconomicsDefaultRate
7from Economics.reference_data.unit_options import default_rate_unit_options_by_type
8from Economics.settings_profiles.models import EconomicsAssumptions, EconomicsBaseline, EconomicsSettingsProfile
9from Economics.settings_profiles.unit_options import (
10 ANNUAL_HEAT_BASIS_UNIT_OPTIONS,
11 AVERAGE_POWER_UNIT_OPTIONS,
12 ELECTRICAL_UPGRADE_RATE_UNIT_OPTIONS,
13)
14from Economics.shared.choices import DefaultRateType
15from Economics.shared.serializer_base import FlowsheetScopedSerializer
16from Economics.shared.serializers import UnitOptionSerializer
17from Economics.shared.unit_options import with_current_unit
20class DefaultRateUnitOptionsSerializer(serializers.Serializer):
21 electricity = UnitOptionSerializer(many=True)
22 natural_gas = UnitOptionSerializer(many=True)
23 diesel = UnitOptionSerializer(many=True)
24 fuel_oil = UnitOptionSerializer(many=True)
25 steam = UnitOptionSerializer(many=True)
26 maintenance = UnitOptionSerializer(many=True)
29class EconomicsAssumptionsSerializer(FlowsheetScopedSerializer):
30 same_flowsheet_fields = ("study",)
31 electrical_upgrade_rate_unit_options = serializers.SerializerMethodField()
32 default_rate_unit_options = serializers.SerializerMethodField()
34 class Meta:
35 model = EconomicsAssumptions
36 fields = (
37 "id",
38 "flowsheet",
39 "study",
40 "currency",
41 "location",
42 "basis_date",
43 "discount_rate_percent",
44 "project_lifetime_years",
45 "inflation_method",
46 "annual_operating_hours",
47 "tax_rate_percent",
48 "depreciation_enabled",
49 "default_depreciation_life_years",
50 "default_depreciation_salvage_percent",
51 "contingency_percent",
52 "electrical_upgrade_rate_amount",
53 "electrical_upgrade_rate_unit",
54 "electrical_upgrade_rate_unit_options",
55 "default_lang_factor",
56 "capital_index_series",
57 "operating_index_series",
58 "default_rate_overrides",
59 "default_rate_unit_options",
60 "notes",
61 "created_at",
62 "updated_at",
63 )
64 read_only_fields = (
65 "id",
66 "flowsheet",
67 "electrical_upgrade_rate_unit",
68 "electrical_upgrade_rate_unit_options",
69 "default_rate_unit_options",
70 "created_at",
71 "updated_at",
72 )
74 @extend_schema_field(UnitOptionSerializer(many=True))
75 def get_electrical_upgrade_rate_unit_options(self, instance) -> list[dict[str, str]]:
76 return with_current_unit(ELECTRICAL_UPGRADE_RATE_UNIT_OPTIONS, instance.electrical_upgrade_rate_unit)
78 @extend_schema_field(DefaultRateUnitOptionsSerializer)
79 def get_default_rate_unit_options(self, instance) -> dict[str, list[dict[str, str]]]:
80 return default_rate_unit_options_by_type(currency=instance.currency or "NZD")
82 def validate(self, attrs):
83 attrs = super().validate(attrs)
84 tax_rate_percent = attrs.get("tax_rate_percent", getattr(self.instance, "tax_rate_percent", None))
85 annual_operating_hours = attrs.get(
86 "annual_operating_hours",
87 getattr(self.instance, "annual_operating_hours", None),
88 )
89 depreciation_enabled = attrs.get(
90 "depreciation_enabled",
91 getattr(self.instance, "depreciation_enabled", False),
92 )
93 default_depreciation_life_years = attrs.get(
94 "default_depreciation_life_years",
95 getattr(self.instance, "default_depreciation_life_years", None),
96 )
97 default_depreciation_salvage_percent = attrs.get(
98 "default_depreciation_salvage_percent",
99 getattr(self.instance, "default_depreciation_salvage_percent", None),
100 )
101 errors = {}
102 if tax_rate_percent is not None and not Decimal("0") <= tax_rate_percent <= Decimal("100"):
103 errors["tax_rate_percent"] = "Tax rate must be between 0 and 100 percent."
104 if annual_operating_hours is not None and annual_operating_hours <= 0:
105 errors["annual_operating_hours"] = "Annual operating hours must be positive."
106 if depreciation_enabled and default_depreciation_life_years in (None, 0):
107 errors["default_depreciation_life_years"] = (
108 "Default equipment life is required when depreciation is enabled."
109 )
110 if (
111 default_depreciation_salvage_percent is not None
112 and not Decimal("0") <= default_depreciation_salvage_percent <= Decimal("100")
113 ):
114 errors["default_depreciation_salvage_percent"] = (
115 "Default residual value must be between 0 and 100 percent."
116 )
117 if errors:
118 raise serializers.ValidationError(errors)
119 return attrs
121 def create(self, validated_data):
122 _prefill_default_cpi_index_series(validated_data)
123 return super().create(validated_data)
125 def validate_default_rate_overrides(self, value):
126 if value in (None, ""): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 return {}
128 if not isinstance(value, dict): 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 raise serializers.ValidationError("Default-rate overrides must be an object keyed by rate type.")
131 allowed_rate_types = set(DefaultRateType.values)
132 cleaned = {}
133 for rate_type, override in value.items():
134 if rate_type not in allowed_rate_types: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 raise serializers.ValidationError({rate_type: "Unsupported default-rate type."})
136 if not isinstance(override, dict): 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 raise serializers.ValidationError({rate_type: "Override must be an object."})
139 mode = override.get("mode")
140 if mode == "source":
141 source_default_rate = override.get("source_default_rate")
142 if source_default_rate in (None, ""): 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true
143 continue
144 try:
145 default_rate = EconomicsDefaultRate.objects.get(pk=source_default_rate)
146 except (EconomicsDefaultRate.DoesNotExist, TypeError, ValueError):
147 raise serializers.ValidationError({rate_type: "Selected source default does not exist."}) from None
148 if default_rate.rate_type != rate_type:
149 raise serializers.ValidationError({rate_type: "Selected source default does not match the rate type."})
150 cleaned_override = {
151 "mode": "source",
152 "source_default_rate": default_rate.pk,
153 }
154 if rate_type == DefaultRateType.STEAM:
155 efficiency = self._clean_positive_decimal_override(
156 override.get("boiler_efficiency_percent"),
157 fallback=(default_rate.metadata or {}).get("default_boiler_efficiency_percent")
158 if isinstance(default_rate.metadata, dict)
159 else None,
160 label="Boiler efficiency",
161 )
162 steam_energy = self._clean_positive_decimal_override(
163 override.get("steam_energy_gj_per_t"),
164 fallback=(default_rate.metadata or {}).get("steam_energy_gj_per_t")
165 if isinstance(default_rate.metadata, dict)
166 else None,
167 label="Steam energy basis",
168 )
169 if efficiency: 169 ↛ 171line 169 didn't jump to line 171 because the condition on line 169 was always true
170 cleaned_override["boiler_efficiency_percent"] = efficiency
171 if steam_energy: 171 ↛ 173line 171 didn't jump to line 173 because the condition on line 171 was always true
172 cleaned_override["steam_energy_gj_per_t"] = steam_energy
173 cleaned[rate_type] = cleaned_override
174 continue
176 if mode == "custom": 176 ↛ 195line 176 didn't jump to line 195 because the condition on line 176 was always true
177 raw_value = override.get("value")
178 unit = override.get("unit")
179 if raw_value in (None, ""): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 raise serializers.ValidationError({rate_type: "Custom default value is required."})
181 try:
182 decimal_value = Decimal(str(raw_value))
183 except (ArithmeticError, TypeError, ValueError):
184 raise serializers.ValidationError({rate_type: "Custom default value must be numeric."}) from None
185 if not decimal_value.is_finite(): 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true
186 raise serializers.ValidationError({rate_type: "Custom default value must be finite."})
187 cleaned[rate_type] = {
188 "mode": "custom",
189 "source_default_rate": None,
190 "value": str(raw_value).strip(),
191 "unit": unit.strip() if isinstance(unit, str) else "",
192 }
193 continue
195 raise serializers.ValidationError({rate_type: "Override mode must be source or custom."})
197 return cleaned
199 def _clean_positive_decimal_override(self, raw_value, *, fallback=None, label: str) -> str:
200 value = raw_value if raw_value not in (None, "") else fallback
201 if value in (None, ""): 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 return ""
203 try:
204 decimal_value = Decimal(str(value))
205 except (ArithmeticError, TypeError, ValueError):
206 raise serializers.ValidationError(f"{label} must be numeric.") from None
207 if not decimal_value.is_finite() or decimal_value <= 0: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 raise serializers.ValidationError(f"{label} must be a positive finite number.")
209 return str(value).strip()
212class EconomicsBaselineSerializer(FlowsheetScopedSerializer):
213 same_flowsheet_fields = ("study",)
214 manual_annual_heat_basis_unit_options = serializers.SerializerMethodField()
215 average_power_unit_options = serializers.SerializerMethodField()
216 unsupported_v1_fields = {
217 "manual_currency": "Manual baseline currency is inherited from study assumptions in v1.",
218 "manual_basis_date": "Manual baseline basis date is inherited from study assumptions in v1.",
219 "inherit_project_lifetime": "Manual baseline project lifetime is inherited from study assumptions in v1.",
220 "project_lifetime_years": "Manual baseline project lifetime is inherited from study assumptions in v1.",
221 "inherit_discount_rate": "Manual baseline discount rate is inherited from study assumptions in v1.",
222 "discount_rate_percent": "Manual baseline discount rate is inherited from study assumptions in v1.",
223 }
225 class Meta:
226 model = EconomicsBaseline
227 fields = (
228 "id",
229 "flowsheet",
230 "study",
231 "manual_capex",
232 "manual_annual_opex",
233 "annual_heat_basis_mode",
234 "manual_annual_heat_basis",
235 "manual_annual_heat_basis_unit",
236 "manual_annual_heat_basis_unit_options",
237 "average_power_input",
238 "average_power_unit",
239 "average_power_unit_options",
240 "residual_value",
241 "notes",
242 "created_at",
243 "updated_at",
244 )
245 read_only_fields = (
246 "id",
247 "flowsheet",
248 "manual_annual_heat_basis_unit_options",
249 "average_power_unit_options",
250 "created_at",
251 "updated_at",
252 )
254 @extend_schema_field(UnitOptionSerializer(many=True))
255 def get_manual_annual_heat_basis_unit_options(self, instance) -> list[dict[str, str]]:
256 return list(ANNUAL_HEAT_BASIS_UNIT_OPTIONS)
258 @extend_schema_field(UnitOptionSerializer(many=True))
259 def get_average_power_unit_options(self, instance) -> list[dict[str, str]]:
260 return list(AVERAGE_POWER_UNIT_OPTIONS)
262 def to_internal_value(self, data):
263 if isinstance(data, dict): 263 ↛ 271line 263 didn't jump to line 271 because the condition on line 263 was always true
264 errors = {
265 field: message
266 for field, message in self.unsupported_v1_fields.items()
267 if field in data
268 }
269 if errors: 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 raise serializers.ValidationError(errors)
271 return super().to_internal_value(data)
273 def create(self, validated_data):
274 _normalize_manual_baseline_defaults(validated_data)
275 return super().create(validated_data)
277 def update(self, instance, validated_data):
278 _normalize_manual_baseline_defaults(validated_data)
279 return super().update(instance, validated_data)
282def _prefill_default_cpi_index_series(validated_data: dict) -> None:
283 default_cpi = CostIndexSeries.objects.filter(key="stats_nz_cpi_all_groups").first()
284 if default_cpi is None: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true
285 return
286 validated_data.setdefault("capital_index_series", default_cpi)
287 validated_data.setdefault("operating_index_series", default_cpi)
290def _normalize_manual_baseline_defaults(validated_data: dict) -> None:
291 validated_data.update(
292 {
293 "manual_currency": "",
294 "manual_basis_date": None,
295 "inherit_project_lifetime": True,
296 "project_lifetime_years": None,
297 "inherit_discount_rate": True,
298 "discount_rate_percent": None,
299 }
300 )
303class EconomicsSettingsProfileSerializer(FlowsheetScopedSerializer):
304 electrical_upgrade_rate_unit_options = serializers.SerializerMethodField()
305 default_rate_unit_options = serializers.SerializerMethodField()
306 manual_annual_heat_basis_unit_options = serializers.SerializerMethodField()
307 average_power_unit_options = serializers.SerializerMethodField()
308 usage_count = serializers.SerializerMethodField()
310 class Meta:
311 model = EconomicsSettingsProfile
312 fields = (
313 "id",
314 "flowsheet",
315 "name",
316 "is_default",
317 "currency",
318 "location",
319 "basis_date",
320 "discount_rate_percent",
321 "project_lifetime_years",
322 "inflation_method",
323 "annual_operating_hours",
324 "tax_rate_percent",
325 "depreciation_enabled",
326 "default_depreciation_life_years",
327 "default_depreciation_salvage_percent",
328 "contingency_percent",
329 "electrical_upgrade_rate_amount",
330 "electrical_upgrade_rate_unit",
331 "electrical_upgrade_rate_unit_options",
332 "default_lang_factor",
333 "capital_index_series",
334 "operating_index_series",
335 "default_rate_overrides",
336 "default_rate_unit_options",
337 "manual_capex",
338 "manual_annual_opex",
339 "annual_heat_basis_mode",
340 "manual_annual_heat_basis",
341 "manual_annual_heat_basis_unit",
342 "manual_annual_heat_basis_unit_options",
343 "average_power_input",
344 "average_power_unit",
345 "average_power_unit_options",
346 "residual_value",
347 "notes",
348 "baseline_notes",
349 "usage_count",
350 "created_at",
351 "updated_at",
352 )
353 read_only_fields = (
354 "id",
355 "flowsheet",
356 "electrical_upgrade_rate_unit",
357 "electrical_upgrade_rate_unit_options",
358 "default_rate_unit_options",
359 "manual_annual_heat_basis_unit_options",
360 "average_power_unit_options",
361 "usage_count",
362 "created_at",
363 "updated_at",
364 )
366 @extend_schema_field(UnitOptionSerializer(many=True))
367 def get_electrical_upgrade_rate_unit_options(self, instance) -> list[dict[str, str]]:
368 return with_current_unit(ELECTRICAL_UPGRADE_RATE_UNIT_OPTIONS, instance.electrical_upgrade_rate_unit)
370 @extend_schema_field(DefaultRateUnitOptionsSerializer)
371 def get_default_rate_unit_options(self, instance) -> dict[str, list[dict[str, str]]]:
372 return default_rate_unit_options_by_type(currency=instance.currency or "NZD")
374 @extend_schema_field(UnitOptionSerializer(many=True))
375 def get_manual_annual_heat_basis_unit_options(self, instance) -> list[dict[str, str]]:
376 return list(ANNUAL_HEAT_BASIS_UNIT_OPTIONS)
378 @extend_schema_field(UnitOptionSerializer(many=True))
379 def get_average_power_unit_options(self, instance) -> list[dict[str, str]]:
380 return list(AVERAGE_POWER_UNIT_OPTIONS)
382 @extend_schema_field(serializers.IntegerField())
383 def get_usage_count(self, instance) -> int:
384 return getattr(instance, "usage_count", None) or instance.studies.count()
386 def validate(self, attrs):
387 attrs = super().validate(attrs)
388 tax_rate_percent = attrs.get("tax_rate_percent", getattr(self.instance, "tax_rate_percent", None))
389 annual_operating_hours = attrs.get(
390 "annual_operating_hours",
391 getattr(self.instance, "annual_operating_hours", None),
392 )
393 depreciation_enabled = attrs.get(
394 "depreciation_enabled",
395 getattr(self.instance, "depreciation_enabled", False),
396 )
397 default_depreciation_life_years = attrs.get(
398 "default_depreciation_life_years",
399 getattr(self.instance, "default_depreciation_life_years", None),
400 )
401 default_depreciation_salvage_percent = attrs.get(
402 "default_depreciation_salvage_percent",
403 getattr(self.instance, "default_depreciation_salvage_percent", None),
404 )
405 errors = {}
406 if not (attrs.get("name") or getattr(self.instance, "name", "")).strip(): 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 errors["name"] = "Profile name is required."
408 if tax_rate_percent is not None and not Decimal("0") <= tax_rate_percent <= Decimal("100"): 408 ↛ 409line 408 didn't jump to line 409 because the condition on line 408 was never true
409 errors["tax_rate_percent"] = "Tax rate must be between 0 and 100 percent."
410 if annual_operating_hours is not None and annual_operating_hours <= 0:
411 errors["annual_operating_hours"] = "Annual operating hours must be positive."
412 if depreciation_enabled and default_depreciation_life_years in (None, 0): 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 errors["default_depreciation_life_years"] = (
414 "Default equipment life is required when depreciation is enabled."
415 )
416 if ( 416 ↛ 420line 416 didn't jump to line 420 because the condition on line 416 was never true
417 default_depreciation_salvage_percent is not None
418 and not Decimal("0") <= default_depreciation_salvage_percent <= Decimal("100")
419 ):
420 errors["default_depreciation_salvage_percent"] = (
421 "Default residual value must be between 0 and 100 percent."
422 )
423 if errors:
424 raise serializers.ValidationError(errors)
425 return attrs
427 def validate_default_rate_overrides(self, value):
428 return EconomicsAssumptionsSerializer(context=self.context).validate_default_rate_overrides(value)
430 def create(self, validated_data):
431 _prefill_default_cpi_index_series(validated_data)
432 if validated_data.get("is_default") and validated_data.get("flowsheet") is not None:
433 EconomicsSettingsProfile.objects.filter(
434 flowsheet=validated_data["flowsheet"],
435 is_default=True,
436 ).update(is_default=False)
437 instance = super().create(validated_data)
438 return instance
440 def update(self, instance, validated_data):
441 if validated_data.get("is_default") is True:
442 EconomicsSettingsProfile.objects.filter(
443 flowsheet=instance.flowsheet,
444 is_default=True,
445 ).exclude(pk=instance.pk).update(is_default=False)
446 instance = super().update(instance, validated_data)
447 if instance.is_default:
448 EconomicsSettingsProfile.objects.filter(
449 flowsheet=instance.flowsheet,
450 is_default=True,
451 ).exclude(pk=instance.pk).update(is_default=False)
452 elif not EconomicsSettingsProfile.objects.filter( 452 ↛ 456line 452 didn't jump to line 456 because the condition on line 452 was never true
453 flowsheet=instance.flowsheet,
454 is_default=True,
455 ).exclude(pk=instance.pk).exists():
456 instance.is_default = True
457 instance.save(update_fields=["is_default"])
458 return instance
460class SettingsProfileCopyRequestSerializer(serializers.Serializer):
461 name = serializers.CharField(max_length=128, trim_whitespace=True)
462 is_default = serializers.BooleanField(required=False, default=False)
464 def validate_name(self, value: str) -> str:
465 source = self.context.get("source")
466 if (
467 isinstance(source, EconomicsSettingsProfile)
468 and EconomicsSettingsProfile.objects.filter(flowsheet=source.flowsheet, name=value).exists()
469 ):
470 raise serializers.ValidationError("A settings profile with this name already exists.")
471 return value