Coverage for backend/django/Economics/studies/services/study_configuration.py: 98%
102 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Normalize and validate Economics study baseline and schedule selections."""
3from __future__ import annotations
5from dataclasses import dataclass
6from typing import Any, TYPE_CHECKING
8from core.auxiliary.models.Scenario import Scenario
9from Economics.scheduling.composite.plan_services import validate_composite_schedule_selection
10from Economics.scheduling.services import validate_schedule_selection
11from Economics.shared.choices import EconomicsBaselineMode, EconomicsScheduleMode
13if TYPE_CHECKING:
14 from Economics.studies.models import EconomicsStudy
17@dataclass(frozen=True)
18class PendingStudyConfiguration:
19 """Resolved study configuration values used by model and serializer adapters."""
21 schedule_mode: str
22 schedule_scenario: Scenario | None
23 baseline_mode: str
24 baseline_study: EconomicsStudy | None
25 flowsheet_id: int | None
26 flowsheet_state_id: int | None
27 name: str
28 instance_pk: int | None
29 instance: EconomicsStudy | None
32class StudyConfigurationValidationError(ValueError):
33 """Structured validation errors for study baseline and schedule selections."""
35 def __init__(self, errors: dict[str, str]):
36 super().__init__("Invalid economics study configuration.")
37 self.errors = errors
40def normalize_study_configuration_attrs(
41 *,
42 attrs: dict[str, Any],
43 instance: EconomicsStudy | None,
44) -> dict[str, Any]:
45 """Return serializer attrs with mode-dependent stale selections cleared."""
47 normalized = dict(attrs)
48 schedule_mode = normalized.get(
49 "schedule_mode",
50 getattr(instance, "schedule_mode", EconomicsScheduleMode.STEADY_STATE),
51 )
52 if schedule_mode in {EconomicsScheduleMode.STEADY_STATE, EconomicsScheduleMode.COMPOSITE}:
53 normalized["schedule_scenario"] = None
55 baseline_mode = normalized.get(
56 "baseline_mode",
57 getattr(instance, "baseline_mode", EconomicsBaselineMode.MANUAL),
58 )
59 if baseline_mode == EconomicsBaselineMode.MANUAL:
60 normalized["baseline_study"] = None
61 return normalized
64def validate_study_configuration_attrs(
65 *,
66 attrs: dict[str, Any],
67 instance: EconomicsStudy | None,
68 current_flowsheet_id: int | None = None,
69) -> None:
70 """Validate normalized serializer attrs against the shared study policy."""
72 configuration = _configuration_from_attrs(
73 attrs=attrs,
74 instance=instance,
75 current_flowsheet_id=current_flowsheet_id,
76 )
77 _validate_configuration(configuration)
80def validate_study_configuration_for_model(study: EconomicsStudy) -> None:
81 """Validate a model instance before persistence."""
83 configuration = _configuration_from_attrs(
84 attrs={},
85 instance=study,
86 current_flowsheet_id=(
87 study.flowsheet_state.flowsheet_id
88 if study.flowsheet_state_id is not None
89 else None
90 ),
91 )
92 _validate_configuration(configuration)
95def _configuration_from_attrs(
96 *,
97 attrs: dict[str, Any],
98 instance: EconomicsStudy | None,
99 current_flowsheet_id: int | None,
100) -> PendingStudyConfiguration:
101 """Resolve update attrs and existing instance values into one policy input."""
103 schedule_mode = attrs.get(
104 "schedule_mode",
105 getattr(instance, "schedule_mode", EconomicsScheduleMode.STEADY_STATE),
106 )
107 baseline_mode = attrs.get(
108 "baseline_mode",
109 getattr(instance, "baseline_mode", EconomicsBaselineMode.MANUAL),
110 )
111 schedule_scenario = attrs.get(
112 "schedule_scenario",
113 getattr(instance, "schedule_scenario", None),
114 )
115 if "baseline_study" in attrs:
116 baseline_study = attrs["baseline_study"]
117 elif instance is not None:
118 from Economics.studies.services.baseline_access import resolve_baseline_study
120 baseline_study = resolve_baseline_study(instance)
121 else:
122 baseline_study = None
123 flowsheet_state = attrs.get("flowsheet_state")
124 instance_state = getattr(instance, "flowsheet_state", None)
125 resolved_state = instance_state or flowsheet_state
126 if resolved_state is None and current_flowsheet_id is not None:
127 from core.auxiliary.models.Flowsheet import Flowsheet
129 stable_flowsheet = (
130 Flowsheet.objects.select_related("current_state")
131 .filter(pk=current_flowsheet_id)
132 .first()
133 )
134 resolved_state = (
135 stable_flowsheet.current_state
136 if stable_flowsheet is not None
137 else None
138 )
139 flowsheet_id = (
140 getattr(getattr(resolved_state, "flowsheet", None), "pk", None)
141 or current_flowsheet_id
142 )
143 return PendingStudyConfiguration(
144 schedule_mode=schedule_mode,
145 schedule_scenario=schedule_scenario,
146 baseline_mode=baseline_mode,
147 baseline_study=baseline_study,
148 flowsheet_id=flowsheet_id,
149 flowsheet_state_id=getattr(resolved_state, "pk", None),
150 name=getattr(instance, "name", "") or attrs.get("name", ""),
151 instance_pk=getattr(instance, "pk", None),
152 instance=instance,
153 )
156def _validate_configuration(configuration: PendingStudyConfiguration) -> None:
157 """Apply the canonical baseline and schedule selection rules."""
159 errors: dict[str, str] = {}
160 if configuration.baseline_mode == EconomicsBaselineMode.STUDY:
161 _validate_study_baseline(configuration, errors)
162 if configuration.schedule_mode == EconomicsScheduleMode.SCENARIO:
163 _validate_scenario_schedule(configuration, errors)
164 if configuration.schedule_mode == EconomicsScheduleMode.COMPOSITE:
165 _validate_composite_schedule(configuration, errors)
166 if errors:
167 raise StudyConfigurationValidationError(errors)
170def _validate_study_baseline(
171 configuration: PendingStudyConfiguration,
172 errors: dict[str, str],
173) -> None:
174 """Require a distinct project-local baseline study when study mode is selected."""
176 baseline_study = configuration.baseline_study
177 if baseline_study is None:
178 errors["baseline_study"] = "Select a study to use as the financial baseline."
179 return
180 if configuration.instance_pk is not None and baseline_study.pk == configuration.instance_pk:
181 errors["baseline_study"] = "Select a different study as the financial baseline."
182 return
183 if not _baseline_study_matches_project_flowsheet(
184 baseline_study=baseline_study,
185 flowsheet_id=configuration.flowsheet_id,
186 ):
187 errors["baseline_study"] = "Select a baseline study from a non-template flowsheet in this project."
190def _validate_scenario_schedule(
191 configuration: PendingStudyConfiguration,
192 errors: dict[str, str],
193) -> None:
194 """Validate scenario schedule selection with a minimal candidate study."""
196 try:
197 validate_schedule_selection(
198 _schedule_candidate(configuration),
199 configuration.schedule_scenario,
200 )
201 except ValueError as exc:
202 errors["schedule_scenario"] = str(exc)
205def _validate_composite_schedule(
206 configuration: PendingStudyConfiguration,
207 errors: dict[str, str],
208) -> None:
209 """Validate that a persisted study has a usable saved composite schedule."""
211 if configuration.instance is None or configuration.instance_pk is None:
212 errors["schedule_mode"] = "Create the economics study before selecting a composite schedule."
213 return
214 try:
215 validate_composite_schedule_selection(configuration.instance)
216 except ValueError as exc:
217 errors["schedule_plan"] = str(exc)
220def _schedule_candidate(configuration: PendingStudyConfiguration) -> EconomicsStudy:
221 """Build the minimal study object expected by schedule validation."""
223 from Economics.studies.models import EconomicsStudy
225 candidate = EconomicsStudy(
226 flowsheet_state_id=configuration.flowsheet_state_id,
227 name=configuration.name,
228 )
229 candidate.schedule_mode = configuration.schedule_mode
230 candidate.schedule_scenario = configuration.schedule_scenario
231 if configuration.schedule_scenario is not None: 231 ↛ 233line 231 didn't jump to line 233 because the condition on line 231 was always true
232 candidate.schedule_scenario_id = configuration.schedule_scenario.pk
233 return candidate
236def _baseline_study_matches_project_flowsheet(
237 *,
238 baseline_study: EconomicsStudy,
239 flowsheet_id: int | None,
240) -> bool:
241 """Return whether a baseline study belongs to the target study's project."""
243 if flowsheet_id is None or baseline_study.flowsheet_state_id is None: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 return False
246 from Economics.studies.services.project_scope import study_is_in_current_project
248 return study_is_in_current_project(baseline_study, flowsheet_id)