frontend

backend/app.py

@@ -1,9 +1,11 @@
+from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Any
 from datetime import date, datetime
 import os
 
 from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel, Field
 
 import db
@@ -15,9 +17,7 @@ from scheduler import SchedulerTest, compile_schedule, remove_from_active, reset
 APP_ROOT = Path(__file__).resolve().parent
 DB_PATH = APP_ROOT / "scheduler.db"
 DUT = os.getenv("DUT", "CGW453").strip()
-REF = os.getenv("REF", "CGW452").strip() 
-
-app = FastAPI(title="Scheduler API", version="0.1.0")
+REF = os.getenv("REF", "CGW452").strip()
 
 
 class LoadTestsRequest(BaseModel):
@@ -40,10 +40,25 @@ class RemoveActiveTestsRequest(BaseModel):
 	test_ids: list[str] = Field(default_factory=list)
 
 
-@app.on_event("startup")
-def on_startup() -> None:
+class SaveHolidaysRequest(BaseModel):
+	dates: list[str] = Field(default_factory=list)
+
+
+@asynccontextmanager
+async def lifespan(application: FastAPI):
 	db.init_db(DB_PATH)
 	graph.reset_graph_state()
+	yield
+
+
+app = FastAPI(title="Scheduler API", version="0.1.0", lifespan=lifespan)
+
+app.add_middleware(
+	CORSMiddleware,
+	allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"],
+	allow_methods=["*"],
+	allow_headers=["*"],
+)
 
 
 @app.get("/api/health")
@@ -154,6 +169,18 @@ def compile_schedule_endpoint(request: CompileScheduleRequest) -> dict[str, Any]
 	}
 
 
+@app.post("/api/holidays")
+def save_holidays(request: SaveHolidaysRequest) -> dict[str, Any]:
+	dates = [d.strip() for d in request.dates if d.strip()]
+	db.upsert_holidays(dates, DB_PATH)
+	return {"status": "saved", "count": len(dates)}
+
+
+@app.get("/api/holidays")
+def get_holidays() -> dict[str, Any]:
+	return {"dates": sorted(db.list_holidays(DB_PATH))}
+
+
 @app.get("/api/schedule/week")
 def get_schedule_week(start: str | None = None) -> dict[str, Any]:
 	week_start = start or date.today().isoformat()
@@ -174,6 +201,7 @@ def get_schedule_week(start: str | None = None) -> dict[str, Any]:
 				"sequence_in_shift": row.sequence_in_shift,
 				"test_type": row.test_type,
 				"rotation": row.rotation,
+				"config": row.config,
 				"status": row.status,
 				"priority": row.priority,
 				"estimated_minutes": row.estimated_minutes,
@@ -181,3 +209,8 @@ def get_schedule_week(start: str | None = None) -> dict[str, Any]:
 			for row in rows
 		],
 	}
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)
+	

backend/db.py

@@ -47,6 +47,7 @@ class ScheduleRow:
 	sequence_in_shift: int
 	test_type: str
 	rotation: str | None
+	config: dict[str, dict[str, str | None]]
 	status: str
 	priority: int
 	estimated_minutes: int
