Source code for semi_cr.core.lab.station.graphing.validation

from __future__ import annotations

from collections.abc import Iterable

import networkx as nx




class GraphValidationError(ValueError):
    pass





def validate_physical_edge_metadata(
    graph: nx.MultiDiGraph,
) -> None:
    required = {
        "source_instrument",
        "source_channel",
        "target_instrument",
        "target_channel",
    }

    for source, target, key, attrs in graph.edges(
        keys=True,
        data=True,
    ):
        if attrs.get("kind") != "physical":
            continue

        missing = required - attrs.keys()
        if missing:
            raise GraphValidationError(
                f"Physical edge {(source, target, key)!r} "
                f"is missing attributes {sorted(missing)!r}."
            )





def validate_device_pin_wiring(
    graph: nx.MultiDiGraph,
) -> None:
    for node, attrs in graph.nodes(data=True):
        if attrs.get("kind") != "device_pin":
            continue

        if graph.in_degree(node) == 0:
            raise GraphValidationError(
                f"Device pin {node!r} has no incoming wiring."
            )





def validate_unknown_nodes(
    graph: nx.MultiDiGraph,
    allowed: Iterable[str] = (),
) -> None:
    allowed_nodes = set(allowed)

    unknown = [
        node
        for node, attrs in graph.nodes(data=True)
        if attrs.get("kind") == "unknown"
        and node not in allowed_nodes
    ]

    if unknown:
        raise GraphValidationError(
            f"Graph contains unresolved nodes: {unknown!r}."
        )





def validate_graph(
    graph: nx.MultiDiGraph,
) -> None:
    validate_physical_edge_metadata(graph)
    validate_device_pin_wiring(graph)
    validate_unknown_nodes(graph)