scheduler/backend/graph.py

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] = {}
_GRAPH: dict[str, set[str]] = {}


def reset_graph_state() -> None:
	global _TESTS_BY_ID, _GRAPH
	_TESTS_BY_ID = {}
	_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)


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)
	# 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:
		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

def _same_station_to_testpoint(
	station_map_a: dict[str, str],
	station_map_b: dict[str, str], test_a: Any, test_b: Any
) -> bool:
	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]:
	station_to_testpoint: dict[str, str] = {}

	def _add_entry(entry: dict[str, str | None]) -> None:
		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(","):
			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())