File size: 1,861 Bytes
a521353 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | """rustworkx-backed Living Mycelial Hypergraph."""
import rustworkx as rx
from typing import Dict
from ..schema import HypergraphDelta
class Hypergraph:
def __init__(self):
self.g = rx.PyDiGraph(multigraph=True)
self._idx: Dict[str, int] = {}
def apply_delta(self, delta: HypergraphDelta):
for n in delta.nodes_added:
if n.node_id in self._idx:
continue
i = self.g.add_node({
"node_id": n.node_id, "node_type": n.node_type, "label": n.label,
"payload": n.payload, "embedding_hint": n.embedding_hint,
})
self._idx[n.node_id] = i
for u in delta.nodes_updated:
if u.node_id in self._idx:
d = self.g[self._idx[u.node_id]]
d.update(u.fields_changed)
for e in delta.edges_added:
s = self._idx.get(e.source_node_id)
t = self._idx.get(e.target_node_id)
if s is None or t is None:
continue
self.g.add_edge(s, t, {
"edge_id": e.edge_id, "edge_type": e.edge_type,
"weight": e.weight, "payload": e.payload,
})
def conflict_centrality(self) -> Dict[str, float]:
try:
bc = rx.betweenness_centrality(self.g)
return {self.g[i]["node_id"]: v for i, v in enumerate(bc)}
except Exception:
return {}
def context_summary(self, max_nodes: int = 12) -> str:
items = []
for i in list(self.g.node_indexes())[-max_nodes:]:
n = self.g[i]
items.append(f"{n['label']}({n['node_type']})")
return "; ".join(items)
def node_count(self) -> int: return self.g.num_nodes()
def edge_count(self) -> int: return self.g.num_edges()
|