Mia.Wu/scheduler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dual device on weekend feature

Browse Source
This commit is contained in:
Mia Wu
2026-07-12 18:34:28 -04:00
parent 9e100d60d0
commit 31caa13c3d
6 changed files with 214 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app.py
+4
View File
@@ -75,6 +75,8 @@ class CompileScheduleRequest(BaseModel):
start_date: str | None = Field(default=None, description="YYYY-MM-DD")
rule: str = ""
daytime_testing_today: bool = False
dual_device_weekend_start_enabled: bool = False
dual_device_weekend_start_dates: list[str] = Field(default_factory=list)
top_priority_tests: list[str] = Field(default_factory=list)
lowest_priority_tests: list[str] = Field(default_factory=list)
@@ -378,6 +380,8 @@ def compile_schedule_endpoint(request: CompileScheduleRequest) -> dict[str, Any]
start_date=request.start_date,
holiday_dates=holiday_dates,
daytime_testing_today=request.daytime_testing_today,
dual_device_weekend_start_enabled=request.dual_device_weekend_start_enabled,
dual_device_window_start_dates=set(item.strip() for item in request.dual_device_weekend_start_dates if item.strip()),
)
try:
backend/scheduler.py
+33 -4
View File
@@ -2,8 +2,8 @@ from __future__ import annotations
import os
from dataclasses import dataclass
from datetime import date
from test_window import get_shift_sequence_with_capacity, next_window_start_date
from datetime import date, timedelta
from test_window import get_shift_sequence_with_capacity, is_off_day, next_window_start_date
from test_bundle import Test, TestBundle, build_test_bundles, bundle_pair_lookup
DUT = (os.getenv("DUT") or "DUT").strip()
@@ -36,6 +36,8 @@ class Scheduler:
start_date: str | None = None,
holiday_dates: set[str] = set(),
daytime_testing_today: bool = False,
dual_device_weekend_start_enabled: bool = False,
dual_device_window_start_dates: set[str] | None = None,
priority_weight: int = 100,
):
self.top_priority_tests = top_priority_tests
@@ -45,6 +47,8 @@ class Scheduler:
self.start_date = date.fromisoformat(start_date) if start_date else date.today()
self.holiday_dates = holiday_dates
self.daytime_testing_today = daytime_testing_today
self.dual_device_weekend_start_enabled = dual_device_weekend_start_enabled
self.dual_device_window_start_dates = dual_device_window_start_dates or set()
self.priority_weight = priority_weight
self.pending_dut_mirror: list[TestBundle] = []
self.pending_ref_mirror: list[TestBundle] = []
@@ -102,6 +106,10 @@ class Scheduler:
active_unscheduled = dut_unscheduled if window_device == DUT else ref_unscheduled
shifts, capacity = get_shift_sequence_with_capacity(cursor_date, self.holiday_dates, self.daytime_testing_today)
cursor_date_key = cursor_date.isoformat()
dual_device_window = cursor_date_key in self.dual_device_window_start_dates
if not dual_device_window and self.dual_device_weekend_start_enabled:
dual_device_window = self._is_weekend_start_day(cursor_date)
# Evaluate both devices without mutating queue state, then commit once.
mirrored_bundles, knapsack_bundles, remaining_time = self._select_bundles(
@@ -114,6 +122,7 @@ class Scheduler:
selected_bundles = mirrored_bundles + knapsack_bundles
if len(selected_bundles) == 0:
window_device = REF if window_device == DUT else DUT
active_unscheduled = dut_unscheduled if window_device == DUT else ref_unscheduled
mirrored_bundles, knapsack_bundles, remaining_time = self._select_bundles(
window_device,
capacity,
@@ -125,6 +134,20 @@ class Scheduler:
if len(selected_bundles) == 0:
break
primary_device = window_device
if dual_device_window and remaining_time > 0:
secondary_device = REF if primary_device == DUT else DUT
secondary_unscheduled = dut_unscheduled if secondary_device == DUT else ref_unscheduled
secondary_mirrored, secondary_knapsack, remaining_time = self._select_bundles(
secondary_device,
remaining_time,
secondary_unscheduled,
tc,
mutate=True,
)
selected_bundles.extend(secondary_mirrored + secondary_knapsack)
# Create a mirror of the other device for the next window
bundle_pairs = bundle_pair_lookup(knapsack_bundles, dut_unscheduled, ref_unscheduled)
@@ -134,9 +157,10 @@ class Scheduler:
pending_target = self.pending_dut_mirror
existing_pending = {(b.index, b.device) for b in pending_target}
selected_bundle_keys = {(b.index, b.device) for b in selected_bundles}
for pair_bundle in bundle_pairs:
pair_key = (pair_bundle.index, pair_bundle.device)
if pair_key in self.scheduled_bundle_keys or pair_key in existing_pending:
if pair_key in self.scheduled_bundle_keys or pair_key in existing_pending or pair_key in selected_bundle_keys:
continue
pending_target.append(pair_bundle)
existing_pending.add(pair_key)
@@ -148,7 +172,7 @@ class Scheduler:
capacity_minutes=capacity,
remaining_minutes=remaining_time,
assigned_config=tc,
assigned_device=window_device,
assigned_device=primary_device,
)
self._place_bundles_in_window(window, selected_bundles)
@@ -170,6 +194,11 @@ class Scheduler:
return None
def _is_weekend_start_day(self, current_date: date) -> bool:
if is_off_day(current_date, self.holiday_dates):
return False
return is_off_day(current_date + timedelta(days=1), self.holiday_dates)
def _select_bundles(
self,
device: str,
backend/test_scheduler_dual_device.py
+111
View File
@@ -0,0 +1,111 @@
from __future__ import annotations
from pathlib import Path
import sys
import unittest
sys.path.append(str(Path(__file__).resolve().parent))
from scheduler import DUT, REF, Scheduler
from test_bundle import Test
def _test_config() -> dict[str, dict[str, str | None]]:
return {
"Station 1": {
"test_point": "T1D",
"sta": "STA5",
}
}
def _make_test(test_id: str, device: str, minutes: int) -> Test:
return Test(
test_id=test_id,
device=device,
test_type="P2P",
rotation=None,
rx_tx=None,
has_coe_pair=False,
coe_pairing=[],
config=_test_config(),
throttled=False,
estimated_minutes=minutes,
)
class SchedulerDualDeviceWeekendStartTests(unittest.TestCase):
def _first_window_devices(self, entries, start_date: str) -> set[str]:
return {
entry.device
for entry in entries
if entry.scheduled_date == start_date and entry.shift_index == 3
}
def test_weekend_start_window_stays_single_device_when_disabled(self) -> None:
start_date = "2026-07-10" # Friday
tests = [
_make_test("P2PRXAX001", DUT, 900),
_make_test("P2PRXAX001", REF, 300),
]
scheduler = Scheduler(
tests=tests,
top_priority_tests=set(),
start_date=start_date,
holiday_dates=set(),
dual_device_weekend_start_enabled=False,
)
err = scheduler.compile_schedule()
self.assertIsNone(err)
entries = scheduler.get_schedule()
self.assertEqual(self._first_window_devices(entries, start_date), {DUT})
def test_weekend_start_window_runs_both_devices_when_enabled(self) -> None:
start_date = "2026-07-10" # Friday
tests = [
_make_test("P2PRXAX001", DUT, 900),
_make_test("P2PRXAX001", REF, 300),
]
scheduler = Scheduler(
tests=tests,
top_priority_tests=set(),
start_date=start_date,
holiday_dates=set(),
dual_device_weekend_start_enabled=True,
)
err = scheduler.compile_schedule()
self.assertIsNone(err)
entries = scheduler.get_schedule()
self.assertEqual(self._first_window_devices(entries, start_date), {DUT, REF})
def test_holiday_adjusted_weekend_start_runs_both_devices(self) -> None:
start_date = "2026-07-09" # Thursday with Friday as holiday => weekend-start window
holidays = {"2026-07-10"}
tests = [
_make_test("P2PRXAX001", DUT, 700),
_make_test("P2PRXAX001", REF, 400),
]
scheduler = Scheduler(
tests=tests,
top_priority_tests=set(),
start_date=start_date,
holiday_dates=holidays,
dual_device_weekend_start_enabled=True,
)
err = scheduler.compile_schedule()
self.assertIsNone(err)
entries = scheduler.get_schedule()
self.assertEqual(self._first_window_devices(entries, start_date), {DUT, REF})
if __name__ == "__main__":
unittest.main()