@@ -479,11 +480,12 @@ def get_schedule_week(start_date: str, db_path: str | Path = DB_PATH) -> list[Sc
 				s.sequence_in_shift,
 				t.test_type,
 				t.rotation,
+				t.config_json,
 				t.status,
 				t.priority,
 				t.estimated_minutes
 			FROM schedules s
-			JOIN tests t ON t.test_id = s.test_id
+			JOIN tests t ON t.test_id = s.test_id AND t.device = s.device
 			WHERE s.schedule_version = ?
 			  AND s.scheduled_date >= ?
 			  AND s.scheduled_date < date(?, '+7 day')
@@ -503,6 +505,7 @@ def get_schedule_week(start_date: str, db_path: str | Path = DB_PATH) -> list[Sc
 				sequence_in_shift=int(row["sequence_in_shift"]),
 				test_type=row["test_type"],
 				rotation=row["rotation"],
+				config=json.loads(row["config_json"] or "{}"),
 				status=row["status"],
 				priority=int(row["priority"]),
 				estimated_minutes=int(row["estimated_minutes"]),
@@ -527,3 +530,14 @@ def mark_tests_completed(test_ids_with_device: list[tuple[str, str]], db_path: s
 			test_ids_with_device,
 		)
 
+
+def upsert_holidays(dates: list[str], db_path: str | Path = DB_PATH) -> None:
+	"""Replace all holidays with the provided list of YYYY-MM-DD date strings."""
+	with get_connection(db_path) as conn:
+		conn.execute("DELETE FROM holidays")
+		if dates:
+			conn.executemany(
+				"INSERT OR IGNORE INTO holidays(date) VALUES (?)",
+				[(d,) for d in dates],
+			)
+

backend/graph.py

@@ -49,6 +49,10 @@ def _build_graph(tests: dict[str, Any]) -> dict[str, set[str]]:
 
 
 def _compatible(a: Any, b: Any) -> bool:
+	# Hard conflict: same testpoint but different STA assignment means they cannot share a window.
+	if _has_testpoint_sta_conflict(a.config, b.config):
+		return False
+
 	# Compatible when same rotation, or both non-P3P and overlap has identical testpoints.
 	if a.rotation == b.rotation:
 		return True
@@ -57,6 +61,20 @@ def _compatible(a: Any, b: Any) -> bool:
 	return False
 
 
+def _has_testpoint_sta_conflict(
+	config_a: dict[str, dict[str, str | None]],
+	config_b: dict[str, dict[str, str | None]],
+) -> bool:
+	testpoint_to_sta_a = _build_testpoint_sta_map(config_a)
+	testpoint_to_sta_b = _build_testpoint_sta_map(config_b)
+
+	overlap = set(testpoint_to_sta_a.keys()) & set(testpoint_to_sta_b.keys())
+	for testpoint in overlap:
+		if testpoint_to_sta_a[testpoint] != testpoint_to_sta_b[testpoint]:
+			return True
+	return False
+
+
 def _same_test_points_for_overlap(
 	config_a: dict[str, dict[str, str | None]],
 	config_b: dict[str, dict[str, str | None]],
@@ -95,6 +113,30 @@ def _build_station_testpoint_map(config: dict[str, dict[str, str | None]]) -> di
 	return station_to_testpoint
 
 
+def _build_testpoint_sta_map(config: dict[str, dict[str, str | None]]) -> dict[str, set[str]]:
+	testpoint_to_sta: dict[str, set[str]] = {}
+
+	def _add_entry(entry: dict[str, str | None]) -> None:
+		testpoint = _norm(entry.get("test_point") or entry.get("Testpoint"))
+		sta_raw = _norm(entry.get("sta") or entry.get("STA"))
+		if not testpoint or not sta_raw:
+			return
+		for sta in sta_raw.split(","):
+			sta_clean = _norm(sta)
+			if sta_clean:
+				testpoint_to_sta.setdefault(testpoint, set()).add(sta_clean)
+
+	for band in ("5G", "6G", "2G"):
+		entry = config.get(band) or {}
+		_add_entry(entry)
+
+	for station_key in ("Station 1", "Station 2", "Station 3"):
+		entry = config.get(station_key) or {}
+		_add_entry(entry)
+
+	return testpoint_to_sta
+
+
 def _norm(value: Any) -> str:
 	if value is None:
 		return ""

backend/scheduler.py

@@ -136,25 +136,37 @@ def compile_schedule(
 	ref_active_priority: dict[TestKey, int] = dict(_ACTIVE_REF)
 	entries: list[ScheduleEntry] = []
 
-	current_date = _parse_date(start_date)
+	window_start_date = _parse_date(start_date)
+	current_date = window_start_date
 	last_date: str | None = None
+	daytime_shift2_window_pending = daytime_testing_today
 
 	while dut_active_priority or ref_active_priority:
+		is_special_daytime_shift2_window = (
+			daytime_shift2_window_pending
+			and current_date == window_start_date
+			and current_date.weekday() < 5
+			and current_date.isoformat() not in holiday_dates
+		)
+
 		# Get the shift sequence for current date (respects Mon/Fri/weekend rules)
-		shift_sequence = _get_shift_sequence(current_date, holiday_dates)
+		shift_sequence = _get_shift_sequence(
+			current_date,
+			holiday_dates,
+			daytime_shift2_only=is_special_daytime_shift2_window,
+		)
 		
 		if not shift_sequence:
 			break  # No valid shift sequence
 
 		dut_active_test_ids: set[TestKey] = set(dut_active_priority.keys())
 		ref_active_test_ids: set[TestKey] = set(ref_active_priority.keys())
-		bundles = _create_bundles(dut_active_test_ids, ref_active_test_ids, _TESTS_BY_ID)
+		bundles = _create_bundles(dut_active_test_ids, ref_active_test_ids, _TESTS_BY_ID, top_priority_tests)
 		shift_capacities = {
 			(date_obj, shift_idx): _shift_capacity_for_date(
 				current_date=date_obj,
 				holiday_dates=holiday_dates,
-				start_date_override=start_date,
-				daytime_testing_today=daytime_testing_today,
+				daytime_shift2_only=is_special_daytime_shift2_window and date_obj == current_date,
 			).get(shift_idx, 0)
 			for date_obj, shift_idx in shift_sequence
 		}
@@ -173,6 +185,9 @@ def compile_schedule(
 		if placed_last_date is not None:
 			last_date = placed_last_date
 
+		# Daytime testing special handling applies only to the first scheduling window.
+		daytime_shift2_window_pending = False
+
 		# Move to next scheduling window start date
 		# After shift 1 (1am-9am), there's shift 2 if daytime testing, then shift 3 (5pm)
 		# After shift 3 (5pm), shift 1 is next day (1am)
@@ -347,6 +362,7 @@ def _create_bundles(
 	dut_active_test_ids: set[TestKey],
 	ref_active_test_ids: set[TestKey],
 	tests: dict[TestKey, SchedulerTest],
+	top_priority_tests: set[str] | None = None,
 ) -> list[TestBundle]:
 	"""Create bundles of tests that must run together, sorted by priority tier and efficiency.
 	
@@ -358,6 +374,9 @@ def _create_bundles(
 	
 	Returned list is sorted by (priority_tier, total_minutes) to place small/high-priority bundles first.
 	"""
+	if top_priority_tests is None:
+		top_priority_tests = set()
+
 	bundles: list[TestBundle] = []
 	processed: set[TestKey] = set()
 
@@ -393,7 +412,12 @@ def _create_bundles(
 				if not bundle_test_ids:
 					continue
 				processed.update(bundle_test_ids)
-				priority_tier = BUNDLE_PRIORITY_P2P_WITH_COE
+				# Check if any test in bundle is top priority
+				bundle_test_ids_only = {key[0] for key in bundle_test_ids}
+				if bundle_test_ids_only & top_priority_tests:
+					priority_tier = BUNDLE_PRIORITY_FAILED  # Highest priority (0)
+				else:
+					priority_tier = BUNDLE_PRIORITY_P2P_WITH_COE
 				bundles.append(TestBundle(
 					test_ids=bundle_test_ids,
 					priority_tier=priority_tier,
@@ -415,7 +439,12 @@ def _create_bundles(
 				if not bundle_test_ids:
 					continue
 				processed.update(bundle_test_ids)
-				priority_tier = BUNDLE_PRIORITY_P2P_ONLY
+				# Check if any test in bundle is top priority
+				bundle_test_ids_only = {key[0] for key in bundle_test_ids}
+				if bundle_test_ids_only & top_priority_tests:
+					priority_tier = BUNDLE_PRIORITY_FAILED  # Highest priority (0)
+				else:
+					priority_tier = BUNDLE_PRIORITY_P2P_ONLY
 				bundles.append(TestBundle(
 					test_ids=bundle_test_ids,
 					priority_tier=priority_tier,
@@ -425,7 +454,12 @@ def _create_bundles(
 			bundle_test_ids = [key for key in (_active_key(test_id, DUT), _active_key(test_id, REF)) if key is not None and key not in processed]
 			if not bundle_test_ids:
 				continue
-			priority_tier = BUNDLE_PRIORITY_P3P
+			# Check if any test in bundle is top priority
+			bundle_test_ids_only = {key[0] for key in bundle_test_ids}
+			if bundle_test_ids_only & top_priority_tests:
+				priority_tier = BUNDLE_PRIORITY_FAILED  # Highest priority (0)
+			else:
+				priority_tier = BUNDLE_PRIORITY_P3P
 			bundles.append(TestBundle(
 				test_ids=bundle_test_ids,
 				priority_tier=priority_tier,
@@ -439,8 +473,12 @@ def _create_bundles(
 		if key in processed:
 			continue
 		test = tests[key]
-		priority_tier = BUNDLE_PRIORITY_COE_ONLY
 		bundle_test_ids = [key]
+		# Check if this test is top priority
+		if key[0] in top_priority_tests:
+			priority_tier = BUNDLE_PRIORITY_FAILED  # Highest priority (0)
+		else:
+			priority_tier = BUNDLE_PRIORITY_COE_ONLY
 		bundles.append(TestBundle(
 			test_ids=bundle_test_ids,
 			priority_tier=priority_tier,
@@ -466,8 +504,7 @@ def _rx_tx_pair_id(test_id: str, active: set[str]) -> str | None:
 def _shift_capacity_for_date(
 	current_date: date,
 	holiday_dates: set[str],
-	start_date_override: str | None,
-	daytime_testing_today: bool,
+	daytime_shift2_only: bool,
 ) -> dict[int, int]:
 	iso = current_date.isoformat()
 	if iso in holiday_dates:
@@ -477,8 +514,8 @@ def _shift_capacity_for_date(
 	if is_weekend:
 		return {1: 480, 2: 480, 3: 480}
 
-	if start_date_override and iso == _parse_date(start_date_override).isoformat() and daytime_testing_today:
-		return {1: 480, 2: 480, 3: 480}
+	if daytime_shift2_only:
+		return {1: 0, 2: 480, 3: 0}
 
 	return {1: 600, 2: 0, 3: 420}
 
@@ -489,15 +526,20 @@ def _parse_date(value: str | None) -> date:
 	return datetime.strptime(value, "%Y-%m-%d").date()
 
 
-def _get_shift_sequence(start_date: date, holiday_dates: set[str]) -> list[tuple[date, int]]:
+def _get_shift_sequence(
+	start_date: date,
+	holiday_dates: set[str],
+	daytime_shift2_only: bool = False,
+) -> list[tuple[date, int]]:
 	"""Generate the sequence of (date, shift_index) tuples for a scheduling window.
 	
 	Rules per design:
+	- Daytime testing first window (weekday only): [2] (shift 2 today only)
 	- Monday-Thursday: [3, 1] (shift 3 today, shift 1 next day) = 16 hours
 	- Friday: [3, 1, 2, 3, 1, 2, 3, 1] (Fri-Mon) = 24 hours continuous
 	- Saturday/Sunday/Holiday weekday: all 3 shifts
 	
-	Always starts on shift 3 for weekdays.
+	Weekday default starts on shift 3 unless daytime testing shift-2-only is enabled.
 	Returns ordered list of (date, shift_index) pairs.
 	"""
 	iso = start_date.isoformat()
@@ -505,6 +547,10 @@ def _get_shift_sequence(start_date: date, holiday_dates: set[str]) -> list[tuple
 	weekday = start_date.weekday()  # 0=Mon, 4=Fri, 5=Sat, 6=Sun
 	
 	shifts: list[tuple[date, int]] = []
+
+	if daytime_shift2_only and weekday < 5 and not is_holiday:
+		shifts.append((start_date, 2))
+		return shifts
 	
 	# If holiday on a weekday, treat as weekend (all 3 shifts)
 	if is_holiday and weekday < 5: