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