p2p/coe backend implemented

backend/graph.py

@@ -0,0 +1,92 @@
+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] = {}
+	for band in ("5G", "6G", "2G"):
+		entry = config.get(band) or {}
+		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:
+			continue
+		for sta in sta_raw.split(","):
+			sta_clean = _norm(sta)
+			if sta_clean:
+				station_to_testpoint[sta_clean] = testpoint
+	return station_to_testpoint
+
+
+def _norm(value: Any) -> str:
+	if value is None:
+		return ""
+	return " ".join(str(value).strip().upper().split())