Coverage for backend/django/Economics/studies/serializers.py: 92%
194 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from django.core.exceptions import ObjectDoesNotExist
2from drf_spectacular.utils import extend_schema_field
3from rest_framework import serializers
5from core.auxiliary.models.Scenario import Scenario
6from core.validation import flowsheet_context, get_current_flowsheet
7from Economics.costing.capital.serializers import CapitalCostLineSerializer
8from Economics.costing.costable_items.serializers import CostableItemSerializer
9from Economics.costing.operating.serializers import OperatingCostLineSerializer
10from Economics.results.serializers import (
11 EconomicsResultRunSerializer,
12 ResultStatePayloadSerializer,
13 _warning_summaries,
14 serializer_result_run_classification,
15)
16from Economics.scheduling.serializers import SchedulePreviewSerializer
17from Economics.scheduling.services import schedule_preview
18from Economics.settings_profiles.models import EconomicsSettingsProfile
19from Economics.settings_profiles.serializers import EconomicsSettingsProfileSerializer
20from Economics.shared.serializer_base import FlowsheetScopedSerializer
21from Economics.studies.models import EconomicsStudy
22from Economics.studies.services.baseline_access import (
23 baseline_study_is_readable_by_user,
24 resolve_baseline_study,
25 set_baseline_study_reference,
26)
27from Economics.studies.services.project_scope import resolve_target_study_flowsheet
28from Economics.studies.services.study_configuration import (
29 StudyConfigurationValidationError,
30 normalize_study_configuration_attrs,
31 validate_study_configuration_attrs,
32)
33from Economics.settings_profiles.services.settings_profiles import get_or_create_default_settings_profile_for_project
36BASELINE_STUDY_NOT_ACCESSIBLE = "Selected baseline study is not accessible."
39class BaselineStudyRelatedField(serializers.PrimaryKeyRelatedField):
40 """Resolve study-baseline IDs without exposing whether hidden IDs exist."""
42 default_error_messages = {
43 "incorrect_type": "Incorrect type. Expected pk value, received {data_type}.",
44 }
46 def get_queryset(self):
47 # Study baselines may intentionally reference another readable flowsheet,
48 # so this lookup cannot use the request-scoped manager for the active flowsheet.
49 return EconomicsStudy._base_manager.select_related(
50 "flowsheet_state__flowsheet"
51 )
53 def get_attribute(self, instance):
54 """Resolve the stored stable-flow/lineage pair for representation."""
56 return resolve_baseline_study(instance)
58 def to_internal_value(self, data):
59 if self.pk_field is not None: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 data = self.pk_field.to_internal_value(data)
61 try:
62 baseline_study = self.get_queryset().get(pk=data)
63 except ObjectDoesNotExist as exc:
64 raise serializers.ValidationError(
65 BASELINE_STUDY_NOT_ACCESSIBLE,
66 code="does_not_exist",
67 ) from exc
68 except (TypeError, ValueError):
69 self.fail("incorrect_type", data_type=type(data).__name__)
71 request = self.context.get("request")
72 if request is not None and not baseline_study_is_readable_by_user(request.user, baseline_study):
73 raise serializers.ValidationError(
74 BASELINE_STUDY_NOT_ACCESSIBLE,
75 code="does_not_exist",
76 )
77 return baseline_study
80class EconomicsStudySerializer(FlowsheetScopedSerializer):
81 same_flowsheet_fields = ()
82 baseline_study = BaselineStudyRelatedField(
83 required=False,
84 allow_null=True,
85 )
86 settings_profile = serializers.PrimaryKeyRelatedField(
87 queryset=EconomicsSettingsProfile._base_manager.all(),
88 required=False,
89 allow_null=True,
90 )
91 schedule_scenario = serializers.PrimaryKeyRelatedField(
92 queryset=Scenario._base_manager.all(),
93 required=False,
94 allow_null=True,
95 )
96 target_flowsheet = serializers.IntegerField(required=False, write_only=True, allow_null=True)
97 result_state = serializers.SerializerMethodField()
98 settings_profile_name = serializers.CharField(source="settings_profile.name", read_only=True)
99 schedule_scenario_name = serializers.SerializerMethodField()
100 schedule_preview = serializers.SerializerMethodField()
101 baseline_study_name = serializers.SerializerMethodField()
102 baseline_study_flowsheet = serializers.SerializerMethodField()
103 baseline_study_flowsheet_name = serializers.SerializerMethodField()
104 flowsheet_name = serializers.SerializerMethodField()
106 class Meta:
107 model = EconomicsStudy
108 fields = (
109 "id",
110 "flowsheet",
111 "target_flowsheet",
112 "flowsheet_name",
113 "settings_profile",
114 "settings_profile_name",
115 "schedule_mode",
116 "schedule_scenario",
117 "schedule_scenario_name",
118 "schedule_preview",
119 "baseline_mode",
120 "baseline_study",
121 "baseline_study_name",
122 "baseline_study_flowsheet",
123 "baseline_study_flowsheet_name",
124 "name",
125 "description",
126 "result_state",
127 "created_at",
128 "updated_at",
129 )
130 read_only_fields = (
131 "id",
132 "flowsheet",
133 "flowsheet_name",
134 "settings_profile_name",
135 "schedule_scenario_name",
136 "schedule_preview",
137 "baseline_study_name",
138 "baseline_study_flowsheet",
139 "baseline_study_flowsheet_name",
140 "result_state",
141 "created_at",
142 "updated_at",
143 )
145 def validate(self, attrs):
146 attrs = super().validate(attrs)
147 current_flowsheet = get_current_flowsheet() or {}
148 current_flowsheet_id = current_flowsheet.get("flowsheet")
149 validation_flowsheet_id = self._validation_flowsheet_id(
150 attrs=attrs,
151 current_flowsheet_id=current_flowsheet_id,
152 )
153 attrs = normalize_study_configuration_attrs(
154 attrs=attrs,
155 instance=self.instance,
156 )
157 self._validate_settings_profile_project(
158 attrs,
159 flowsheet_id=validation_flowsheet_id,
160 )
161 self._validate_related_flowsheet(
162 attrs,
163 field_name="schedule_scenario",
164 flowsheet_id=validation_flowsheet_id,
165 )
166 try:
167 validate_study_configuration_attrs(
168 attrs=attrs,
169 instance=self.instance,
170 current_flowsheet_id=validation_flowsheet_id,
171 )
172 except StudyConfigurationValidationError as exc:
173 raise serializers.ValidationError(exc.errors) from exc
174 return attrs
176 def _validation_flowsheet_id(self, *, attrs, current_flowsheet_id):
177 """Return the flowsheet that create/update relations must belong to."""
179 if self.instance is not None:
180 if "target_flowsheet" in attrs: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true
181 raise serializers.ValidationError(
182 {"target_flowsheet": "Target flowsheet can only be selected when creating a study."}
183 )
184 self._validation_state_id = self.instance.flowsheet_state_id
185 return self.instance.flowsheet_state.flowsheet_id
187 request = self.context.get("request")
188 if current_flowsheet_id is None or request is None: 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true
189 raise serializers.ValidationError({"flowsheet": "A flowsheet context is required."})
190 target_flowsheet = resolve_target_study_flowsheet(
191 int(current_flowsheet_id),
192 attrs.get("target_flowsheet"),
193 request.user,
194 )
195 self._resolved_target_flowsheet = target_flowsheet
196 self._validation_state_id = target_flowsheet.current_state_id
197 return target_flowsheet.pk
199 def _validate_related_flowsheet(self, attrs, *, field_name: str, flowsheet_id: int | None) -> None:
200 """Require flowsheet-scoped relations to match the resolved study flowsheet."""
202 value = attrs.get(field_name)
203 if value is None and self.instance is not None:
204 value = getattr(self.instance, field_name, None)
205 validation_state_id = getattr(self, "_validation_state_id", None)
206 if (
207 value is not None
208 and validation_state_id is not None
209 and value.flowsheet_state_id != validation_state_id
210 ):
211 raise serializers.ValidationError(
212 {field_name: "Referenced row must belong to the study flowsheet."}
213 )
215 def _validate_settings_profile_project(self, attrs, *, flowsheet_id: int | None) -> None:
216 """Require selected settings profiles to belong to the study project."""
218 profile = attrs.get("settings_profile")
219 if profile is None and self.instance is not None:
220 profile = self.instance.settings_profile
221 if profile is None or flowsheet_id is None:
222 return
223 from core.auxiliary.models.Flowsheet import Flowsheet
225 project_id = Flowsheet.objects.filter(pk=flowsheet_id).values_list("project_id", flat=True).first()
226 if profile.project_id != project_id:
227 raise serializers.ValidationError(
228 {"settings_profile": "Referenced profile must belong to the study project."}
229 )
231 def create(self, validated_data):
232 validated_data.pop("target_flowsheet", None)
233 baseline_study = validated_data.pop("baseline_study", None)
234 target_flowsheet = getattr(self, "_resolved_target_flowsheet", None)
235 if target_flowsheet is not None: 235 ↛ 237line 235 didn't jump to line 237 because the condition on line 235 was always true
236 validated_data["flowsheet_state"] = target_flowsheet.current_state
237 if validated_data.get("settings_profile") is None:
238 flowsheet_state = validated_data.get("flowsheet_state")
239 if flowsheet_state is not None: 239 ↛ 243line 239 didn't jump to line 243 because the condition on line 239 was always true
240 validated_data["settings_profile"] = get_or_create_default_settings_profile_for_project(
241 flowsheet_state.flowsheet.project
242 )
243 instance = EconomicsStudy(**validated_data)
244 set_baseline_study_reference(instance, baseline_study)
245 request = self.context.get("request")
246 if target_flowsheet is not None and request is not None: 246 ↛ 257line 246 didn't jump to line 257 because the condition on line 246 was always true
247 # Sibling-flowsheet study creation is explicitly authorized during
248 # validation. Persist under that target context so mutation guards
249 # validate the selected current state rather than the source route.
250 with flowsheet_context(
251 target_flowsheet.pk,
252 request.user,
253 write_intent=True,
254 ):
255 instance.save()
256 else:
257 instance.save()
258 return instance
260 def update(self, instance, validated_data):
261 baseline_marker = object()
262 baseline_study = validated_data.pop("baseline_study", baseline_marker)
263 for field_name, value in validated_data.items():
264 setattr(instance, field_name, value)
265 if baseline_study is not baseline_marker: 265 ↛ 267line 265 didn't jump to line 267 because the condition on line 265 was always true
266 set_baseline_study_reference(instance, baseline_study)
267 instance.save()
268 return instance
270 @extend_schema_field(ResultStatePayloadSerializer)
271 def get_result_state(self, obj: EconomicsStudy) -> dict:
272 prefetched_runs = getattr(obj, "_economics_latest_result_runs", None)
273 run = prefetched_runs[0] if prefetched_runs else None
274 if prefetched_runs is None:
275 run = obj.result_runs.order_by("-created_at", "-pk").first()
276 if run is None:
277 return {
278 "run_id": None,
279 "status": "not_calculated",
280 "classification": "missing",
281 "warnings": [],
282 "requires_solve": False,
283 }
284 payload = run.warning_payload or {}
285 return {
286 "run_id": run.pk,
287 "status": run.status,
288 "classification": serializer_result_run_classification(run, self.context),
289 "completed_at": run.completed_at,
290 "warnings": _warning_summaries(payload),
291 "requires_solve": bool(payload.get("requires_solve", False)),
292 "latest_stale_reason": payload.get("latest_stale_reason", ""),
293 }
295 def get_schedule_scenario_name(self, obj: EconomicsStudy) -> str:
296 return obj.schedule_scenario.displayName if obj.schedule_scenario_id else ""
298 @extend_schema_field(serializers.CharField())
299 def get_flowsheet_name(self, obj: EconomicsStudy) -> str:
300 return obj.flowsheet_state.flowsheet.name
302 @extend_schema_field(SchedulePreviewSerializer)
303 def get_schedule_preview(self, obj: EconomicsStudy) -> dict:
304 return SchedulePreviewSerializer(schedule_preview(obj)).data
306 @extend_schema_field(serializers.CharField(allow_null=True))
307 def get_baseline_study_name(self, obj: EconomicsStudy) -> str | None:
308 baseline_study = resolve_baseline_study(obj)
309 return baseline_study.name if baseline_study is not None else None
311 @extend_schema_field(serializers.IntegerField(allow_null=True))
312 def get_baseline_study_flowsheet(self, obj: EconomicsStudy) -> int | None:
313 return obj.baseline_flowsheet_id
315 @extend_schema_field(serializers.CharField(allow_null=True))
316 def get_baseline_study_flowsheet_name(self, obj: EconomicsStudy) -> str | None:
317 baseline_study = resolve_baseline_study(obj)
318 return (
319 baseline_study.flowsheet_state.flowsheet.name
320 if baseline_study is not None
321 else None
322 )
325class EconomicsStudyFullSerializer(EconomicsStudySerializer):
326 assumptions = serializers.SerializerMethodField()
327 baseline = serializers.SerializerMethodField()
328 settings_profile_detail = serializers.SerializerMethodField()
329 costable_items = CostableItemSerializer(many=True, read_only=True)
330 capital_lines = CapitalCostLineSerializer(many=True, read_only=True)
331 operating_lines = OperatingCostLineSerializer(many=True, read_only=True)
332 current_result = serializers.SerializerMethodField()
334 class Meta(EconomicsStudySerializer.Meta):
335 fields = EconomicsStudySerializer.Meta.fields + (
336 "settings_profile_detail",
337 "assumptions",
338 "baseline",
339 "costable_items",
340 "capital_lines",
341 "operating_lines",
342 "current_result",
343 )
345 @extend_schema_field(EconomicsSettingsProfileSerializer(allow_null=True))
346 def get_settings_profile_detail(self, obj: EconomicsStudy) -> dict | None:
347 if obj.settings_profile_id is None: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 return None
349 return EconomicsSettingsProfileSerializer(obj.settings_profile, context=self.context).data
351 @extend_schema_field(EconomicsSettingsProfileSerializer(allow_null=True))
352 def get_assumptions(self, obj: EconomicsStudy) -> dict | None:
353 if obj.settings_profile_id is None: 353 ↛ 354line 353 didn't jump to line 354 because the condition on line 353 was never true
354 return None
355 return EconomicsSettingsProfileSerializer(obj.settings_profile, context=self.context).data
357 @extend_schema_field(EconomicsSettingsProfileSerializer(allow_null=True))
358 def get_baseline(self, obj: EconomicsStudy) -> dict | None:
359 if obj.settings_profile_id is None: 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true
360 return None
361 return EconomicsSettingsProfileSerializer(obj.settings_profile, context=self.context).data
363 @extend_schema_field(EconomicsResultRunSerializer(allow_null=True))
364 def get_current_result(self, obj: EconomicsStudy) -> dict | None:
365 prefetched_runs = getattr(obj, "_economics_current_result_runs", None)
366 if prefetched_runs is None:
367 run = obj.result_runs.filter(status="current").order_by("-created_at", "-pk").first()
368 else:
369 run = prefetched_runs[0] if prefetched_runs else None
370 if run is None:
371 return None
372 return EconomicsResultRunSerializer(run, context=self.context).data
374class EnableCostingRequestSerializer(serializers.Serializer):
375 simulation_object = serializers.IntegerField()
378class DuplicateStudyRequestSerializer(serializers.Serializer):
379 name = serializers.CharField(required=False, allow_blank=True, default="")