diff --git a/backend/db.py b/backend/db.py index 2c08919..302fcb4 100644 --- a/backend/db.py +++ b/backend/db.py @@ -6,6 +6,7 @@ from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Iterator +from graph import build_station_testpoint_map APP_ROOT = Path(__file__).resolve().parent @@ -36,6 +37,7 @@ class TestRecord: status: str = "pending" excluded: bool = False raw_payload: dict[str, Any] | None = None + station_testpoint_map: str | None = None @dataclass(frozen=True) @@ -141,6 +143,7 @@ def init_db(db_path: str | Path = DB_PATH) -> None: _ensure_column(conn, "tests", "victim_band", "TEXT") _ensure_column(conn, "tests", "excluded", "INTEGER NOT NULL DEFAULT 0") _ensure_column(conn, "tests", "throttled", "INTEGER NOT NULL DEFAULT 0") + _ensure_column(conn, "tests", "station_testpoint_map", "TEXT") # Seed runtime defaults for schedule estimation. conn.execute( @@ -154,6 +157,12 @@ def init_db(db_path: str | Path = DB_PATH) -> None: ) +def _serialize_station_testpoint_map(config: dict[str, dict[str, str | None]]) -> str: + """Compute and serialize the station-to-testpoint map from a test config.""" + station_map = build_station_testpoint_map(config) + return json.dumps(station_map) + + def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> int: if not records: return 0 @@ -163,9 +172,9 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in """ INSERT INTO tests( test_id, device, test_type, rotation, rx_tx, has_coe_pair, - coe_pairing_json, config_json, priority, victim_band, throttled, estimated_minutes, status, excluded, raw_payload + coe_pairing_json, config_json, priority, victim_band, throttled, estimated_minutes, status, excluded, raw_payload, station_testpoint_map ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(test_id, device) DO UPDATE SET test_type = excluded.test_type, device = excluded.device, @@ -181,6 +190,7 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in status = excluded.status, excluded = excluded.excluded, raw_payload = excluded.raw_payload, + station_testpoint_map = excluded.station_testpoint_map, updated_at = CURRENT_TIMESTAMP """, [ @@ -200,6 +210,7 @@ def upsert_tests(records: list[TestRecord], db_path: str | Path = DB_PATH) -> in r.status, int(r.excluded), json.dumps(r.raw_payload or {}), + _serialize_station_testpoint_map(r.config), ) for r in records ], @@ -403,7 +414,7 @@ def list_tests_for_device(device: str, db_path: str | Path = DB_PATH) -> list[Te test_id, device, test_type, rotation, rx_tx, has_coe_pair, config_json, victim_band, coe_pairing_json, priority, throttled, estimated_minutes, - excluded, status, raw_payload + excluded, status, raw_payload, station_testpoint_map FROM tests WHERE device = ? """, @@ -429,6 +440,7 @@ def list_tests_for_device(device: str, db_path: str | Path = DB_PATH) -> list[Te status=row["status"], excluded=bool(row["excluded"]), raw_payload=json.loads(row["raw_payload"] or "{}"), + station_testpoint_map=row["station_testpoint_map"], ) ) diff --git a/backend/graph.py b/backend/graph.py index 615f5a2..d1a5cbf 100644 --- a/backend/graph.py +++ b/backend/graph.py @@ -1,6 +1,8 @@ from __future__ import annotations +import json from typing import Any +from pathlib import Path # Immutable-in-practice graph state for the loaded DUT test set. _TESTS_BY_ID: dict[str, Any] = {} @@ -13,6 +15,19 @@ def reset_graph_state() -> None: _GRAPH = {} +def _get_station_map(test: Any) -> dict[str, str]: + """Get station testpoint map from test record or compute it from config.""" + # Try to get from serialized map first (cached from DB) + if hasattr(test, 'station_testpoint_map') and test.station_testpoint_map: + try: + return json.loads(test.station_testpoint_map) + except (json.JSONDecodeError, TypeError): + pass + + # Safeguard: return empty map + return {} + + def is_graph_built() -> bool: return bool(_GRAPH) @@ -45,56 +60,49 @@ def _build_graph(tests: dict[str, Any]) -> dict[str, set[str]]: if _compatible(a, b): graph[a_id].add(b_id) graph[b_id].add(a_id) + # Print out graph for debugging + out = Path('output1.txt') + with out.open('w', encoding='utf-8') as f: + f.write(json.dumps({k: list(v) for k, v in graph.items()}, indent=2)) return graph 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): + station_map_a = _get_station_map(a) + station_map_b = _get_station_map(b) + if _same_station_to_testpoint(station_map_a, station_map_b, a, b) and _same_testpoint_to_station(station_map_a, station_map_b, a, b): + return True return False - # Compatible when same rotation, or both non-P3P and overlap has identical testpoints. - if a.rotation == b.rotation: - return True - if a.test_type != "P3P" and b.test_type != "P3P" and _same_test_points_for_overlap(a.config, b.config): - return True - return False - - -def _has_testpoint_sta_conflict( - config_a: dict[str, dict[str, str | None]], - config_b: dict[str, dict[str, str | None]], +def _same_station_to_testpoint( + station_map_a: dict[str, str], + station_map_b: dict[str, str], test_a: Any, test_b: Any ) -> 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]], -) -> bool: - station_map_a = _build_station_testpoint_map(config_a) - station_map_b = _build_station_testpoint_map(config_b) - - overlap = set(station_map_a.keys()) & set(station_map_b.keys()) - for station in overlap: + overlap_stations = set(station_map_a.keys()) & set(station_map_b.keys()) + for station in overlap_stations: if station_map_a[station] != station_map_b[station]: return False return True +def _same_testpoint_to_station( + station_map_a: dict[str, str], + station_map_b: dict[str, str], test_a: Any, test_b: Any +) -> bool: + overlap_testpoints = set(station_map_a.values()) & set(station_map_b.values()) + for testpoint in overlap_testpoints: + stations_a = {s for s, t in station_map_a.items() if t == testpoint} + stations_b = {s for s, t in station_map_b.items() if t == testpoint} + if stations_a != stations_b: + print(f"Tests {test_a.test_id} and {test_b.test_id} have conflicting stations for testpoint. {station_map_a} vs {station_map_b}") + return False + return True -def _build_station_testpoint_map(config: dict[str, dict[str, str | None]]) -> dict[str, str]: +def build_station_testpoint_map(config: dict[str, dict[str, str | None]]) -> dict[str, str]: station_to_testpoint: dict[str, 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")) + testpoint = _norm(entry.get("test_point")) + sta_raw = _norm(entry.get("sta")) if not testpoint or not sta_raw: return for sta in sta_raw.split(","): @@ -113,29 +121,6 @@ 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: diff --git a/backend/scheduler.py b/backend/scheduler.py index 7ae9bff..8948612 100644 --- a/backend/scheduler.py +++ b/backend/scheduler.py @@ -314,7 +314,7 @@ def _fit_bundles_to_shifts( state = shift_state[date_shift] state["remaining_minutes"] += prev_remaining # Add back any leftover from previous shift if state['remaining_minutes'] >= test.estimated_minutes: - print(f"remaining minutes for {date_shift}: {state['remaining_minutes']} - placing {key} ({test.estimated_minutes}m)") + #print(f"remaining minutes for {date_shift}: {state['remaining_minutes']} - placing {key} ({test.estimated_minutes}m)") entries.append( ScheduleEntry( test_id=test.test_id, diff --git a/frontend/src/components/RightPanel.jsx b/frontend/src/components/RightPanel.jsx index a3942fc..a3392f0 100644 --- a/frontend/src/components/RightPanel.jsx +++ b/frontend/src/components/RightPanel.jsx @@ -64,7 +64,7 @@ export default function RightPanel({ {/* Top priority */}