"""Mesh Network tab โ live topology from the capability bus registry.
Shows all peers this node has discovered, their capabilities, and an SVG
topology graph. Data is sourced exclusively from bus.registry.all_remote()
and bus.registry.all_local() โ no hardcoded or simulated nodes.
Spec: docs/M02-discovery.md, docs/M03-bus.md ยง4 (registry)
"""
from __future__ import annotations
import html as html_lib
import math
def _topology_svg(this_node: str, peers: list[dict]) -> str:
"""Build an SVG graph from live registry data. No fake data."""
all_nodes = [{"id": this_node[:24], "role": "this node", "is_self": True}]
for p in peers:
all_nodes.append(
{
"id": p["node_id"][:24],
"role": f"{p['capability_count']} caps",
"is_self": False,
}
)
if len(all_nodes) == 1:
return (
"
"
"
No peers discovered yet.
"
"
Start a second HearthNet node and run net.mesh_discover(),"
" or enable mDNS/UDP discovery.
"
"
See docs/HOWTO.md ยง3 for step-by-step instructions.
"
"
"
)
n = len(all_nodes)
cx, cy, r_orbit = 250, 220, 150
items: list[tuple[float, float, dict]] = []
for i, node in enumerate(all_nodes):
angle = (i / n) * math.tau - math.pi / 2
x = cx + r_orbit * math.cos(angle)
y = cy + r_orbit * math.sin(angle)
items.append((x, y, node))
lines: list[str] = []
circles: list[str] = []
labels: list[str] = []
# Lines from this node to each peer
self_x, self_y = items[0][0], items[0][1]
for x, y, node in items[1:]:
lines.append(
f''
)
for x, y, node in items:
fill = "#4CAF50" if node["is_self"] else "#2196F3"
circles.append(f'')
labels.append(
f''
f"{html_lib.escape(node['id'])}"
)
labels.append(
f'{html_lib.escape(node["role"])}'
)
svg = (
'"
'
'
"๐ข this node | ๐ต peers | "
"dashed lines = active capability-bus connections
"
)
return svg
def build_mesh_tab(bus=None):
import gradio as gr
with gr.Column():
gr.Markdown("""### ๐ Mesh Network
Live view of every node this HearthNet instance has discovered.
Each entry is a real peer registered in the capability bus โ no simulated data.
**How peers appear here:**
1. Run a second HearthNet node on the same LAN
2. Both nodes auto-discover each other via mDNS/UDP (M02)
3. Each node advertises its capabilities on the bus (M03)
4. Click **Refresh** to pull the current registry snapshot
""")
with gr.Row():
refresh_btn = gr.Button("๐ Refresh Mesh", variant="primary", scale=2)
mesh_html = gr.HTML(
value="
Click Refresh to load live mesh topology.
"
)
with gr.Row():
stats_out = gr.JSON(label="Mesh Statistics", visible=False, scale=2)
caps_out = gr.JSON(label="Capability Matrix", visible=False, scale=3)
async def get_mesh():
if bus is None:
svg = (
"
"
"Bus not connected. Run as a real HearthNet node to see live mesh topology."
"
"
)
return svg, gr.update(visible=False), gr.update(visible=False)
try:
remote_entries = list(bus.registry.all_remote())
local_entries = list(bus.registry.all_local())
peer_caps: dict[str, list[str]] = {}
for e in remote_entries:
nid = e.node_id
peer_caps.setdefault(nid, []).append(
f"{e.descriptor.name}@{e.descriptor.version[0]}.{e.descriptor.version[1]}"
)
peers = [
{
"node_id": nid,
"capabilities": caps,
"capability_count": len(caps),
}
for nid, caps in peer_caps.items()
]
this_node = getattr(bus, "node_id_full", "this-node")
local_caps = [
f"{e.descriptor.name}@{e.descriptor.version[0]}.{e.descriptor.version[1]}"
for e in local_entries
]
svg = _topology_svg(this_node, peers)
stats = {
"this_node": this_node,
"peer_count": len(peers),
"local_capabilities": len(local_caps),
"total_mesh_capabilities": len(local_caps)
+ sum(p["capability_count"] for p in peers),
}
# Capability matrix: which node has what
all_cap_names: set[str] = set(local_caps)
for p in peers:
all_cap_names.update(p["capabilities"])
matrix = {
"this_node": {c: (c in local_caps) for c in sorted(all_cap_names)},
}
for p in peers:
matrix[p["node_id"][:20]] = {
c: (c in p["capabilities"]) for c in sorted(all_cap_names)
}
return (
svg,
gr.update(visible=True, value=stats),
gr.update(visible=True, value=matrix),
)
except Exception as exc:
err = f"