Spaces:
Running
Running
| """In-memory Decidron network state with version history. | |
| Holds the live topology (nodes + directed couplings), applies topology | |
| modifications, and snapshots the network whenever it changes so the UI | |
| can show before/after diagrams. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| from .models import DecidronNode, Edge, NetworkOp, NetworkSnapshot | |
| _DATA_DIR = Path(__file__).resolve().parent.parent / "data" / "decidron" | |
| _DEFAULT_NETWORK = _DATA_DIR / "default_network.json" | |
| class NetworkStore: | |
| def __init__(self) -> None: | |
| self.nodes: dict[str, DecidronNode] = {} | |
| self.edges: list[Edge] = [] | |
| self.version: int = 0 | |
| self.history: list[NetworkSnapshot] = [] | |
| self.load_default() | |
| def load_default(self) -> None: | |
| raw = json.loads(_DEFAULT_NETWORK.read_text(encoding="utf-8")) | |
| self.nodes = {n["id"]: DecidronNode(**n) for n in raw.get("nodes", [])} | |
| self.edges = [Edge(**e) for e in raw.get("edges", [])] | |
| self.version = 0 | |
| self.history = [] | |
| self._snapshot("initial network") | |
| def reset(self) -> None: | |
| self.load_default() | |
| # -- queries --------------------------------------------------------- | |
| def neighbors_out(self, node_id: str) -> list[str]: | |
| """Nodes reachable from ``node_id`` via an outgoing coupling.""" | |
| return [e.target for e in self.edges if e.source == node_id] | |
| def nodes_list(self) -> list[DecidronNode]: | |
| return list(self.nodes.values()) | |
| def find_sensor_node(self, channel: str) -> DecidronNode | None: | |
| """Match a sensor channel to a sensor node by id or label.""" | |
| for node in self.nodes.values(): | |
| if node.type != "sensor": | |
| continue | |
| if node.id == channel or node.label.lower() == channel.lower(): | |
| return node | |
| return None | |
| # -- mutations ------------------------------------------------------- | |
| def _snapshot(self, reason: str) -> None: | |
| self.history.append( | |
| NetworkSnapshot( | |
| version=self.version, | |
| reason=reason, | |
| nodes=[n.model_copy(deep=True) for n in self.nodes.values()], | |
| edges=[e.model_copy(deep=True) for e in self.edges], | |
| ) | |
| ) | |
| def _edge_exists(self, source: str, target: str) -> bool: | |
| return any(e.source == source and e.target == target for e in self.edges) | |
| def apply_op(self, op: NetworkOp) -> bool: | |
| """Apply a single topology op. Returns True if the graph changed.""" | |
| if op.op == "add_node" and op.node is not None: | |
| if op.node.id not in self.nodes: | |
| self.nodes[op.node.id] = op.node | |
| return True | |
| return False | |
| if op.op == "add_edge" and op.source and op.target: | |
| if not self._edge_exists(op.source, op.target): | |
| self.edges.append(Edge(source=op.source, target=op.target)) | |
| return True | |
| return False | |
| if op.op == "remove_edge" and op.source and op.target: | |
| before = len(self.edges) | |
| self.edges = [ | |
| e for e in self.edges | |
| if not (e.source == op.source and e.target == op.target) | |
| ] | |
| return len(self.edges) != before | |
| return False | |
| def apply_ops(self, ops: list[NetworkOp], reason: str) -> int: | |
| """Apply ops, snapshotting once if anything changed. Returns count.""" | |
| changed = 0 | |
| for op in ops: | |
| if self.apply_op(op): | |
| changed += 1 | |
| if changed: | |
| self.version += 1 | |
| self._snapshot(reason) | |
| return changed | |