Source code for semi_cr.core.lab.station.routing_v1.models

# Contains graph-independent descriptions of routing requests and results:

from dataclasses import dataclass
from typing import Any
from collections.abc import Sequence

from abc import ABC, abstractmethod
import networkx as nx

from semi_cr.core.lab.station.graphing.runtime_builder import (
    build_connection_graph_for_instrument,
)

from semi_cr.core.lab.pins.base import BasePinChannel

# Static Routing



@dataclass(frozen=True)
class RouteRequest:
    pin: BasePinChannel
    role: str
    quantity: str | None = None
    modality: str | None = None




class Routable(ABC):
    @property
    @abstractmethod
    def graph(self) -> nx.MultiDiGraph:
        ...




class ConnectionRoutable(Routable):

    def _init_connections(
        self,
        connections: Sequence[dict[str, Any]] | None = None,
    ) -> None:
        self._connections = tuple(connections or ())

    @property
    def connections(self) -> tuple[dict[str, Any], ...]:
        return self._connections

    @property
    def graph(self) -> nx.MultiDiGraph:
        parent = getattr(self, "parent", None)

        parent_name = None
        if parent is not None:
            parent_name = getattr(parent, "short_name", parent.name)

        return build_connection_graph_for_instrument(
            instrument=self,
            connections=self.connections,
            parent_name=parent_name,
        )


# Configurable Routing



@dataclass(frozen=True)
class RouteAction:
    instrument: object
    parameter: str
    value: object




@dataclass(frozen=True)
class RouteEdge:
    source: str
    target: str
    key: object
    attributes: dict[str, Any]





@dataclass(frozen=True)
class RoutePlan:
    device_pin: BasePinChannel
    terminal: object
    path: tuple[object, ...]
    actions: tuple[RouteAction, ...]




@dataclass(frozen=True)
class Route:
    request: RouteRequest
    target_node: str
    path: tuple[str, ...]

    # edges: tuple[RouteEdge, ...]
    # actions: tuple[RouteAction, ...] = ()