from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any
import networkx as nx
from semi_cr.core.lab.station.graphing.resolver import GraphNodeResolver
EDGE_METADATA_KEYS = {
"dependency_key",
}
CANONICAL_IDENTITY_ATTRIBUTES = {
"kind",
"name",
"instrument_name",
"terminal_name",
}
def _runtime_node_attrs(
node_id: Any,
node_attrs: dict[str, Any],
) -> dict[str, Any]:
attrs = dict(node_attrs)
# Some dependency graphs store the runtime object
# under "value", others under "obj".
value = attrs.pop(
"value",
None,
)
if value is not None:
attrs["obj"] = value
obj = attrs.get("obj")
attrs.setdefault(
"name",
(
getattr(
obj,
"name",
str(node_id),
)
if obj is not None
else str(node_id)
),
)
return attrs
def _upsert_node(
graph: nx.MultiDiGraph,
canonical_id: str,
node_id: Any,
node_attrs: dict[str, Any],
*,
default_kind: str,
) -> None:
incoming = _runtime_node_attrs(
node_id,
node_attrs,
)
if canonical_id not in graph:
incoming.setdefault(
"kind",
default_kind,
)
graph.add_node(
canonical_id,
**incoming,
)
return
existing = graph.nodes[canonical_id]
for key, value in incoming.items():
# Preserve the identity assigned by the
# canonical/base graph.
if key in CANONICAL_NODE_IDENTITY_KEYS:
continue
# Do not erase useful canonical data with None.
if value is None:
continue
existing[key] = value
def _edge_identity_attrs(
attrs: dict[str, Any],
) -> dict[str, Any]:
return {key: value for key, value in attrs.items() if key not in EDGE_METADATA_KEYS}
def _upsert_edge(
graph: nx.MultiDiGraph,
source: str,
target: str,
attrs: dict[str, Any],
) -> None:
requested_identity = _edge_identity_attrs(attrs)
existing_edges = graph.get_edge_data(
source,
target,
default={},
)
for key, existing_attrs in existing_edges.items():
existing_identity = _edge_identity_attrs(existing_attrs)
if existing_identity != requested_identity:
continue
# Same logical edge: enrich the existing edge.
existing_attrs.update(attrs)
return
# Genuinely different parallel relation.
graph.add_edge(
source,
target,
**attrs,
)
[docs]
@dataclass(frozen=True)
class DependencyOverlayBuilder:
"""
Overlays dependency edges from Graph graphs onto an existing MultiDiGraph
(typically a containment graph).
"""
resolver: GraphNodeResolver
edge_kind: str = "depends"
# policy knobs
include_edges_with_none_state: bool = True
skip_disabled_edges: bool = True
disabled_state_names: tuple[str, ...] = ("Edge.Disabled", "Disabled") # robust across enum/string
default_node_kind: str = "logical"
@staticmethod
def _state_name(state: Any) -> str | None:
"""
Return a comparable name for string or enum-like states.
"""
if state is None:
return None
return str(state)
def _should_include_edge(
self,
state: Any,
) -> bool:
if state is None and not self.include_edges_with_none_state:
return False
if self.skip_disabled_edges and self._state_name(state) in self.disabled_state_names:
return False
return True
[docs]
def overlay(
self,
base: nx.MultiDiGraph,
dep_graphs: Iterable[Any],
) -> nx.MultiDiGraph:
graph = base.copy()
for dependency_graph in dep_graphs:
# ---------------------------------------------
# Merge resolved nodes
# ---------------------------------------------
for node_id, node_attrs in dependency_graph.nodes(
data=True,
):
canonical_id = self.resolver(
dependency_graph,
node_id,
)
incoming_attrs = dict(node_attrs)
value = incoming_attrs.pop(
"value",
None,
)
if value is not None:
incoming_attrs["obj"] = value
obj = incoming_attrs.get("obj")
incoming_attrs.setdefault(
"name",
(
getattr(
obj,
"name",
str(node_id),
)
if obj is not None
else str(node_id)
),
)
if canonical_id not in graph:
incoming_attrs.setdefault(
"kind",
self.default_node_kind,
)
graph.add_node(
canonical_id,
**incoming_attrs,
)
continue
canonical_attrs = graph.nodes[canonical_id]
for key, value in incoming_attrs.items():
if key in CANONICAL_IDENTITY_ATTRIBUTES and key in canonical_attrs:
continue
if value is None:
continue
canonical_attrs[key] = value
# ---------------------------------------------
# Merge enabled dependency edges
# ---------------------------------------------
for (
source,
target,
original_key,
edge_attrs,
) in dependency_graph.edges(
keys=True,
data=True,
):
state = edge_attrs.get("state")
if not self._should_include_edge(state):
continue
canonical_source = self.resolver(
dependency_graph,
source,
)
canonical_target = self.resolver(
dependency_graph,
target,
)
attrs = dict(edge_attrs)
attrs.setdefault(
"kind",
self.edge_kind,
)
attrs["dependency_key"] = original_key
_upsert_edge(
graph,
canonical_source,
canonical_target,
attrs,
)
return graph