Coverage for backend/django/Economics/scheduling/composite/plan_services.py: 95%
48 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Composite production-schedule plan services."""
3from __future__ import annotations
5from django.db import transaction
7from Economics.scheduling.composite.compiler import (
8 CompiledCompositeSchedule,
9 CompiledCompositeSource,
10 CompiledCompositeStep,
11 CompositeScheduleRuleInput,
12 compile_composite_schedule,
13 empty_invalid_composite_schedule,
14 saved_compiled_composite_schedule,
15)
16from .contracts import (
17 CompositeSchedulePlanPayload,
18 CompositeSchedulePlanRequest,
19 CompositeScheduleRulePayload,
20 CompiledScheduleStep,
21 ScheduleCoverageDiagnostic,
22 ScheduleDiagnosticSeverity,
23 ScheduleSourceSummary,
24)
25from Economics.shared.choices import EconomicsScheduleMode
26from Economics.studies.models import EconomicsSchedulePlan, EconomicsScheduleRule, EconomicsStudy
29class CompositeScheduleValidationError(ValueError):
30 """Raised when a saved composite schedule is invalid for study use."""
33def saved_schedule_plan_payload(study: EconomicsStudy) -> CompositeSchedulePlanPayload:
34 """Return the saved composite schedule, or an empty invalid payload."""
36 plan = getattr(study, "schedule_plan", None)
37 if plan is None:
38 return _payload_from_compiled(study=study, compiled=empty_invalid_composite_schedule(), mode=study.schedule_mode)
40 return _payload_from_compiled(
41 study=study,
42 compiled=saved_compiled_composite_schedule(study),
43 mode=study.schedule_mode,
44 )
47def preview_schedule_plan(
48 *,
49 study: EconomicsStudy,
50 request: CompositeSchedulePlanRequest,
51) -> CompositeSchedulePlanPayload:
52 """Validate and compile an unsaved composite schedule draft."""
54 return _payload_from_drafts(study=study, request=request, saved_ids=[], mode=EconomicsScheduleMode.COMPOSITE)
57def replace_schedule_plan(
58 *,
59 study: EconomicsStudy,
60 request: CompositeSchedulePlanRequest,
61) -> CompositeSchedulePlanPayload:
62 """Replace a study's composite schedule rules atomically and select the mode."""
64 payload = preview_schedule_plan(study=study, request=request)
65 _raise_if_invalid(payload)
66 with transaction.atomic():
67 plan, _ = EconomicsSchedulePlan.objects.get_or_create(
68 study=study,
69 defaults={"flowsheet_state": study.flowsheet_state},
70 )
71 if plan.flowsheet_state_id != study.flowsheet_state_id: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 plan.flowsheet_state = study.flowsheet_state
73 plan.save(update_fields=["flowsheet_state", "updated_at"])
74 plan.rules.all().delete()
75 for sort_order, rule in enumerate(request.rules):
76 EconomicsScheduleRule.objects.create(
77 flowsheet_state=study.flowsheet_state,
78 plan=plan,
79 source_scenario_id=rule.source_scenario,
80 label=rule.label,
81 days_mask=rule.days_mask,
82 start_time=rule.start_time,
83 sort_order=sort_order,
84 )
85 plan.save(update_fields=["updated_at"])
86 study.schedule_mode = EconomicsScheduleMode.COMPOSITE
87 study.schedule_scenario = None
88 study.save(update_fields=["schedule_mode", "schedule_scenario", "updated_at"])
89 study.refresh_from_db()
90 return saved_schedule_plan_payload(study)
93def validate_composite_schedule_selection(study: EconomicsStudy) -> None:
94 """Validate the saved composite schedule selected by a study."""
96 payload = saved_schedule_plan_payload(study)
97 _raise_if_invalid(payload)
100def _payload_from_drafts(
101 *,
102 study: EconomicsStudy,
103 request: CompositeSchedulePlanRequest,
104 saved_ids: list[int],
105 mode: str,
106) -> CompositeSchedulePlanPayload:
107 return _payload_from_compiled(
108 study=study,
109 compiled=compile_composite_schedule(
110 study=study,
111 rules=[_rule_input_from_request(rule) for rule in request.rules],
112 saved_ids=saved_ids,
113 ),
114 mode=mode,
115 )
118def _payload_from_compiled(
119 *,
120 study: EconomicsStudy,
121 compiled: CompiledCompositeSchedule,
122 mode: str,
123) -> CompositeSchedulePlanPayload:
124 return CompositeSchedulePlanPayload(
125 study_id=study.pk,
126 mode=mode,
127 status=compiled.status,
128 rules=[
129 CompositeScheduleRulePayload(
130 id=rule.saved_id,
131 source_scenario=rule.input.source_scenario,
132 source_scenario_name=rule.source_name,
133 label=rule.input.label or rule.source_name or "Schedule rule",
134 days_mask=rule.input.days_mask,
135 start_time=rule.input.start_time,
136 sort_order=rule.input.sort_order,
137 )
138 for rule in compiled.rules
139 ],
140 diagnostics=[
141 ScheduleCoverageDiagnostic(
142 code=diagnostic.code,
143 severity=diagnostic.severity,
144 message=diagnostic.message,
145 rule_indices=list(diagnostic.rule_indices),
146 )
147 for diagnostic in compiled.diagnostics
148 ],
149 timeline=[_schedule_step_payload(step) for step in compiled.weekly_steps],
150 sources=[_source_payload(source) for source in compiled.sources],
151 )
154def _rule_input_from_request(rule) -> CompositeScheduleRuleInput:
155 """Convert validated API request rule into compiler input."""
157 return CompositeScheduleRuleInput(
158 source_scenario=rule.source_scenario,
159 label=rule.label,
160 days_mask=rule.days_mask,
161 start_time=rule.start_time,
162 sort_order=rule.sort_order,
163 )
166def _source_payload(source: CompiledCompositeSource) -> ScheduleSourceSummary:
167 """Serialize a compiled source summary to the public API contract."""
169 return ScheduleSourceSummary(
170 id=source.id,
171 name=source.name,
172 row_count=source.row_count,
173 interval_value=source.interval_value,
174 interval_unit=source.interval_unit,
175 interval_hours=source.interval_hours,
176 schedule_length_hours=source.schedule_length_hours,
177 )
180def _schedule_step_payload(step: CompiledCompositeStep) -> CompiledScheduleStep:
181 """Serialize a compiled domain step to the public API contract."""
183 return CompiledScheduleStep(
184 index=step.index,
185 day=step.day,
186 elapsed_hours=step.elapsed_hours,
187 start_time=step.start_time,
188 end_time=step.end_time,
189 duration_hours=step.duration_hours,
190 source_kind=step.source_kind,
191 source_scenario=step.source_scenario,
192 source_scenario_name=step.source_scenario_name,
193 source_rule=step.source_rule,
194 source_rule_label=step.source_rule_label,
195 source_row_index=step.source_row_index,
196 source_row_elapsed_hours=step.source_row_elapsed_hours,
197 source_row_duration_hours=step.source_row_duration_hours,
198 )
201def _raise_if_invalid(payload: CompositeSchedulePlanPayload) -> None:
202 errors = [
203 diagnostic.message
204 for diagnostic in payload.diagnostics
205 if diagnostic.severity == ScheduleDiagnosticSeverity.ERROR
206 ]
207 if errors:
208 raise CompositeScheduleValidationError(errors[0])