Coverage for backend/django/Economics/studies/services/configuration_api.py: 76%
116 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""API workflow helpers for study configuration endpoints."""
3from __future__ import annotations
5from django.core.exceptions import ObjectDoesNotExist
6from pydantic import ValidationError as PydanticValidationError
7from rest_framework.exceptions import NotFound, ValidationError
9from Economics.costing.operating.stream_properties import sync_project_default_operating_line_rates_for_study
10from Economics.results.services.comparison.config import save_comparison_config
11from Economics.results.services.comparison.contracts import (
12 ComparisonValidationError,
13 EconomicsStudyComparisonRequest,
14)
15from Economics.results.services.comparison.payloads import saved_comparison_payload
16from Economics.scheduling.composite.contracts import CompositeSchedulePlanRequest
17from Economics.scheduling.composite.plan_services import (
18 CompositeScheduleValidationError,
19 preview_schedule_plan,
20 replace_schedule_plan,
21 saved_schedule_plan_payload,
22)
23from Economics.scheduling.serializers import ScheduleScenarioOptionSerializer
24from Economics.scheduling.services import compatible_schedule_scenarios
25from Economics.settings_profiles.serializers import EconomicsAssumptionsSerializer, EconomicsBaselineSerializer
26from Economics.settings_profiles.services.settings_profiles import (
27 assumptions_from_settings_profile,
28 baseline_from_settings_profile,
29 update_settings_profile_from_assumptions,
30 update_settings_profile_from_baseline,
31)
32from Economics.shared.access import require_write_access
33from Economics.studies.models import EconomicsStudy
34from Economics.studies.services.result_staleness import mark_study_stale
37def assumptions_data(
38 *,
39 study: EconomicsStudy,
40 method: str,
41 data,
42 user,
43 serializer_context: dict,
44) -> dict:
45 """Return or mutate legacy assumptions backed by the selected settings profile."""
47 if method == "GET":
48 assumptions = assumptions_from_settings_profile(study)
49 if assumptions is not None: 49 ↛ 51line 49 didn't jump to line 51 because the condition on line 49 was always true
50 return EconomicsAssumptionsSerializer(assumptions, context=serializer_context).data
51 try:
52 assumptions = study.assumptions
53 except ObjectDoesNotExist as exc:
54 raise NotFound("Economics assumptions have not been configured for this study.") from exc
55 return EconomicsAssumptionsSerializer(assumptions, context=serializer_context).data
57 require_write_access(user)
58 try:
59 instance = study.assumptions
60 except ObjectDoesNotExist:
61 instance = None
62 serializer = EconomicsAssumptionsSerializer(
63 instance,
64 data={**data, "study": study.pk},
65 partial=method == "PATCH",
66 context=serializer_context,
67 )
68 serializer.is_valid(raise_exception=True)
69 field_names = None if method != "PATCH" else set(serializer.validated_data)
70 assumptions = serializer.save()
71 update_settings_profile_from_assumptions(assumptions, field_names=field_names)
72 sync_project_default_operating_line_rates_for_study(study, reason="economics_assumptions_api_saved")
73 mark_study_stale(study, reason="economics_assumptions_api_saved")
74 return EconomicsAssumptionsSerializer(assumptions, context=serializer_context).data
77def baseline_data(
78 *,
79 study: EconomicsStudy,
80 method: str,
81 data,
82 user,
83 serializer_context: dict,
84) -> dict:
85 """Return or mutate legacy baseline values backed by the selected settings profile."""
87 if method == "GET":
88 baseline = baseline_from_settings_profile(study)
89 if baseline is not None: 89 ↛ 91line 89 didn't jump to line 91 because the condition on line 89 was always true
90 return EconomicsBaselineSerializer(baseline, context=serializer_context).data
91 try:
92 baseline = study.baseline
93 except ObjectDoesNotExist as exc:
94 raise NotFound("Economics baseline has not been configured for this study.") from exc
95 return EconomicsBaselineSerializer(baseline, context=serializer_context).data
97 require_write_access(user)
98 try:
99 instance = study.baseline
100 except ObjectDoesNotExist:
101 instance = None
102 serializer = EconomicsBaselineSerializer(
103 instance,
104 data={**data, "study": study.pk},
105 partial=method == "PATCH",
106 context=serializer_context,
107 )
108 serializer.is_valid(raise_exception=True)
109 field_names = None if method != "PATCH" else set(serializer.validated_data)
110 baseline = serializer.save()
111 update_settings_profile_from_baseline(baseline, field_names=field_names)
112 mark_study_stale(study, reason="economics_baseline_api_saved")
113 return EconomicsBaselineSerializer(baseline, context=serializer_context).data
116def comparison_data(
117 *,
118 study: EconomicsStudy,
119 user,
120 method: str,
121 data,
122) -> dict:
123 """Return or replace a study's saved comparison configuration."""
125 if method == "PUT":
126 require_write_access(user)
127 try:
128 comparison_request = EconomicsStudyComparisonRequest.model_validate(data)
129 except PydanticValidationError as exc:
130 raise ValidationError(exc.errors()) from exc
131 try:
132 save_comparison_config(
133 active_study=study,
134 user=user,
135 comparison_study_ids=comparison_request.comparison_study_ids,
136 baseline_mode=comparison_request.baseline_mode,
137 baseline_study_id=comparison_request.baseline_study_id,
138 )
139 except ComparisonValidationError as exc:
140 raise ValidationError({"comparison_study_ids": str(exc)}) from exc
141 return saved_comparison_payload(active_study=study, user=user).model_dump(mode="json")
144def schedule_options_data(study: EconomicsStudy) -> list[dict]:
145 """Return selectable schedule scenarios for a study."""
147 return ScheduleScenarioOptionSerializer(compatible_schedule_scenarios(study), many=True).data
150def schedule_plan_data(
151 *,
152 study: EconomicsStudy,
153 user,
154 method: str,
155 data,
156) -> dict:
157 """Return or replace a study's saved composite schedule plan."""
159 if method == "PUT":
160 require_write_access(user)
161 try:
162 schedule_request = CompositeSchedulePlanRequest.model_validate(data)
163 except PydanticValidationError as exc:
164 raise _schedule_plan_request_validation_error(exc) from exc
165 try:
166 payload = replace_schedule_plan(study=study, request=schedule_request)
167 except CompositeScheduleValidationError as exc:
168 raise ValidationError({"rules": str(exc)}) from exc
169 mark_study_stale(study, reason="economics_schedule_selected")
170 return payload.model_dump(mode="json")
171 return saved_schedule_plan_payload(study).model_dump(mode="json")
174def schedule_plan_preview_data(*, study: EconomicsStudy, data) -> dict:
175 """Return a composite schedule preview without mutating the saved plan."""
177 try:
178 schedule_request = CompositeSchedulePlanRequest.model_validate(data)
179 except PydanticValidationError as exc:
180 raise _schedule_plan_request_validation_error(exc) from exc
181 return preview_schedule_plan(study=study, request=schedule_request).model_dump(mode="json")
184def _schedule_plan_request_validation_error(exc: PydanticValidationError) -> ValidationError:
185 """Convert Pydantic internals into product-language schedule errors."""
187 messages = [_schedule_plan_error_message(error) for error in exc.errors()]
188 return ValidationError({"rules": list(dict.fromkeys(messages))})
191def _schedule_plan_error_message(error: dict) -> str:
192 """Return the user-facing error for one invalid composite schedule field."""
194 loc = tuple(error.get("loc") or ())
195 error_type = str(error.get("type") or "")
196 if "label" in loc:
197 return "Rule labels must be 128 characters or fewer."
198 if loc == ("rules",) and error_type in {"too_short", "too_long"}: 198 ↛ 200line 198 didn't jump to line 200 because the condition on line 198 was always true
199 return "Add between 1 and 64 timetable rules."
200 if "source_scenario" in loc:
201 return "Select a source scenario for each timetable rule."
202 if "days_mask" in loc:
203 return "Choose at least one day for each timetable rule."
204 if "start_time" in loc:
205 return "Check each rule's start time."
206 message = str(error.get("msg") or "")
207 if message.startswith("Value error, "):
208 message = message.removeprefix("Value error, ")
209 return message or "Check the composite schedule rules."