Coverage for backend/django/Economics/scheduling/composite/compiler.py: 92%
381 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Composite schedule compilation and annualized traversal."""
3from __future__ import annotations
5from dataclasses import dataclass
6from datetime import time
7from decimal import Decimal
9from core.auxiliary.models.DataRow import DataRow
10from core.auxiliary.models.Scenario import Scenario
11from Economics.scheduling.composite.domain import (
12 ScheduleDiagnosticCode,
13 ScheduleDiagnosticSeverity,
14 ScheduleTimelineSourceKind,
15 ScheduleValidationStatus,
16 ScheduleWeekday,
17)
18from Economics.scheduling.durations import (
19 HOURS_PER_DAY,
20 HOURS_PER_WEEK,
21 annualized_remainder_cycle_steps,
22 cycle_duration_hours,
23)
24from Economics.scheduling.services import schedule_scenario_option
25from Economics.settings_profiles.services.settings_profiles import get_settings_profile
28@dataclass(frozen=True)
29class CompositeScheduleRuleInput:
30 """Transport-independent composite schedule rule input."""
32 source_scenario: int
33 label: str
34 days_mask: int
35 start_time: time
36 sort_order: int
39@dataclass(frozen=True)
40class CompiledCompositeSource:
41 """Resolved source-scenario metadata used by compiled composite schedules."""
43 id: int
44 name: str
45 row_count: int
46 interval_value: int
47 interval_unit: str
48 interval_hours: Decimal
49 schedule_length_hours: Decimal
52@dataclass(frozen=True)
53class CompiledCompositeDiagnostic:
54 """One validation diagnostic emitted by the composite compiler."""
56 code: ScheduleDiagnosticCode
57 severity: ScheduleDiagnosticSeverity
58 message: str
59 rule_indices: tuple[int, ...] = ()
62@dataclass(frozen=True)
63class CompiledCompositeRule:
64 """One resolved composite schedule rule."""
66 input: CompositeScheduleRuleInput
67 index: int
68 source: CompiledCompositeSource | None
69 source_name: str
70 saved_id: int | None = None
73@dataclass(frozen=True)
74class CompiledCompositeStep:
75 """One compiled weekly timetable block, including off periods."""
77 index: int
78 day: ScheduleWeekday
79 elapsed_hours: Decimal
80 start_time: time
81 end_time: time
82 duration_hours: Decimal
83 source_kind: ScheduleTimelineSourceKind
84 source_scenario: int | None = None
85 source_scenario_name: str = ""
86 source_rule: int | None = None
87 source_rule_label: str = ""
88 source_row_index: int | None = None
89 source_row_elapsed_hours: Decimal | None = None
90 source_row_duration_hours: Decimal | None = None
93@dataclass(frozen=True)
94class CompiledCompositeSchedule:
95 """Resolved and compiled composite schedule domain object."""
97 status: ScheduleValidationStatus
98 rules: tuple[CompiledCompositeRule, ...]
99 diagnostics: tuple[CompiledCompositeDiagnostic, ...]
100 weekly_steps: tuple[CompiledCompositeStep, ...]
101 sources: tuple[CompiledCompositeSource, ...]
104@dataclass(frozen=True)
105class CompositeAnnualStep:
106 """One active annualized composite step."""
108 index: int
109 row_index: int
110 elapsed_hours: Decimal
111 duration_hours: Decimal
112 source_step: CompiledCompositeStep
115@dataclass(frozen=True)
116class CompositeAnnualBucketSegment:
117 """Duration contributed by one composite source row inside a grouped bucket."""
119 row_index: int
120 duration_hours: Decimal
123@dataclass(frozen=True)
124class CompositeAnnualBucket:
125 """A bounded active composite timeline bucket."""
127 index: int
128 row_index: int
129 elapsed_hours: Decimal
130 duration_hours: Decimal
131 segments: tuple[CompositeAnnualBucketSegment, ...]
134@dataclass(frozen=True)
135class _RemainderActiveStep:
136 """One active step in the partial final annualized cycle."""
138 row_index: int
139 elapsed_hours: Decimal
140 duration_hours: Decimal
141 source_step: CompiledCompositeStep
144@dataclass(frozen=True)
145class CompositeAnnualTraversal:
146 """Shared annualized traversal facts for all composite timeline surfaces."""
148 active_steps: tuple[CompiledCompositeStep, ...]
149 full_cycle_hours: Decimal
150 full_cycles: int
151 remainder_active_steps: tuple[_RemainderActiveStep, ...]
152 full_active_count: int
153 active_step_count: int
155 def durations_by_step(self) -> dict[int, Decimal]:
156 """Return annual operating hours by compiled weekly step index."""
158 durations = {
159 step.index: step.duration_hours * Decimal(self.full_cycles)
160 for step in self.active_steps
161 }
162 for step in self.remainder_active_steps:
163 durations[step.row_index] = durations.get(step.row_index, Decimal("0")) + step.duration_hours
164 return durations
166 def steps(self) -> tuple[CompositeAnnualStep, ...]:
167 """Expand active composite steps over annual operating hours."""
169 steps: list[CompositeAnnualStep] = []
170 for cycle_index in range(self.full_cycles):
171 cycle_elapsed_hours = Decimal(cycle_index) * self.full_cycle_hours
172 for step in self.active_steps:
173 steps.append(
174 CompositeAnnualStep(
175 index=len(steps),
176 row_index=step.index,
177 elapsed_hours=cycle_elapsed_hours + step.elapsed_hours,
178 duration_hours=step.duration_hours,
179 source_step=step,
180 )
181 )
182 for step in self.remainder_active_steps:
183 steps.append(
184 CompositeAnnualStep(
185 index=len(steps),
186 row_index=step.row_index,
187 elapsed_hours=step.elapsed_hours,
188 duration_hours=step.duration_hours,
189 source_step=step.source_step,
190 )
191 )
192 return tuple(steps)
194 def buckets(self, bucket_count: int) -> tuple[CompositeAnnualBucket, ...]:
195 """Group annualized active composite steps into bounded timeline buckets."""
197 bucket_total = min(bucket_count, self.active_step_count)
198 buckets = []
199 for bucket_index in range(bucket_total):
200 start_step = bucket_index * self.active_step_count // bucket_total
201 end_step = (bucket_index + 1) * self.active_step_count // bucket_total
202 if start_step >= end_step: 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true
203 continue
204 elapsed_hours = _active_step_elapsed(
205 active_index=start_step,
206 traversal=self,
207 )
208 row_index = _active_step_row_index(
209 active_index=end_step - 1,
210 traversal=self,
211 )
212 segments = _active_bucket_segments(
213 start=start_step,
214 end=end_step,
215 traversal=self,
216 )
217 end_elapsed_hours = _active_step_end_elapsed(
218 active_index=end_step - 1,
219 traversal=self,
220 )
221 duration_hours = end_elapsed_hours - elapsed_hours
222 if duration_hours <= 0: 222 ↛ 223line 222 didn't jump to line 223 because the condition on line 222 was never true
223 continue
224 buckets.append(
225 CompositeAnnualBucket(
226 index=len(buckets),
227 row_index=row_index,
228 elapsed_hours=elapsed_hours,
229 duration_hours=duration_hours,
230 segments=segments,
231 )
232 )
233 return tuple(buckets)
236@dataclass(frozen=True)
237class _RuleInterval:
238 """One absolute weekly interval occupied by a resolved composite rule."""
240 rule: CompiledCompositeRule
241 start_hours: Decimal
242 end_hours: Decimal
245@dataclass(frozen=True)
246class _CompiledSegment:
247 """One source-row segment clipped to a single weekly day for display."""
249 rule: CompiledCompositeRule
250 day: ScheduleWeekday
251 start_hours: Decimal
252 end_hours: Decimal
253 source_row_index: int
254 source_elapsed_hours: Decimal
255 source_row_duration_hours: Decimal
258def compile_composite_schedule(
259 *,
260 study,
261 rules: list[CompositeScheduleRuleInput],
262 saved_ids: list[int],
263) -> CompiledCompositeSchedule:
264 """Resolve, validate, and compile a composite schedule draft."""
266 resolved_rules, diagnostics = _resolve_rules(study=study, rules=rules, saved_ids=saved_ids)
267 diagnostics.extend(_rule_invariant_diagnostics(resolved_rules))
268 intervals = _rule_intervals(resolved_rules)
269 diagnostics.extend(_overlap_diagnostics(intervals))
270 weekly_steps = _compiled_weekly_timeline(intervals)
271 sources = [
272 source
273 for source in {
274 rule.source.id: rule.source
275 for rule in resolved_rules
276 if rule.source is not None
277 }.values()
278 ]
279 status = (
280 ScheduleValidationStatus.INVALID
281 if any(diagnostic.severity == ScheduleDiagnosticSeverity.ERROR for diagnostic in diagnostics)
282 else ScheduleValidationStatus.VALID
283 )
284 return CompiledCompositeSchedule(
285 status=status,
286 rules=tuple(resolved_rules),
287 diagnostics=tuple(diagnostics),
288 weekly_steps=tuple(weekly_steps),
289 sources=tuple(sorted(sources, key=lambda source: (source.name, source.id))),
290 )
293def saved_compiled_composite_schedule(study) -> CompiledCompositeSchedule:
294 """Compile the saved schedule plan for runtime calculations."""
296 plan = getattr(study, "schedule_plan", None)
297 if plan is None: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 return empty_invalid_composite_schedule()
300 rules = list(plan.rules.select_related("source_scenario").order_by("sort_order", "pk"))
301 drafts = [_draft_from_saved_rule(rule) for rule in rules]
302 saved_ids = [rule.pk for rule in rules]
303 return compile_composite_schedule(study=study, rules=drafts, saved_ids=saved_ids)
306def empty_invalid_composite_schedule() -> CompiledCompositeSchedule:
307 """Return an invalid schedule object for studies without saved rules."""
309 return CompiledCompositeSchedule(
310 status=ScheduleValidationStatus.INVALID,
311 rules=(),
312 diagnostics=(
313 CompiledCompositeDiagnostic(
314 code=ScheduleDiagnosticCode.MISSING_SOURCE_SCENARIO,
315 severity=ScheduleDiagnosticSeverity.ERROR,
316 message="Add at least one timetable rule before using a composite schedule.",
317 ),
318 ),
319 weekly_steps=(),
320 sources=(),
321 )
324def active_composite_steps(compiled: CompiledCompositeSchedule) -> tuple[CompiledCompositeStep, ...]:
325 """Return active source-backed weekly steps from a compiled schedule."""
327 if compiled.status != ScheduleValidationStatus.VALID: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 return ()
329 return tuple(step for step in compiled.weekly_steps if is_active_composite_step(step))
332def composite_source_scenario_ids(compiled: CompiledCompositeSchedule) -> set[int]:
333 """Return active source scenario IDs used by a compiled schedule."""
335 return {
336 step.source_scenario
337 for step in active_composite_steps(compiled)
338 if step.source_scenario is not None
339 }
342def annual_composite_traversal(*, study, compiled: CompiledCompositeSchedule) -> CompositeAnnualTraversal | None:
343 """Return annualized traversal facts for a compiled composite schedule."""
345 annual_operating_hours = _annual_operating_hours(study)
346 weekly_steps = _positive_weekly_steps(compiled)
347 active_steps = tuple(step for step in weekly_steps if is_active_composite_step(step))
348 if annual_operating_hours is None or annual_operating_hours <= 0 or not weekly_steps or not active_steps:
349 return None
351 full_cycle_hours = cycle_duration_hours(weekly_steps, lambda step: step.duration_hours)
352 if full_cycle_hours <= 0: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 return None
354 full_cycles = int(annual_operating_hours // full_cycle_hours)
355 remainder_active_steps = _remainder_active_steps(
356 weekly_steps=weekly_steps,
357 annual_operating_hours=annual_operating_hours,
358 full_cycle_hours=full_cycle_hours,
359 full_cycles=full_cycles,
360 )
361 full_active_count = full_cycles * len(active_steps)
362 active_step_count = full_active_count + len(remainder_active_steps)
363 if active_step_count <= 0: 363 ↛ 364line 363 didn't jump to line 364 because the condition on line 363 was never true
364 return None
365 return CompositeAnnualTraversal(
366 active_steps=active_steps,
367 full_cycle_hours=full_cycle_hours,
368 full_cycles=full_cycles,
369 remainder_active_steps=remainder_active_steps,
370 full_active_count=full_active_count,
371 active_step_count=active_step_count,
372 )
375def is_active_composite_step(step: CompiledCompositeStep) -> bool:
376 """Return whether a compiled step represents active source production."""
378 return (
379 step.source_scenario is not None
380 and step.source_row_index is not None
381 and step.duration_hours > 0
382 )
385def _resolve_rules(
386 *,
387 study,
388 rules: list[CompositeScheduleRuleInput],
389 saved_ids: list[int],
390) -> tuple[list[CompiledCompositeRule], list[CompiledCompositeDiagnostic]]:
391 resolved = []
392 diagnostics = []
393 for index, rule in enumerate(rules):
394 source = None
395 source_name = ""
396 scenario = Scenario.objects.filter(pk=rule.source_scenario).first()
397 if scenario is None:
398 diagnostics.append(
399 CompiledCompositeDiagnostic(
400 code=ScheduleDiagnosticCode.MISSING_SOURCE_SCENARIO,
401 severity=ScheduleDiagnosticSeverity.ERROR,
402 message="Select a source scenario for each timetable rule.",
403 rule_indices=(index,),
404 )
405 )
406 elif scenario.flowsheet_state_id != study.flowsheet_state_id: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 source_name = scenario.displayName or "Production schedule"
408 diagnostics.append(
409 CompiledCompositeDiagnostic(
410 code=ScheduleDiagnosticCode.SOURCE_DIFFERENT_FLOWSHEET,
411 severity=ScheduleDiagnosticSeverity.ERROR,
412 message="Source scenarios must belong to this flowsheet.",
413 rule_indices=(index,),
414 )
415 )
416 else:
417 source_name = scenario.displayName or "Production schedule"
418 option = schedule_scenario_option(scenario)
419 if option is None: 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 diagnostics.append(
421 CompiledCompositeDiagnostic(
422 code=ScheduleDiagnosticCode.SOURCE_NOT_SOLVED,
423 severity=ScheduleDiagnosticSeverity.ERROR,
424 message="Run this source scenario before using it in a composite schedule.",
425 rule_indices=(index,),
426 )
427 )
428 else:
429 source = CompiledCompositeSource(
430 id=option.id,
431 name=option.name,
432 row_count=option.row_count,
433 interval_value=option.interval_value,
434 interval_unit=option.interval_unit,
435 interval_hours=option.interval_hours,
436 schedule_length_hours=option.schedule_length_hours,
437 )
438 if source.schedule_length_hours > HOURS_PER_WEEK:
439 diagnostics.append(
440 CompiledCompositeDiagnostic(
441 code=ScheduleDiagnosticCode.SOURCE_LONGER_THAN_WEEK,
442 severity=ScheduleDiagnosticSeverity.ERROR,
443 message="Source schedules must not be longer than one week.",
444 rule_indices=(index,),
445 )
446 )
447 resolved.append(
448 CompiledCompositeRule(
449 input=rule,
450 index=index,
451 source=source,
452 source_name=source_name,
453 saved_id=saved_ids[index] if index < len(saved_ids) else None,
454 )
455 )
456 return resolved, diagnostics
459def _draft_from_saved_rule(rule) -> CompositeScheduleRuleInput:
460 """Map persisted rule fields to the compiler's transport-independent input."""
462 return CompositeScheduleRuleInput(
463 source_scenario=rule.source_scenario_id or 0,
464 label=rule.label,
465 days_mask=rule.days_mask,
466 start_time=rule.start_time,
467 sort_order=rule.sort_order,
468 )
471def _overlap_diagnostics(intervals: list[_RuleInterval]) -> list[CompiledCompositeDiagnostic]:
472 diagnostics = []
473 reported_rule_sets = set()
474 interval_parts = [
475 (part_start, part_end, interval)
476 for interval in intervals
477 for part_start, part_end in _weekly_interval_parts(
478 interval.start_hours,
479 interval.end_hours,
480 )
481 ]
482 for left_index, (left_start, left_end, left) in enumerate(interval_parts):
483 for right_start, right_end, right in interval_parts[left_index + 1 :]:
484 if left.rule.index == right.rule.index and left.start_hours == right.start_hours: 484 ↛ 485line 484 didn't jump to line 485 because the condition on line 484 was never true
485 continue
486 if left_start < right_end and right_start < left_end:
487 rule_indices = tuple(sorted({left.rule.index, right.rule.index}))
488 if rule_indices in reported_rule_sets: 488 ↛ 489line 488 didn't jump to line 489 because the condition on line 488 was never true
489 continue
490 reported_rule_sets.add(rule_indices)
491 diagnostics.append(
492 CompiledCompositeDiagnostic(
493 code=ScheduleDiagnosticCode.OVERLAPPING_RULES,
494 severity=ScheduleDiagnosticSeverity.ERROR,
495 message="Timetable rules cannot overlap.",
496 rule_indices=rule_indices,
497 )
498 )
499 return diagnostics
502def _rule_invariant_diagnostics(rules: list[CompiledCompositeRule]) -> list[CompiledCompositeDiagnostic]:
503 """Validate rule invariants that must hold for request and saved-plan paths."""
505 diagnostics: list[CompiledCompositeDiagnostic] = []
506 for rule in rules:
507 if rule.input.days_mask < 1 or rule.input.days_mask > 127:
508 diagnostics.append(
509 CompiledCompositeDiagnostic(
510 code=ScheduleDiagnosticCode.INVALID_DAYS,
511 severity=ScheduleDiagnosticSeverity.ERROR,
512 message="Choose at least one day for each timetable rule.",
513 rule_indices=(rule.index,),
514 )
515 )
516 return diagnostics
519def _compiled_weekly_timeline(intervals: list[_RuleInterval]) -> list[CompiledCompositeStep]:
520 steps: list[CompiledCompositeStep] = []
521 scheduled_segments = _compiled_segments(intervals)
522 for day in ScheduleWeekday:
523 cursor_hours = Decimal("0")
524 day_segments = sorted(
525 (segment for segment in scheduled_segments if segment.day == day),
526 key=lambda segment: (segment.start_hours, segment.end_hours, segment.rule.index),
527 )
528 for segment in day_segments:
529 if cursor_hours < segment.start_hours:
530 steps.append(_off_step(index=len(steps), day=day, start_hours=cursor_hours, end_hours=segment.start_hours))
531 steps.append(_scheduled_segment_step(index=len(steps), segment=segment))
532 if segment.end_hours > cursor_hours:
533 cursor_hours = segment.end_hours
534 if cursor_hours < HOURS_PER_DAY:
535 steps.append(_off_step(index=len(steps), day=day, start_hours=cursor_hours, end_hours=HOURS_PER_DAY))
536 return steps
539def _compiled_segments(intervals: list[_RuleInterval]) -> list[_CompiledSegment]:
540 """Expand rule placements into source-row segments clipped to weekly days."""
542 segments: list[_CompiledSegment] = []
543 source_rows_by_scenario: dict[int, list[int]] = {}
544 for interval in intervals:
545 if interval.rule.source is None: 545 ↛ 546line 545 didn't jump to line 546 because the condition on line 545 was never true
546 continue
547 source_id = interval.rule.source.id
548 if source_id not in source_rows_by_scenario:
549 source_rows_by_scenario[source_id] = _schedule_row_indices(source_id)
550 row_indices = source_rows_by_scenario[source_id]
551 segments.extend(_compiled_interval_segments(interval=interval, row_indices=row_indices))
552 return segments
555def _rule_intervals(rules: list[CompiledCompositeRule]) -> list[_RuleInterval]:
556 """Return absolute weekly intervals occupied by each rule's source sequence."""
558 intervals: list[_RuleInterval] = []
559 for rule in rules:
560 if rule.source is None or rule.source.schedule_length_hours <= 0:
561 continue
562 for day in ScheduleWeekday:
563 if not rule.input.days_mask & _day_mask(day):
564 continue
565 start_hours = _day_start_hours(day) + _time_offset_hours(rule.input.start_time)
566 intervals.append(
567 _RuleInterval(
568 rule=rule,
569 start_hours=start_hours,
570 end_hours=start_hours + rule.source.schedule_length_hours,
571 )
572 )
573 return intervals
576def _compiled_interval_segments(*, interval: _RuleInterval, row_indices: list[int]) -> list[_CompiledSegment]:
577 """Split one rule interval by source rows, days, and weekly boundaries."""
579 rule = interval.rule
580 if rule.source is None: 580 ↛ 581line 580 didn't jump to line 581 because the condition on line 580 was never true
581 return []
582 if not row_indices: 582 ↛ 583line 582 didn't jump to line 583 because the condition on line 582 was never true
583 return []
585 segments: list[_CompiledSegment] = []
586 cursor_hours = interval.start_hours
587 source_position = 0
588 source_elapsed_hours = Decimal("0")
589 while cursor_hours < interval.end_hours:
590 row_index = row_indices[source_position % len(row_indices)]
591 row_elapsed = source_elapsed_hours % rule.source.schedule_length_hours
592 row_end_hours = min(cursor_hours + rule.source.interval_hours, interval.end_hours)
593 row_duration = row_end_hours - cursor_hours
594 segment_start = cursor_hours
595 while segment_start < row_end_hours:
596 weekly_start = segment_start % HOURS_PER_WEEK
597 day_index = int(weekly_start // HOURS_PER_DAY)
598 day_start_hours = Decimal(day_index) * HOURS_PER_DAY
599 segment_end = min(row_end_hours, segment_start + (HOURS_PER_DAY - (weekly_start - day_start_hours)))
600 weekly_end = segment_end % HOURS_PER_WEEK
601 if segment_end > segment_start and weekly_end == Decimal("0"):
602 weekly_end = HOURS_PER_WEEK
603 segments.append(
604 _CompiledSegment(
605 rule=rule,
606 day=ScheduleWeekday(day_index),
607 start_hours=weekly_start - day_start_hours,
608 end_hours=weekly_end - day_start_hours,
609 source_row_index=row_index,
610 source_elapsed_hours=row_elapsed + (segment_start - cursor_hours),
611 source_row_duration_hours=rule.source.interval_hours,
612 )
613 )
614 segment_start = segment_end
615 cursor_hours += row_duration
616 source_elapsed_hours += row_duration
617 source_position += 1
618 return segments
621def _scheduled_segment_step(*, index: int, segment: _CompiledSegment) -> CompiledCompositeStep:
622 return CompiledCompositeStep(
623 index=index,
624 day=segment.day,
625 elapsed_hours=_day_start_hours(segment.day) + segment.start_hours,
626 start_time=_hours_to_time(segment.start_hours),
627 end_time=_hours_to_time(segment.end_hours),
628 duration_hours=segment.end_hours - segment.start_hours,
629 source_kind=ScheduleTimelineSourceKind.SCHEDULED_SOURCE,
630 source_scenario=segment.rule.source.id if segment.rule.source is not None else None,
631 source_scenario_name=segment.rule.source.name if segment.rule.source is not None else "",
632 source_rule=segment.rule.saved_id,
633 source_rule_label=segment.rule.input.label or segment.rule.source_name,
634 source_row_index=segment.source_row_index,
635 source_row_elapsed_hours=segment.source_elapsed_hours,
636 source_row_duration_hours=segment.source_row_duration_hours,
637 )
640def _off_step(*, index: int, day: ScheduleWeekday, start_hours: Decimal, end_hours: Decimal) -> CompiledCompositeStep:
641 return CompiledCompositeStep(
642 index=index,
643 day=day,
644 elapsed_hours=_day_start_hours(day) + start_hours,
645 start_time=_hours_to_time(start_hours),
646 end_time=_hours_to_time(end_hours),
647 duration_hours=end_hours - start_hours,
648 source_kind=ScheduleTimelineSourceKind.OFF_PERIOD,
649 )
652def _positive_weekly_steps(compiled: CompiledCompositeSchedule) -> tuple[CompiledCompositeStep, ...]:
653 if compiled.status != ScheduleValidationStatus.VALID: 653 ↛ 654line 653 didn't jump to line 654 because the condition on line 653 was never true
654 return ()
655 return tuple(step for step in compiled.weekly_steps if step.duration_hours > 0)
658def _annual_operating_hours(study) -> Decimal | None:
659 profile = get_settings_profile(study)
660 return profile.annual_operating_hours if profile is not None else None
663def _remainder_active_steps(
664 *,
665 weekly_steps,
666 annual_operating_hours: Decimal,
667 full_cycles: int,
668 full_cycle_hours: Decimal,
669) -> tuple[_RemainderActiveStep, ...]:
670 """Return active steps in the partial final cycle."""
672 return tuple(
673 _RemainderActiveStep(
674 row_index=step.source.index,
675 elapsed_hours=step.elapsed_hours,
676 duration_hours=step.duration_hours,
677 source_step=step.source,
678 )
679 for step in annualized_remainder_cycle_steps(
680 cycle_steps=weekly_steps,
681 duration_accessor=lambda source_step: source_step.duration_hours,
682 active_step_predicate=is_active_composite_step,
683 annual_operating_hours=annual_operating_hours,
684 full_cycle_hours=full_cycle_hours,
685 full_cycles=full_cycles,
686 )
687 if step.active
688 )
691def _active_step_elapsed(
692 *,
693 active_index: int,
694 traversal: CompositeAnnualTraversal,
695) -> Decimal:
696 if active_index < traversal.full_active_count:
697 cycle_index, step_position = divmod(active_index, len(traversal.active_steps))
698 return (Decimal(cycle_index) * traversal.full_cycle_hours) + traversal.active_steps[step_position].elapsed_hours
699 return traversal.remainder_active_steps[active_index - traversal.full_active_count].elapsed_hours
702def _active_step_end_elapsed(
703 *,
704 active_index: int,
705 traversal: CompositeAnnualTraversal,
706) -> Decimal:
707 if active_index < traversal.full_active_count:
708 cycle_index, step_position = divmod(active_index, len(traversal.active_steps))
709 step = traversal.active_steps[step_position]
710 return (Decimal(cycle_index) * traversal.full_cycle_hours) + step.elapsed_hours + step.duration_hours
711 step = traversal.remainder_active_steps[active_index - traversal.full_active_count]
712 return step.elapsed_hours + step.duration_hours
715def _active_step_row_index(
716 *,
717 active_index: int,
718 traversal: CompositeAnnualTraversal,
719) -> int:
720 if active_index < traversal.full_active_count:
721 return traversal.active_steps[active_index % len(traversal.active_steps)].index
722 return traversal.remainder_active_steps[active_index - traversal.full_active_count].row_index
725def _active_bucket_segments(
726 *,
727 start: int,
728 end: int,
729 traversal: CompositeAnnualTraversal,
730) -> tuple[CompositeAnnualBucketSegment, ...]:
731 durations_by_row: dict[int, Decimal] = {}
732 if start < traversal.full_active_count:
733 full_end = min(end, traversal.full_active_count)
734 _add_full_cycle_segments(
735 durations_by_row=durations_by_row,
736 start=start,
737 end=full_end,
738 active_steps=traversal.active_steps,
739 )
740 if end > traversal.full_active_count:
741 remainder_start = max(start, traversal.full_active_count) - traversal.full_active_count
742 remainder_end = end - traversal.full_active_count
743 for step in traversal.remainder_active_steps[remainder_start:remainder_end]:
744 durations_by_row[step.row_index] = durations_by_row.get(step.row_index, Decimal("0")) + step.duration_hours
745 return tuple(
746 CompositeAnnualBucketSegment(row_index=row_index, duration_hours=duration_hours)
747 for row_index, duration_hours in durations_by_row.items()
748 if duration_hours > 0
749 )
752def _add_full_cycle_segments(
753 *,
754 durations_by_row: dict[int, Decimal],
755 start: int,
756 end: int,
757 active_steps,
758) -> None:
759 if end <= start: 759 ↛ 760line 759 didn't jump to line 760 because the condition on line 759 was never true
760 return
761 active_count = len(active_steps)
762 start_cycle, start_position = divmod(start, active_count)
763 end_cycle, end_position = divmod(end, active_count)
764 if start_cycle == end_cycle:
765 _add_active_step_slice(durations_by_row, active_steps[start_position:end_position])
766 return
768 _add_active_step_slice(durations_by_row, active_steps[start_position:])
769 complete_cycles = end_cycle - start_cycle - 1
770 if complete_cycles > 0: 770 ↛ 771line 770 didn't jump to line 771 because the condition on line 770 was never true
771 for step in active_steps:
772 duration_hours = step.duration_hours * Decimal(complete_cycles)
773 durations_by_row[step.index] = durations_by_row.get(step.index, Decimal("0")) + duration_hours
774 if end_position:
775 _add_active_step_slice(durations_by_row, active_steps[:end_position])
778def _add_active_step_slice(durations_by_row: dict[int, Decimal], steps) -> None:
779 for step in steps:
780 durations_by_row[step.index] = durations_by_row.get(step.index, Decimal("0")) + step.duration_hours
783def _day_mask(day: ScheduleWeekday) -> int:
784 return 1 << int(day)
787def _day_start_hours(day: ScheduleWeekday) -> Decimal:
788 return Decimal(int(day)) * HOURS_PER_DAY
791def _weekly_interval_parts(start_hours: Decimal, end_hours: Decimal) -> list[tuple[Decimal, Decimal]]:
792 """Split an absolute interval into comparable parts on the weekly cycle."""
794 if end_hours <= start_hours: 794 ↛ 795line 794 didn't jump to line 795 because the condition on line 794 was never true
795 return []
796 duration = end_hours - start_hours
797 if duration >= HOURS_PER_WEEK:
798 return [(Decimal("0"), HOURS_PER_WEEK)]
799 start = start_hours % HOURS_PER_WEEK
800 end = start + duration
801 if end <= HOURS_PER_WEEK: 801 ↛ 803line 801 didn't jump to line 803 because the condition on line 801 was always true
802 return [(start, end)]
803 return [(start, HOURS_PER_WEEK), (Decimal("0"), end - HOURS_PER_WEEK)]
806def _time_offset_hours(value: time) -> Decimal:
807 total_seconds = (
808 Decimal(value.hour * 3600)
809 + Decimal(value.minute * 60)
810 + Decimal(value.second)
811 + (Decimal(value.microsecond) / Decimal("1000000"))
812 )
813 return total_seconds / Decimal("3600")
816def _hours_to_time(value: Decimal) -> time:
817 if value == HOURS_PER_DAY:
818 return time.max
819 total_seconds = int((value * Decimal("3600")).to_integral_value())
820 hours, remainder = divmod(total_seconds, 3600)
821 minutes, seconds = divmod(remainder, 60)
822 return time(hour=hours, minute=minutes, second=seconds)
825def _schedule_row_indices(scenario_id: int) -> list[int]:
826 """Load source row indices in the order each child schedule is solved."""
828 return list(
829 DataRow.objects.filter(scenario_id=scenario_id)
830 .order_by("index", "pk")
831 .values_list("index", flat=True)
832 )