import networkx as nx
from typing import Any
from semi_cr.core.lab.station.graphing.models import GraphRelation
ELECTRICAL_EDGE_KINDS = {
GraphRelation.DEPENDENCY,
GraphRelation.REPRESENTS,
GraphRelation.CONNECTION,
GraphRelation.INTERNAL_CONNECTION,
GraphRelation.PHYSICAL,
GraphRelation.DEVICE,
}
MODALITY_NEUTRAL_EDGE_KINDS = {
GraphRelation.REPRESENTS,
GraphRelation.DEVICE,
}
[docs]
def edge_supports_modality(
data: dict[str, Any],
modality: str | None,
) -> bool:
if modality is None:
return True
kind = data.get("kind")
if kind in MODALITY_NEUTRAL_EDGE_KINDS:
return True
modalities = data.get("conductive_modalities")
# if modalities is None:
# return False
# No declaration means that this edge does not
# constrain the modality.
if modalities is None:
return True
return modality.lower() in {
str(item).lower()
for item in modalities
}
[docs]
def is_routable_edge(attrs: dict[str, Any]) -> bool:
# relation = attrs.get("relation")
kind = attrs.get("kind")
# Structural containment must never create an
# electrical path between sibling terminals.
# if relation == "contains" or kind == "contains":
# return False
return kind in ELECTRICAL_EDGE_KINDS
# def build_routing_graph(
# graph: nx.MultiDiGraph,
# ) -> nx.Graph:
# routing = nx.Graph()
# routing.add_nodes_from(
# graph.nodes(data=True)
# )
# for u, v, key, attrs in graph.edges(
# keys=True,
# data=True,
# ):
# if is_routable_edge(attrs):
# routing.add_edge(
# u,
# v,
# **attrs,
# )
# return routing
# def build_electrical_routing_graph(
# graph: nx.MultiDiGraph,
# ) -> nx.Graph:
# routing_graph = nx.Graph()
# routing_graph.add_nodes_from(
# graph.nodes(data=True)
# )
# for source, target, data in graph.edges(data=True):
# edge_kind = data.get("kind")
# if edge_kind not in ELECTRICAL_EDGE_KINDS:
# continue
# routing_graph.add_node(
# source,
# **graph.nodes[source],
# )
# routing_graph.add_node(
# target,
# **graph.nodes[target],
# )
# routing_graph.add_edge(
# source,
# target,
# **data,
# )
# return routing_graph
[docs]
def edge_is_available(
attrs: dict[str, Any],
include_configurable: bool,
) -> bool:
"""
Return whether an edge can participate in route discovery.
Ordinary static edges are always available.
Configurable edges are available when:
- they are currently active; or
- configurable alternatives were explicitly requested.
"""
configurable = bool(
attrs.get("configurable", False)
)
if not configurable:
return True
if attrs.get("state") == "active":
return True
return include_configurable
[docs]
def build_routing_graph(
graph: nx.MultiDiGraph,
modality: str | None = None,
include_configurable: bool = False,
) -> nx.Graph:
"""
Build an undirected routable view of the station graph.
Parameters
----------
graph:
Complete station graph.
modality:
Required signal modality, such as ``"dc"``, ``"ac"``, or
``"rf"``. If None, no modality filtering is performed.
include_configurable:
If False, include only currently active connections.
If True, also include inactive connections that can be activated
through a configurable routing component such as the CQCC switch.
Returns
-------
nx.Graph
Undirected routing view containing only usable or configurable
routing connections.
"""
routing_graph = nx.Graph()
routing_graph.add_nodes_from(
graph.nodes(data=True),
)
for source, target, _key, attrs in graph.edges(
keys=True,
data=True,
):
if not is_routable_edge(attrs):
continue
if not edge_supports_modality(
attrs,
modality,
):
continue
if not edge_is_available(
attrs,
include_configurable=include_configurable,
):
continue
routing_graph.add_edge(
source,
target,
**attrs,
)
return routing_graph