from __future__ import annotations
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
import networkx as nx
from semi_cr.core.lab.station.routing_v1.errors import (
RoutingValidationError,
)
from semi_cr.core.lab.station.routing_v1.graph import (
is_routable_edge,
)
MODALITY_NEUTRAL_EDGE_KINDS = {
"device",
"represents",
}
[docs]
class ValidationSeverity(StrEnum):
ERROR = "error"
WARNING = "warning"
[docs]
@dataclass(frozen=True)
class ValidationIssue:
code: str
message: str
severity: ValidationSeverity
node: object | None = None
edge: tuple[object, object, object] | None = None
def _validate_node_capabilities(
graph: nx.MultiDiGraph,
) -> Iterable[ValidationIssue]:
for node_id, data in graph.nodes(
data=True,
):
capabilities = data.get("capabilities")
if capabilities is None:
continue
if isinstance(capabilities, str):
yield ValidationIssue(
code="capabilities_not_collection",
message=(
f"Node {node_id!r} stores capabilities "
"as a string. Use a collection such as "
"{'voltage_source'}."
),
severity=ValidationSeverity.ERROR,
node=node_id,
)
continue
try:
capabilities = tuple(capabilities)
except TypeError:
yield ValidationIssue(
code="capabilities_not_iterable",
message=(
f"Node {node_id!r} has non-iterable "
f"capabilities {capabilities!r}."
),
severity=ValidationSeverity.ERROR,
node=node_id,
)
continue
if not capabilities:
yield ValidationIssue(
code="empty_capabilities",
message=(
f"Node {node_id!r} has an empty "
"capabilities collection."
),
severity=ValidationSeverity.WARNING,
node=node_id,
)
for capability in capabilities:
if not isinstance(capability, str):
yield ValidationIssue(
code="invalid_capability",
message=(
f"Node {node_id!r} has a "
"non-string capability "
f"{capability!r}."
),
severity=ValidationSeverity.ERROR,
node=node_id,
)
def _validate_edge_modalities(
graph: nx.MultiDiGraph,
) -> Iterable[ValidationIssue]:
for source, target, key, data in graph.edges(
keys=True,
data=True,
):
if not is_routable_edge(data):
continue
kind = data.get("kind")
if kind in MODALITY_NEUTRAL_EDGE_KINDS:
continue
modalities = data.get(
"conductive_modalities"
)
edge = (source, target, key)
if modalities is None:
yield ValidationIssue(
code="missing_conductive_modalities",
message=(
f"Routable edge {edge!r} with kind "
f"{kind!r} has no "
"'conductive_modalities'."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
continue
if isinstance(modalities, str):
yield ValidationIssue(
code="modalities_not_collection",
message=(
f"Edge {edge!r} stores modalities as "
f"a string {modalities!r}. Use a "
"collection such as {'dc'}."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
continue
try:
modalities = tuple(modalities)
except TypeError:
yield ValidationIssue(
code="modalities_not_iterable",
message=(
f"Edge {edge!r} has non-iterable "
f"modalities {modalities!r}."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
continue
if not modalities:
yield ValidationIssue(
code="empty_conductive_modalities",
message=(
f"Routable edge {edge!r} has no "
"conductive modalities."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
for modality in modalities:
if not isinstance(modality, str):
yield ValidationIssue(
code="invalid_modality",
message=(
f"Edge {edge!r} has a non-string "
f"modality {modality!r}."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
def _validate_configurable_edges(
graph: nx.MultiDiGraph,
) -> Iterable[ValidationIssue]:
for source, target, key, data in graph.edges(
keys=True,
data=True,
):
if not data.get("configurable", False):
continue
edge = (source, target, key)
active = data.get("active")
if not isinstance(active, bool):
yield ValidationIssue(
code="missing_configurable_state",
message=(
f"Configurable edge {edge!r} must "
"define a Boolean 'active' state."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
actions = data.get("actions")
if actions is None:
yield ValidationIssue(
code="missing_route_actions",
message=(
f"Configurable edge {edge!r} has no "
"route actions."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
continue
if isinstance(actions, (str, bytes)):
yield ValidationIssue(
code="invalid_route_actions",
message=(
f"Configurable edge {edge!r} stores "
"'actions' as a string instead of a "
"collection."
),
severity=ValidationSeverity.ERROR,
edge=edge,
)
def _validate_represented_backends(
graph: nx.MultiDiGraph,
) -> Iterable[ValidationIssue]:
represented_targets: set[object] = set()
for source, target, data in graph.edges(
data=True,
):
if data.get("kind") != "represents":
continue
represented_targets.add(target)
for node_id in represented_targets:
data = graph.nodes[node_id]
if data.get("obj") is None:
yield ValidationIssue(
code="missing_runtime_object",
message=(
f"Represented backend node "
f"{node_id!r} has no runtime 'obj'."
),
severity=ValidationSeverity.ERROR,
node=node_id,
)
def _validate_duplicate_pin_objects(
graph: nx.MultiDiGraph,
) -> Iterable[ValidationIssue]:
pin_nodes_by_object_id: dict[
int,
list[object],
] = defaultdict(list)
for node_id, data in graph.nodes(
data=True,
):
if data.get("kind") != "device_pin":
continue
pin = data.get("obj")
if pin is None:
yield ValidationIssue(
code="missing_pin_object",
message=(
f"Device-pin node {node_id!r} has "
"no pin object."
),
severity=ValidationSeverity.ERROR,
node=node_id,
)
continue
pin_nodes_by_object_id[id(pin)].append(
node_id
)
for node_ids in pin_nodes_by_object_id.values():
if len(node_ids) <= 1:
continue
yield ValidationIssue(
code="duplicate_pin_object",
message=(
"The same pin object is attached to "
f"multiple graph nodes: {node_ids!r}."
),
severity=ValidationSeverity.ERROR,
node=node_ids[0],
)
[docs]
@dataclass(frozen=True)
class RoutingValidationReport:
issues: tuple[ValidationIssue, ...]
@property
def errors(self) -> tuple[ValidationIssue, ...]:
return tuple(
issue
for issue in self.issues
if issue.severity
== ValidationSeverity.ERROR
)
@property
def warnings(self) -> tuple[ValidationIssue, ...]:
return tuple(
issue
for issue in self.issues
if issue.severity
== ValidationSeverity.WARNING
)
@property
def is_valid(self) -> bool:
return not self.errors
[docs]
def raise_for_errors(self) -> None:
if self.is_valid:
return
details = "\n".join(
f"- [{issue.code}] {issue.message}"
for issue in self.errors
)
raise RoutingValidationError(
"The station graph does not satisfy "
f"the routing contract:\n{details}"
)
[docs]
def validate_routing_graph(
graph: nx.MultiDiGraph,
) -> RoutingValidationReport:
"""
Validate the graph structure required by routing.
Validation checks graph metadata and internal consistency. It does
not require every device pin to have a route because unrouted pins
may be legitimate.
"""
if not isinstance(
graph,
nx.MultiDiGraph,
):
raise TypeError(
"Routing validation expects an "
"nx.MultiDiGraph; received "
f"{type(graph).__name__}."
)
issues: list[ValidationIssue] = []
issues.extend(
_validate_node_capabilities(graph)
)
issues.extend(
_validate_edge_modalities(graph)
)
issues.extend(
_validate_configurable_edges(graph)
)
issues.extend(
_validate_represented_backends(graph)
)
issues.extend(
_validate_duplicate_pin_objects(graph)
)
return RoutingValidationReport(
issues=tuple(issues),
)