CIE-Unified

graph.py

 1from __future__ import annotations
 2
 3from dataclasses import dataclass, field
 4from typing import Dict, Iterable, List
 5
 6
 7@dataclass
 8class Graph:
 9    """Small undirected topology used by the runtime step loop."""
10
11    adjacency: Dict[str, Dict[str, float]] = field(default_factory=dict)
12
13    def ensure_node(self, node: str) -> None:
14        self.adjacency.setdefault(node, {})
15
16    def add_edge(self, source: str, target: str, weight: float = 1.0) -> None:
17        self.ensure_node(source)
18        self.ensure_node(target)
19        self.adjacency[source][target] = self.adjacency[source].get(target, 0.0) + weight
20        self.adjacency[target][source] = self.adjacency[target].get(source, 0.0) + weight
21
22    def connect_path(self, nodes: Iterable[str], weight: float = 1.0) -> None:
23        ordered = [node for node in nodes if node]
24        for left, right in zip(ordered, ordered[1:]):
25            self.add_edge(left, right, weight=weight)
26
27    def neighbors(self, node: str) -> Dict[str, float]:
28        return self.adjacency.get(node, {})
29
30    def edge_weight(self, source: str, target: str) -> float:
31        return self.adjacency.get(source, {}).get(target, 0.0)
32
33    def weighted_degree(self, node: str) -> float:
34        return sum(self.neighbors(node).values())
35
36    def nodes(self) -> List[str]:
37        return sorted(self.adjacency)