demo/rel2graph_networkx_demo.py
#!/usr/bin/env python3
"""
rel2graph → networkx demo (N3).
Loads ONE knut rel2graph result — the 12-table ERP graph produced by the
``rel2graph-export`` bin — into a ``networkx.MultiDiGraph`` and runs a handful of
graph analytics over it. It is a **correctness demo, not a pretty print**: every
count it computes from the networkx graph is ASSERTED equal to knut's own oracle
(``oracle.json``, computed independently in Rust straight from the mapping
output). If the loader dropped, duplicated, or mis-typed a single element, an
assertion fails.
Run:
# 1. produce the export (Rust side, deterministic):
cargo run -p knut-rel2graph-bench --release --bin rel2graph-export -- 1 target/rel2graph-export
# 2. run this demo against it:
python demo/rel2graph_networkx_demo.py [EXPORT_DIR]
``EXPORT_DIR`` defaults to ``target/rel2graph-export``.
"""
from __future__ import annotations
import os
import sys
import networkx as nx
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import knut_networkx as kn # noqa: E402
def main() -> int:
export_dir = sys.argv[1] if len(sys.argv) > 1 else "target/rel2graph-export"
if not os.path.isdir(export_dir):
print(f"error: export dir {export_dir!r} not found — run rel2graph-export first.")
print(" cargo run -p knut-rel2graph-bench --release --bin rel2graph-export -- 1 " + export_dir)
return 2
nodes, edges, oracle = kn.load_json_export(export_dir)
g = kn.build_multidigraph(nodes, edges)
print(f"== rel2graph → networkx.MultiDiGraph ({export_dir}) ==")
print(f"directed : {g.is_directed()} multigraph : {g.is_multigraph()}")
print(f"nodes : {g.number_of_nodes():>6} (oracle {oracle['total_nodes']})")
print(f"edges : {g.number_of_edges():>6} (oracle {oracle['total_edges']})")
# ── 1. node counts by label ─────────────────────────────────────────────────
print("\n-- nodes by label --")
by_label = kn.node_counts_by_label(g)
for label in sorted(by_label):
print(f" {label:<10} {by_label[label]:>6}")
# ── 2. edge counts by (src,rel,dst) ─────────────────────────────────────────
print("\n-- edges by (src)-[rel]->(dst) --")
by_type = kn.edge_counts_by_type(g)
for key in sorted(by_type):
s, r, d = key
print(f" ({s})-[{r}]->({d}) {by_type[key]:>6}")
# ── 3. degree distribution (top hubs) ───────────────────────────────────────
print("\n-- degree distribution (top 5 by total degree) --")
deg = sorted(g.degree(), key=lambda kv: kv[1], reverse=True)[:5]
for (label, id_), d in deg:
print(f" {label}:{id_:<6} in={g.in_degree((label, id_))} out={g.out_degree((label, id_))} total={d}")
# ── 4. weakly-connected components ──────────────────────────────────────────
n_wcc = nx.number_weakly_connected_components(g)
largest = max(nx.weakly_connected_components(g), key=len)
print(f"\n-- weakly-connected components: {n_wcc} (largest has {len(largest)} nodes) --")
# ── 5. a shortest-path example: Order -CONTAINS-> Product -IN_CATEGORY-> Category
print("\n-- shortest path example (Order -> ... -> Category) --")
src = next(n for n in g.nodes if n[0] == "Order")
target_cats = {n for n in g.nodes if n[0] == "Category"}
path = None
for cat in target_cats:
if nx.has_path(g, src, cat):
path = nx.shortest_path(g, source=src, target=cat)
break
assert path is not None, "an Order must reach some Category via CONTAINS→IN_CATEGORY"
print(" " + " -> ".join(f"{lbl}:{i}" for lbl, i in path))
assert path[0][0] == "Order" and path[-1][0] == "Category"
# ── 6. self-referential REPORTS_TO — the a7208dc fix (NO self-loop collapse) ─
reports_to = [(u, v) for u, v, d in g.edges(data=True) if d["rel"] == "REPORTS_TO"]
n_selfloops = nx.number_of_selfloops(g)
print("\n-- self-ref REPORTS_TO (proves no self-loop collapse, fix a7208dc) --")
print(f" REPORTS_TO edges : {len(reports_to)}")
print(f" graph self-loops : {n_selfloops} (oracle {oracle['self_loops']})")
sample = reports_to[:3]
print(f" sample REPORTS_TO : {[(u[1], v[1]) for u, v in sample]}")
# ── ASSERTIONS: every networkx-derived count matches knut's oracle ──────────
print("\n== asserting networkx graph == knut oracle ==")
assert g.is_directed() and g.is_multigraph(), "must be a MultiDiGraph"
assert g.number_of_nodes() == oracle["total_nodes"], "node total != oracle"
assert g.number_of_edges() == oracle["total_edges"], "edge total != oracle"
assert by_label == oracle["node_counts"], "per-label node counts != oracle"
oracle_edges = {
(e["src"], e["rel"], e["dst"]): e["count"] for e in oracle["edge_counts"]
}
assert by_type == oracle_edges, "per-type edge counts != oracle"
# self-ref proof: REPORTS_TO present, but NO edge is a true self-loop.
assert len(reports_to) > 0, "REPORTS_TO edges must exist"
assert len(reports_to) == oracle_edges[("Employee", "REPORTS_TO", "Employee")]
assert n_selfloops == oracle["self_loops"] == 0, (
"no employee reports to itself — the a7208dc fix must not collapse "
"the self-ref edge into a self-loop"
)
assert all(u != v for u, v in reports_to), "no REPORTS_TO edge is a self-loop"
# dangling / null bookkeeping carried through faithfully.
assert oracle["dangling"] == 0, "clean bench has zero dangling endpoints"
print(" OK — node total, edge total, per-label node counts, per-type edge")
print(" counts, and the REPORTS_TO self-ref (no self-loop) all match.")
print("\nALL ASSERTIONS PASSED.")
return 0
if __name__ == "__main__":
raise SystemExit(main())