Files
scheduler/backend/graph.py
T
2026-06-16 16:09:54 -04:00

102 lines
2.7 KiB
Python

from __future__ import annotations
from typing import Any
# Immutable-in-practice graph state for the loaded DUT test set.
_TESTS_BY_ID: dict[str, Any] = {}
_GRAPH: dict[str, set[str]] = {}
def reset_graph_state() -> None:
global _TESTS_BY_ID, _GRAPH
_TESTS_BY_ID = {}
_GRAPH = {}
def is_graph_built() -> bool:
return bool(_GRAPH)
def build_graph_once(tests: list[Any]) -> dict[str, set[str]]:
"""Build DUT-only compatibility adjacency graph once per load lifecycle."""
global _TESTS_BY_ID, _GRAPH
if _GRAPH:
return _GRAPH
# Keep only one representative per DUT test_id.
_TESTS_BY_ID = {str(t.test_id): t for t in tests if getattr(t, "test_id", None)}
_GRAPH = _build_graph(_TESTS_BY_ID)
return _GRAPH
def get_graph() -> dict[str, set[str]]:
return _GRAPH
def _build_graph(tests: dict[str, Any]) -> dict[str, set[str]]:
test_ids = list(tests.keys())
graph = {test_id: set() for test_id in test_ids}
for i in range(len(test_ids)):
for j in range(i + 1, len(test_ids)):
a_id = test_ids[i]
b_id = test_ids[j]
a = tests[a_id]
b = tests[b_id]
if _compatible(a, b):
graph[a_id].add(b_id)
graph[b_id].add(a_id)
return graph
def _compatible(a: Any, b: Any) -> bool:
# 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 _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:
if station_map_a[station] != station_map_b[station]:
return False
return True
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"))
if not testpoint or not sta_raw:
return
for sta in sta_raw.split(","):
sta_clean = _norm(sta)
if sta_clean:
station_to_testpoint[sta_clean] = testpoint
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 station_to_testpoint
def _norm(value: Any) -> str:
if value is None:
return ""
return " ".join(str(value).strip().upper().split())