Coverage for backend/django/Economics/scheduling/services.py: 80%
132 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1from __future__ import annotations
3from dataclasses import dataclass
4from datetime import datetime
5from decimal import Decimal, ROUND_HALF_UP
6from typing import TYPE_CHECKING
8from django.db.models import Exists, Max, OuterRef
10from core.auxiliary.models.DataRow import DataRow
11from core.auxiliary.models.Scenario import Scenario
12from core.auxiliary.models.Solution import Solution
13from core.auxiliary.services.scenario_time_series import is_mss_time_series
14from Economics.scheduling.durations import interval_hours
15from Economics.settings_profiles.services.settings_profiles import get_settings_profile
16from Economics.shared.choices import EconomicsScheduleMode
18if TYPE_CHECKING:
19 from Economics.studies.models import EconomicsStudy
22@dataclass(frozen=True)
23class SchedulePreview:
24 """Display-ready schedule timing details for an economics study."""
26 mode: str
27 scenario_id: int | None
28 scenario_name: str
29 row_count: int
30 interval_value: int | None
31 interval_unit: str
32 interval_hours: Decimal | None
33 schedule_length_hours: Decimal | None
34 schedule_length_display_value: Decimal | None
35 schedule_length_display_unit: str
36 annual_operating_hours: Decimal | None
37 operating_cycles: Decimal | None
38 compatible: bool
39 message: str
42@dataclass(frozen=True)
43class ScheduleScenarioOption:
44 """One selectable production-schedule scenario for an economics study."""
46 id: int
47 name: str
48 row_count: int
49 interval_value: int
50 interval_unit: str
51 interval_hours: Decimal
52 schedule_length_hours: Decimal
55def compatible_schedule_scenarios(study: EconomicsStudy) -> tuple[ScheduleScenarioOption, ...]:
56 """Return solved MSS time-series scenarios that can schedule this study."""
58 scenarios = (
59 Scenario.objects.filter(flowsheet_state=study.flowsheet_state)
60 .order_by("displayName", "pk")
61 )
62 options = []
63 for scenario in scenarios:
64 option = schedule_scenario_option(scenario)
65 if option is not None:
66 options.append(option)
67 return tuple(options)
70def schedule_scenario_option(scenario: Scenario) -> ScheduleScenarioOption | None:
71 """Return a schedule option when the scenario satisfies the v1 contract."""
73 if not is_mss_time_series(scenario):
74 return None
75 row_indices, input_created_at_max = _scenario_row_metadata(scenario)
76 if not row_indices: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true
77 return None
78 solved_indices = _scenario_solved_indices(
79 scenario,
80 input_created_at_min=input_created_at_max,
81 )
82 if not row_indices.issubset(solved_indices):
83 return None
84 interval_hours = scenario_interval_hours(scenario)
85 schedule_length_hours = interval_hours * Decimal(len(row_indices))
86 return ScheduleScenarioOption(
87 id=scenario.pk,
88 name=scenario.displayName or "Production schedule",
89 row_count=len(row_indices),
90 interval_value=scenario.mss_time_series_interval,
91 interval_unit=scenario.mss_time_series_unit,
92 interval_hours=interval_hours,
93 schedule_length_hours=schedule_length_hours,
94 )
97def validate_schedule_selection(study: EconomicsStudy, scenario: Scenario | None) -> None:
98 """Validate the selected schedule scenario using product-facing errors."""
100 if study.schedule_mode != EconomicsScheduleMode.SCENARIO: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true
101 return
102 if scenario is None: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 raise ValueError("Select a production schedule for this study.")
104 if scenario.flowsheet_state_id != study.flowsheet_state_id: 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true
105 raise ValueError("The selected production schedule belongs to a different flowsheet.")
106 if not is_mss_time_series(scenario): 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 raise ValueError("Select a multi-solve scenario with time series enabled.")
108 if not _scenario_row_indices(scenario): 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true
109 raise ValueError("The selected production schedule does not contain any operating states.")
110 if schedule_scenario_option(scenario) is None:
111 raise ValueError(_incomplete_schedule_message(scenario))
114def schedule_preview(study: EconomicsStudy) -> SchedulePreview:
115 """Return timing details for the study's selected schedule."""
117 annual_operating_hours = _study_annual_operating_hours(study)
118 if study.schedule_mode != EconomicsScheduleMode.SCENARIO:
119 if study.schedule_mode == EconomicsScheduleMode.COMPOSITE:
120 from Economics.scheduling.composite.plan_services import saved_schedule_plan_payload
122 payload = saved_schedule_plan_payload(study)
123 scheduled_steps = [
124 step
125 for step in payload.timeline
126 if step.source_kind.value == "scheduled_source"
127 ]
128 message = ""
129 if payload.status.value != "valid": 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 message = payload.diagnostics[0].message if payload.diagnostics else "Complete the composite schedule."
131 return SchedulePreview(
132 mode=EconomicsScheduleMode.COMPOSITE,
133 scenario_id=None,
134 scenario_name="Composite schedule",
135 row_count=len(scheduled_steps),
136 interval_value=None,
137 interval_unit="",
138 interval_hours=None,
139 schedule_length_hours=None,
140 schedule_length_display_value=None,
141 schedule_length_display_unit="",
142 annual_operating_hours=annual_operating_hours,
143 operating_cycles=None,
144 compatible=payload.status.value == "valid",
145 message=message,
146 )
147 return SchedulePreview(
148 mode=EconomicsScheduleMode.STEADY_STATE,
149 scenario_id=None,
150 scenario_name="",
151 row_count=0,
152 interval_value=None,
153 interval_unit="",
154 interval_hours=None,
155 schedule_length_hours=None,
156 schedule_length_display_value=None,
157 schedule_length_display_unit="",
158 annual_operating_hours=annual_operating_hours,
159 operating_cycles=None,
160 compatible=True,
161 message="Study uses steady-state operating assumptions.",
162 )
164 scenario = study.schedule_scenario
165 option = schedule_scenario_option(scenario) if scenario is not None else None
166 if scenario is None or option is None: 166 ↛ 167line 166 didn't jump to line 167 because the condition on line 166 was never true
167 message = "Select a production schedule for this study."
168 if scenario is not None:
169 try:
170 validate_schedule_selection(study, scenario)
171 except ValueError as exc:
172 message = str(exc)
173 return SchedulePreview(
174 mode=EconomicsScheduleMode.SCENARIO,
175 scenario_id=getattr(scenario, "pk", None),
176 scenario_name=getattr(scenario, "displayName", "") or "",
177 row_count=0,
178 interval_value=getattr(scenario, "mss_time_series_interval", None),
179 interval_unit=getattr(scenario, "mss_time_series_unit", "") or "",
180 interval_hours=None,
181 schedule_length_hours=None,
182 schedule_length_display_value=None,
183 schedule_length_display_unit="",
184 annual_operating_hours=annual_operating_hours,
185 operating_cycles=None,
186 compatible=False,
187 message=message,
188 )
190 operating_cycles = (
191 annual_operating_hours / option.schedule_length_hours
192 if annual_operating_hours is not None and option.schedule_length_hours > 0
193 else None
194 )
195 display_value, display_unit = schedule_length_display(option.schedule_length_hours)
196 return SchedulePreview(
197 mode=EconomicsScheduleMode.SCENARIO,
198 scenario_id=option.id,
199 scenario_name=option.name,
200 row_count=option.row_count,
201 interval_value=option.interval_value,
202 interval_unit=option.interval_unit,
203 interval_hours=option.interval_hours,
204 schedule_length_hours=option.schedule_length_hours,
205 schedule_length_display_value=display_value,
206 schedule_length_display_unit=display_unit,
207 annual_operating_hours=annual_operating_hours,
208 operating_cycles=operating_cycles,
209 compatible=True,
210 message="",
211 )
214def scenario_interval_hours(scenario: Scenario) -> Decimal:
215 return interval_hours(scenario.mss_time_series_interval, scenario.mss_time_series_unit)
218def schedule_length_display(hours: Decimal) -> tuple[Decimal, str]:
219 if hours < Decimal("1"):
220 return _display_decimal(hours * Decimal("60")), "minutes"
221 if hours < Decimal("48"): 221 ↛ 223line 221 didn't jump to line 223 because the condition on line 221 was always true
222 return _display_decimal(hours), "hours"
223 days = hours / Decimal("24")
224 if days < Decimal("90"):
225 return _display_decimal(days), "days"
226 months = days / Decimal("30")
227 if months < Decimal("24"):
228 return _display_decimal(months), "months"
229 return _display_decimal(days / Decimal("365")), "years"
232def _scenario_row_indices(scenario: Scenario) -> set[int]:
233 """Return all schedule row indices expected to have solved values."""
234 row_indices, _ = _scenario_row_metadata(scenario)
235 return row_indices
238def _scenario_row_metadata(scenario: Scenario) -> tuple[set[int], datetime | None]:
239 """Return row indices plus the newest input-row timestamp for stale-solve checks."""
240 rows = DataRow._base_manager.filter(
241 flowsheet_state=scenario.flowsheet_state,
242 scenario=scenario,
243 )
244 return set(
245 rows.values_list("index", flat=True)
246 ), rows.aggregate(created_at_max=Max("created_at"))["created_at_max"]
249def _scenario_solved_indices(scenario: Scenario, *, input_created_at_min=None) -> set[int]:
250 """Return solve indices, optionally limited to solves newer than the schedule input rows."""
251 solutions = Solution._base_manager.filter(
252 flowsheet_state=scenario.flowsheet_state,
253 scenario=scenario,
254 solve_index=OuterRef("index"),
255 solve_index__isnull=False,
256 )
257 if input_created_at_min is not None:
258 solutions = solutions.filter(created_at__gte=input_created_at_min)
259 return set(
260 DataRow._base_manager.filter(
261 flowsheet_state=scenario.flowsheet_state,
262 scenario=scenario,
263 )
264 .annotate(has_solution=Exists(solutions))
265 .filter(has_solution=True)
266 .values_list("index", flat=True)
267 )
270def _incomplete_schedule_message(scenario: Scenario) -> str:
271 """Distinguish unsolved schedules from schedules that were edited after solving."""
272 row_indices, input_created_at_max = _scenario_row_metadata(scenario)
273 solved_indices = _scenario_solved_indices(scenario)
274 if row_indices.issubset(solved_indices) and input_created_at_max is not None:
275 return "Run this production schedule again before using it for economics."
276 return "Solve every operating state before using this scenario as a production schedule."
279def _study_annual_operating_hours(study: EconomicsStudy) -> Decimal | None:
280 """Read the study's annual operating-hours setting used to scale schedule previews."""
281 profile = get_settings_profile(study)
282 return profile.annual_operating_hours if profile is not None else None
285def _display_decimal(value: Decimal) -> Decimal:
286 """Quantize schedule preview durations for stable API display."""
287 return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)