mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 00:50:46 +08:00
oliver
b81e5869a6
Topology MCP adds pull/compact/bundle/suggest-hubs with a no-template skill path; API gains fabric level and NE collection policy; web list pages get paging and denser collect/network workflows. Co-authored-by: Cursor <cursoragent@cursor.com>
284 lines
12 KiB
Python
284 lines
12 KiB
Python
"""Dual-portal eye units: detect, CN-first max-cover layout, shared-portal compose."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from netx_topology_mcp.layout_metrics import count_edge_crossings
|
||
from netx_topology_mcp.layout_ops import build_state_from_nodes_edges
|
||
from netx_topology_mcp.layout_ops.compose_views import ComposeBlock, strip_pack_blocks
|
||
from netx_topology_mcp.layout_ops.dual_units import (
|
||
find_dual_portal_units,
|
||
layout_dual_unit,
|
||
)
|
||
from netx_topology_mcp.layout_ops.state import LayoutParams
|
||
from netx_topology_mcp.layout_structure import analyze_graph_structure
|
||
from netx_topology_mcp.layout_tool import ACTIONS, run_layout_on_graph
|
||
|
||
|
||
def _eye_graph():
|
||
"""Two AN portals + three interior-disjoint access corridors."""
|
||
nodes = [
|
||
{"fabric_node_id": "p1", "name": "BTM-AN1-P", "role": "an", "x": 0, "y": 0},
|
||
{"fabric_node_id": "p2", "name": "BTM-AN2-P", "role": "an", "x": 100, "y": 0},
|
||
{"fabric_node_id": "a1", "name": "BTM-EN-A1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "a2", "name": "BTM-EN-A2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "b1", "name": "BTM-EN-B1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "b2", "name": "BTM-EN-B2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "c1", "name": "BTM-EN-C1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "c2", "name": "BTM-EN-C2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "t1", "name": "BTM-EN-T1", "role": "en", "x": 0, "y": 0},
|
||
]
|
||
edges = [
|
||
{"a_node_id": "p1", "b_node_id": "a1"},
|
||
{"a_node_id": "a1", "b_node_id": "a2"},
|
||
{"a_node_id": "a2", "b_node_id": "p2"},
|
||
{"a_node_id": "p1", "b_node_id": "b1"},
|
||
{"a_node_id": "b1", "b_node_id": "b2"},
|
||
{"a_node_id": "b2", "b_node_id": "p2"},
|
||
{"a_node_id": "p1", "b_node_id": "c1"},
|
||
{"a_node_id": "c1", "b_node_id": "c2"},
|
||
{"a_node_id": "c2", "b_node_id": "p2"},
|
||
{"a_node_id": "a1", "b_node_id": "t1"}, # tail off corridor
|
||
]
|
||
return nodes, edges
|
||
|
||
|
||
def _cn_eye_graph():
|
||
"""CN pair covers more NEs than a small AN ring — detection must prefer CN."""
|
||
nodes = [
|
||
{"fabric_node_id": "cn1", "name": "BTM-CN1", "role": "cn", "x": 0, "y": 0},
|
||
{"fabric_node_id": "cn2", "name": "BTM-CN2", "role": "cn", "x": 100, "y": 0},
|
||
{"fabric_node_id": "an1", "name": "BTM-AN1", "role": "an", "x": 0, "y": 0},
|
||
{"fabric_node_id": "an2", "name": "BTM-AN2", "role": "an", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e1", "name": "BTM-EN1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e2", "name": "BTM-EN2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e3", "name": "BTM-EN3", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e4", "name": "BTM-EN4", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e5", "name": "BTM-EN5", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e6", "name": "BTM-EN6", "role": "en", "x": 0, "y": 0},
|
||
]
|
||
# Two CN↔CN corridors via AN+EN, plus a tiny AN–AN eye that would steal
|
||
# interiors if AN rings were claimed first.
|
||
edges = [
|
||
{"a_node_id": "cn1", "b_node_id": "cn2"},
|
||
{"a_node_id": "cn1", "b_node_id": "an1"},
|
||
{"a_node_id": "an1", "b_node_id": "e1"},
|
||
{"a_node_id": "e1", "b_node_id": "e2"},
|
||
{"a_node_id": "e2", "b_node_id": "an2"},
|
||
{"a_node_id": "an2", "b_node_id": "cn2"},
|
||
{"a_node_id": "cn1", "b_node_id": "e3"},
|
||
{"a_node_id": "e3", "b_node_id": "e4"},
|
||
{"a_node_id": "e4", "b_node_id": "cn2"},
|
||
{"a_node_id": "an1", "b_node_id": "e5"},
|
||
{"a_node_id": "e5", "b_node_id": "e6"},
|
||
{"a_node_id": "e6", "b_node_id": "an2"},
|
||
]
|
||
return nodes, edges
|
||
|
||
|
||
def test_actions_include_layout_dual_unit() -> None:
|
||
assert "layout_dual_unit" in ACTIONS
|
||
|
||
|
||
def test_find_dual_portal_units_eye() -> None:
|
||
nodes, edges = _eye_graph()
|
||
st = build_state_from_nodes_edges(nodes, edges)
|
||
units = find_dual_portal_units(st)
|
||
assert len(units) >= 1
|
||
u = units[0]
|
||
assert {u.portal_a, u.portal_b} == {"p1", "p2"}
|
||
assert len(u.paths) >= 2
|
||
interiors = set()
|
||
for p in u.paths:
|
||
interiors |= set(p[1:-1])
|
||
assert "a1" in interiors or "b1" in interiors or "c1" in interiors
|
||
|
||
|
||
def test_find_prefers_cn_eye_max_cover() -> None:
|
||
nodes, edges = _cn_eye_graph()
|
||
st = build_state_from_nodes_edges(nodes, edges)
|
||
units = find_dual_portal_units(st)
|
||
assert units
|
||
u0 = units[0]
|
||
assert {u0.portal_a, u0.portal_b} == {"cn1", "cn2"}
|
||
assert len(u0.member_ids()) >= 8
|
||
|
||
|
||
def test_layout_dual_unit_zero_crossings() -> None:
|
||
nodes, edges = _eye_graph()
|
||
st = build_state_from_nodes_edges(nodes, edges)
|
||
op = layout_dual_unit(st, LayoutParams())
|
||
assert op.params.get("accepted") is True
|
||
members = set()
|
||
unit = op.params.get("unit") or {}
|
||
members.update(unit.get("node_ids") or [])
|
||
unit_links = [e for e in op.state.links if e[0] in members and e[1] in members]
|
||
x = count_edge_crossings(op.state.positions, unit_links)
|
||
# Path-planar 3-corridor eye should stay clean under ellipse bands.
|
||
assert x == 0
|
||
assert op.params.get("zero_cross") is True
|
||
|
||
|
||
def test_petal_ellipse_hollow_axis_portals_only() -> None:
|
||
"""Shared portal-adj nodes stay off the mid-chord (CN gravity / hollow eye)."""
|
||
nodes = [
|
||
{"fabric_node_id": "p1", "name": "P1", "role": "cn", "x": 0, "y": 0},
|
||
{"fabric_node_id": "p2", "name": "P2", "role": "cn", "x": 0, "y": 0},
|
||
{"fabric_node_id": "s", "name": "SHARED-AN", "role": "an", "x": 0, "y": 0},
|
||
{"fabric_node_id": "u1", "name": "U1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "u2", "name": "U2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "d1", "name": "D1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "d2", "name": "D2", "role": "en", "x": 0, "y": 0},
|
||
]
|
||
# Two corridors share first hop `s`: p1-s-u1-u2-p2 and p1-s-d1-d2-p2.
|
||
edges = [
|
||
{"a_node_id": "p1", "b_node_id": "s"},
|
||
{"a_node_id": "s", "b_node_id": "u1"},
|
||
{"a_node_id": "u1", "b_node_id": "u2"},
|
||
{"a_node_id": "u2", "b_node_id": "p2"},
|
||
{"a_node_id": "s", "b_node_id": "d1"},
|
||
{"a_node_id": "d1", "b_node_id": "d2"},
|
||
{"a_node_id": "d2", "b_node_id": "p2"},
|
||
]
|
||
st = build_state_from_nodes_edges(nodes, edges)
|
||
op = layout_dual_unit(st, LayoutParams(pitch=200.0, lane=300.0))
|
||
pos = op.state.positions
|
||
# Axis = portals only; shared AN sits on an arc band.
|
||
assert abs(pos["p1"][1]) < 1e-6 and abs(pos["p2"][1]) < 1e-6
|
||
assert abs(pos["s"][1]) > 50.0
|
||
assert pos["u1"][1] * pos["d1"][1] < 0
|
||
assert abs(pos["u1"][1]) > 200.0
|
||
assert abs(pos["d1"][1]) > 200.0
|
||
|
||
|
||
def test_short_corridor_an_not_on_mid_axis() -> None:
|
||
"""CN–AN–CN must not drop the AN onto (0,0) mid-chord."""
|
||
nodes = [
|
||
{"fabric_node_id": "cn1", "name": "CN1", "role": "cn", "x": 0, "y": 0},
|
||
{"fabric_node_id": "cn2", "name": "CN2", "role": "cn", "x": 0, "y": 0},
|
||
{"fabric_node_id": "an1", "name": "AN1", "role": "an", "x": 0, "y": 0},
|
||
{"fabric_node_id": "an2", "name": "AN2", "role": "an", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e1", "name": "E1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "e2", "name": "E2", "role": "en", "x": 0, "y": 0},
|
||
]
|
||
edges = [
|
||
{"a_node_id": "cn1", "b_node_id": "an1"},
|
||
{"a_node_id": "an1", "b_node_id": "cn2"},
|
||
{"a_node_id": "cn1", "b_node_id": "e1"},
|
||
{"a_node_id": "e1", "b_node_id": "e2"},
|
||
{"a_node_id": "e2", "b_node_id": "cn2"},
|
||
{"a_node_id": "cn1", "b_node_id": "an2"},
|
||
{"a_node_id": "an2", "b_node_id": "cn2"},
|
||
]
|
||
st = build_state_from_nodes_edges(nodes, edges)
|
||
op = layout_dual_unit(
|
||
st, LayoutParams(pitch=200.0, lane=300.0), portal_a="cn1", portal_b="cn2"
|
||
)
|
||
pos = op.state.positions
|
||
assert abs(pos["an1"][1]) > 50.0
|
||
assert abs(pos["an2"][1]) > 50.0
|
||
on_axis = {n for n, (_x, y) in pos.items() if abs(y) < 1e-6}
|
||
assert on_axis <= {"cn1", "cn2"}
|
||
# Hollow-ish: short ANs should not sit at exact mid vertical.
|
||
assert abs(pos["an1"][0]) > 1.0
|
||
assert abs(pos["an2"][0]) > 1.0
|
||
|
||
|
||
def test_long_tail_parks_outside_eye_rings() -> None:
|
||
"""Long deg≤2 tails must sit outside corridor envelope (no ring pierce)."""
|
||
nodes = [
|
||
{"fabric_node_id": "p1", "name": "P1", "role": "cn", "x": 0, "y": 0},
|
||
{"fabric_node_id": "p2", "name": "P2", "role": "cn", "x": 0, "y": 0},
|
||
{"fabric_node_id": "a1", "name": "A1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "a2", "name": "A2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "b1", "name": "B1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "b2", "name": "B2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "t0", "name": "T0", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "t1", "name": "T1", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "t2", "name": "T2", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "t3", "name": "T3", "role": "en", "x": 0, "y": 0},
|
||
{"fabric_node_id": "t4", "name": "T4", "role": "en", "x": 0, "y": 0},
|
||
]
|
||
edges = [
|
||
{"a_node_id": "p1", "b_node_id": "a1"},
|
||
{"a_node_id": "a1", "b_node_id": "a2"},
|
||
{"a_node_id": "a2", "b_node_id": "p2"},
|
||
{"a_node_id": "p1", "b_node_id": "b1"},
|
||
{"a_node_id": "b1", "b_node_id": "b2"},
|
||
{"a_node_id": "b2", "b_node_id": "p2"},
|
||
# Long tail hanging off upper corridor mid a1.
|
||
{"a_node_id": "a1", "b_node_id": "t0"},
|
||
{"a_node_id": "t0", "b_node_id": "t1"},
|
||
{"a_node_id": "t1", "b_node_id": "t2"},
|
||
{"a_node_id": "t2", "b_node_id": "t3"},
|
||
{"a_node_id": "t3", "b_node_id": "t4"},
|
||
]
|
||
st = build_state_from_nodes_edges(nodes, edges)
|
||
op = layout_dual_unit(st, LayoutParams(pitch=200.0, lane=300.0))
|
||
pos = op.state.positions
|
||
corridor = {"p1", "p2", "a1", "a2", "b1", "b2"}
|
||
eye_ry = max(abs(pos[n][1]) for n in corridor if n in pos)
|
||
eye_rx = max(abs(pos[n][0]) for n in corridor if n in pos)
|
||
for tid in ("t0", "t1", "t2", "t3", "t4"):
|
||
x, y = pos[tid]
|
||
assert abs(x) >= eye_rx * 0.9 or abs(y) >= eye_ry * 0.9, (tid, x, y, eye_rx, eye_ry)
|
||
|
||
|
||
def test_layout_dual_unit_accepts_residual_cross_by_default() -> None:
|
||
"""With require_zero_cross=false (default), accepted stays true even if x>0."""
|
||
nodes, edges = _cn_eye_graph()
|
||
st = build_state_from_nodes_edges(nodes, edges)
|
||
op = layout_dual_unit(
|
||
st,
|
||
LayoutParams(),
|
||
portal_a="cn1",
|
||
portal_b="cn2",
|
||
require_zero_cross=False,
|
||
)
|
||
assert op.params.get("accepted") is True
|
||
assert "unit_crossings" in op.params
|
||
# Hard gate still available for callers that want the old behavior.
|
||
op_hard = layout_dual_unit(
|
||
st,
|
||
LayoutParams(),
|
||
portal_a="cn1",
|
||
portal_b="cn2",
|
||
require_zero_cross=True,
|
||
)
|
||
if int(op_hard.params.get("unit_crossings") or 0) > 0:
|
||
assert op_hard.params.get("accepted") is False
|
||
else:
|
||
assert op_hard.params.get("accepted") is True
|
||
|
||
|
||
def test_structure_reports_dual_units() -> None:
|
||
nodes, edges = _eye_graph()
|
||
report = analyze_graph_structure(nodes, edges)
|
||
du = report.get("dual_units") or {}
|
||
assert int(du.get("unit_count") or 0) >= 1
|
||
|
||
|
||
def test_shared_portal_compose_preserves_relative() -> None:
|
||
a = ComposeBlock(
|
||
key="unit-p1-p2",
|
||
positions={
|
||
"p1": (0.0, 0.0),
|
||
"p2": (200.0, 0.0),
|
||
"a1": (100.0, 80.0),
|
||
},
|
||
)
|
||
b = ComposeBlock(
|
||
key="unit-p2-p3",
|
||
positions={
|
||
"p2": (0.0, 0.0),
|
||
"p3": (200.0, 0.0),
|
||
"b1": (100.0, -60.0),
|
||
},
|
||
)
|
||
merged, meta = strip_pack_blocks([a, b], pad=100.0, merge_shared=True)
|
||
assert meta.get("merge_shared") is True
|
||
assert "p2" in merged and "p1" in merged and "p3" in merged
|
||
# Shared portal merge keeps both blocks; relative spans stay order-of-magnitude.
|
||
span_p = ((merged["p3"][0] - merged["p2"][0]) ** 2 + (merged["p3"][1] - merged["p2"][1]) ** 2) ** 0.5
|
||
span_b = ((merged["b1"][0] - merged["p2"][0]) ** 2 + (merged["b1"][1] - merged["p2"][1]) ** 2) ** 0.5
|
||
assert abs(span_p - 200.0) < 10.0
|
||
assert abs(span_b - ((100.0**2 + 60.0**2) ** 0.5)) < 10.0
|