Coverage for backend/django/Economics/scheduling/durations.py: 87%
92 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 decimal import Decimal
5from typing import Callable, Generic, Hashable, Sequence, TypeVar
7from core.auxiliary.models.Scenario import ScenarioTimeSeriesUnitEnum
10T = TypeVar("T")
11K = TypeVar("K", bound=Hashable)
14HOURS_PER_DAY = Decimal("24")
15HOURS_PER_WEEK = HOURS_PER_DAY * Decimal("7")
17HOURS_PER_UNIT = {
18 ScenarioTimeSeriesUnitEnum.Seconds.value: Decimal("0.0002777777777777777777777777778"),
19 ScenarioTimeSeriesUnitEnum.Minutes.value: Decimal("0.01666666666666666666666666667"),
20 ScenarioTimeSeriesUnitEnum.Hours.value: Decimal("1"),
21 ScenarioTimeSeriesUnitEnum.Days.value: HOURS_PER_DAY,
22 ScenarioTimeSeriesUnitEnum.Months.value: Decimal("720"),
23}
26@dataclass(frozen=True)
27class AnnualizedCycleStep(Generic[T]):
28 """One source step placed onto an annualized operating-hours timeline."""
30 source: T
31 cycle_position: int
32 elapsed_hours: Decimal
33 duration_hours: Decimal
34 active: bool
37def hours_per_unit(unit: str | ScenarioTimeSeriesUnitEnum) -> Decimal:
38 """Return the Decimal hour multiplier for an MSS time-series unit."""
40 return HOURS_PER_UNIT.get(_unit_key(unit), Decimal("1"))
43def interval_hours(interval_value: int | Decimal, unit: str | ScenarioTimeSeriesUnitEnum) -> Decimal:
44 """Convert an MSS interval to hours using economics scheduling units."""
46 return Decimal(interval_value) * hours_per_unit(unit)
49def cycle_duration_hours(
50 cycle_steps: Sequence[T],
51 duration_accessor: Callable[[T], Decimal],
52) -> Decimal:
53 """Return the full wall-clock duration of one cycle."""
55 return sum((_positive_duration(duration_accessor(step)) for step in cycle_steps), Decimal("0"))
58def annualized_active_step_count(
59 *,
60 cycle_steps: Sequence[T],
61 duration_accessor: Callable[[T], Decimal],
62 active_step_predicate: Callable[[T], bool],
63 annual_operating_hours: Decimal,
64) -> int:
65 """Count active annualized steps without expanding full cycles."""
67 full_cycle_hours = cycle_duration_hours(cycle_steps, duration_accessor)
68 if annual_operating_hours <= 0 or full_cycle_hours <= 0: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 return 0
71 full_cycles = int(annual_operating_hours // full_cycle_hours)
72 active_steps_per_cycle = sum(1 for step in cycle_steps if active_step_predicate(step))
73 remainder_active_steps = sum(
74 1
75 for step in annualized_remainder_cycle_steps(
76 cycle_steps=cycle_steps,
77 duration_accessor=duration_accessor,
78 active_step_predicate=active_step_predicate,
79 annual_operating_hours=annual_operating_hours,
80 full_cycle_hours=full_cycle_hours,
81 full_cycles=full_cycles,
82 )
83 if step.active
84 )
85 return (full_cycles * active_steps_per_cycle) + remainder_active_steps
88def annualized_active_duration_by_key(
89 *,
90 cycle_steps: Sequence[T],
91 duration_accessor: Callable[[T], Decimal],
92 active_step_predicate: Callable[[T], bool],
93 key_accessor: Callable[[T], K],
94 annual_operating_hours: Decimal,
95) -> dict[K, Decimal]:
96 """Aggregate active durations using full-cycle math plus the final partial cycle."""
98 full_cycle_hours = cycle_duration_hours(cycle_steps, duration_accessor)
99 if annual_operating_hours <= 0 or full_cycle_hours <= 0: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 return {}
102 full_cycles = int(annual_operating_hours // full_cycle_hours)
103 durations: dict[K, Decimal] = {}
104 for step in cycle_steps:
105 if not active_step_predicate(step):
106 continue
107 key = key_accessor(step)
108 duration_hours = _positive_duration(duration_accessor(step)) * Decimal(full_cycles)
109 durations[key] = durations.get(key, Decimal("0")) + duration_hours
111 for step in annualized_remainder_cycle_steps(
112 cycle_steps=cycle_steps,
113 duration_accessor=duration_accessor,
114 active_step_predicate=active_step_predicate,
115 annual_operating_hours=annual_operating_hours,
116 full_cycle_hours=full_cycle_hours,
117 full_cycles=full_cycles,
118 ):
119 if not step.active: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 continue
121 key = key_accessor(step.source)
122 durations[key] = durations.get(key, Decimal("0")) + step.duration_hours
123 return durations
126def annualized_remainder_cycle_steps(
127 *,
128 cycle_steps: Sequence[T],
129 duration_accessor: Callable[[T], Decimal],
130 active_step_predicate: Callable[[T], bool],
131 annual_operating_hours: Decimal,
132 full_cycle_hours: Decimal | None = None,
133 full_cycles: int | None = None,
134) -> tuple[AnnualizedCycleStep[T], ...]:
135 """Return the partial final-cycle steps after whole cycles have elapsed."""
137 resolved_cycle_hours = (
138 full_cycle_hours
139 if full_cycle_hours is not None
140 else cycle_duration_hours(cycle_steps, duration_accessor)
141 )
142 if annual_operating_hours <= 0 or resolved_cycle_hours <= 0: 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true
143 return ()
145 resolved_full_cycles = (
146 full_cycles
147 if full_cycles is not None
148 else int(annual_operating_hours // resolved_cycle_hours)
149 )
150 remainder_hours = annual_operating_hours - (Decimal(resolved_full_cycles) * resolved_cycle_hours)
151 elapsed_hours = Decimal(resolved_full_cycles) * resolved_cycle_hours
152 return tuple(
153 _cycle_step
154 for _cycle_step in _walk_cycle_once(
155 cycle_steps=cycle_steps,
156 duration_accessor=duration_accessor,
157 active_step_predicate=active_step_predicate,
158 remaining_hours=remainder_hours,
159 elapsed_hours=elapsed_hours,
160 )
161 )
164def annualize_cycle(
165 *,
166 cycle_steps: Sequence[T],
167 duration_accessor: Callable[[T], Decimal],
168 active_step_predicate: Callable[[T], bool],
169 annual_operating_hours: Decimal,
170) -> tuple[AnnualizedCycleStep[T], ...]:
171 """Expand one cycle over annual operating hours while prorating the final step."""
173 if annual_operating_hours <= 0 or not cycle_steps: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 return ()
176 full_cycle_hours = cycle_duration_hours(cycle_steps, duration_accessor)
177 if full_cycle_hours <= 0: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true
178 return ()
180 steps: list[AnnualizedCycleStep[T]] = []
181 remaining_hours = annual_operating_hours
182 elapsed_hours = Decimal("0")
183 while remaining_hours > 0:
184 cycle_steps_once = _walk_cycle_once(
185 cycle_steps=cycle_steps,
186 duration_accessor=duration_accessor,
187 active_step_predicate=active_step_predicate,
188 remaining_hours=remaining_hours,
189 elapsed_hours=elapsed_hours,
190 )
191 if not cycle_steps_once: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 break
193 steps.extend(cycle_steps_once)
194 consumed_hours = sum((step.duration_hours for step in cycle_steps_once), Decimal("0"))
195 remaining_hours -= consumed_hours
196 elapsed_hours += consumed_hours
197 return tuple(steps)
200def _walk_cycle_once(
201 *,
202 cycle_steps: Sequence[T],
203 duration_accessor: Callable[[T], Decimal],
204 active_step_predicate: Callable[[T], bool],
205 remaining_hours: Decimal,
206 elapsed_hours: Decimal,
207) -> tuple[AnnualizedCycleStep[T], ...]:
208 steps: list[AnnualizedCycleStep[T]] = []
209 for position, source_step in enumerate(cycle_steps):
210 if remaining_hours <= 0:
211 break
212 source_duration = _positive_duration(duration_accessor(source_step))
213 if source_duration <= 0: 213 ↛ 214line 213 didn't jump to line 214 because the condition on line 213 was never true
214 continue
215 duration_hours = min(source_duration, remaining_hours)
216 steps.append(
217 AnnualizedCycleStep(
218 source=source_step,
219 cycle_position=position,
220 elapsed_hours=elapsed_hours,
221 duration_hours=duration_hours,
222 active=active_step_predicate(source_step),
223 )
224 )
225 remaining_hours -= duration_hours
226 elapsed_hours += duration_hours
227 return tuple(steps)
230def _positive_duration(value: Decimal) -> Decimal:
231 return value if value > 0 else Decimal("0")
234def _unit_key(unit: str | ScenarioTimeSeriesUnitEnum) -> str:
235 return getattr(unit, "value", unit)