diff --git a/frontend/css/status_strip.css b/frontend/css/status_strip.css new file mode 100644 index 00000000..aa3dc0c4 --- /dev/null +++ b/frontend/css/status_strip.css @@ -0,0 +1,105 @@ +/* Operator status footer — console-style telemetry in panel chrome. */ + +.status-strip { + display: flex; + align-items: baseline; + flex-wrap: wrap; + gap: 0.35rem; + margin-top: 1.25rem; + padding: 0.5rem 0.75rem; + border-radius: var(--radius-sm); + border: 1px solid rgba(255, 255, 255, 0.04); + background: rgba(6, 182, 212, 0.03); + font-family: var(--font-mono); + font-size: 0.68rem; + letter-spacing: 0.04em; + color: var(--text-muted); +} + +.status-strip__label { + color: var(--text-secondary); + font-weight: 600; + letter-spacing: 0.1em; + font-size: 0.62rem; +} + +.status-strip__sep { + opacity: 0.35; +} + +.status-strip__items { + color: var(--accent-cyan); + text-shadow: 0 0 6px rgba(6, 182, 212, 0.12); +} + +.status-strip__updated { + color: var(--text-muted); + font-style: italic; +} + +.status-strip--quiet .status-strip__items { + color: var(--text-muted); + text-shadow: none; + font-style: italic; +} + +.map-topology-hint { + position: absolute; + left: 50%; + bottom: 2.75rem; + transform: translateX(-50%); + z-index: 800; + max-width: min(92%, 28rem); + padding: 0.45rem 0.75rem; + border-radius: var(--radius-sm); + border: 1px solid rgba(245, 158, 11, 0.35); + background: rgba(15, 23, 42, 0.92); + color: var(--accent-amber); + font-family: var(--font-mono); + font-size: 0.68rem; + letter-spacing: 0.03em; + text-align: center; + pointer-events: none; + box-shadow: var(--glow-amber); +} + +.mqtt-runtime { + margin: 0 0 1rem; + padding: 0.55rem 0.75rem; + border-radius: var(--radius-sm); + border: 1px solid rgba(255, 255, 255, 0.05); + background: rgba(0, 229, 160, 0.04); + font-family: var(--font-mono); + font-size: 0.68rem; + letter-spacing: 0.03em; + color: var(--text-muted); +} + +.mqtt-runtime--offline { + background: rgba(239, 68, 68, 0.06); + border-color: rgba(239, 68, 68, 0.25); +} + +.mqtt-runtime--disabled { + background: rgba(100, 116, 139, 0.08); +} + +.mqtt-runtime__label { + color: var(--text-secondary); + font-weight: 600; + letter-spacing: 0.1em; + font-size: 0.62rem; + margin-right: 0.35rem; +} + +.mqtt-runtime__state { + color: var(--accent-green); +} + +.mqtt-runtime--offline .mqtt-runtime__state { + color: var(--accent-red); +} + +.mqtt-runtime--disabled .mqtt-runtime__state { + color: var(--text-muted); +} diff --git a/frontend/css/topology.css b/frontend/css/topology.css new file mode 100644 index 00000000..273c1998 --- /dev/null +++ b/frontend/css/topology.css @@ -0,0 +1,129 @@ +.section[data-section="topology"] { + overflow-y: auto; +} + +.topology-panel { + padding: 1.25rem; + max-width: 1100px; + margin: 0 auto; +} + +.topology-panel__header { + display: flex; + flex-wrap: wrap; + align-items: flex-start; + justify-content: space-between; + gap: 1rem; + margin-bottom: 0.75rem; +} + +.topology-panel__title { + margin: 0; + font-size: 1.1rem; + color: var(--text-primary, #f1f5f9); +} + +.topology-panel__desc { + margin: 0.25rem 0 0; + font-size: 0.78rem; + color: var(--text-secondary, #94a3b8); + max-width: 36rem; +} + +.topology-panel__controls { + display: inline-flex; + gap: 0.35rem; +} + +.topology-btn { + background: rgba(15, 23, 42, 0.8); + border: 1px solid rgba(51, 65, 85, 0.8); + color: var(--text-secondary, #94a3b8); + font-size: 0.72rem; + font-family: var(--font-mono); + padding: 0.25rem 0.55rem; + border-radius: 6px; + cursor: pointer; +} + +.topology-btn--active, +.topology-btn:hover { + color: var(--accent-cyan, #06b6d4); + border-color: rgba(6, 182, 212, 0.45); +} + +.topology-panel__meta { + display: flex; + gap: 1rem; + font-family: var(--font-mono); + font-size: 0.68rem; + color: var(--text-muted); + margin-bottom: 0.5rem; +} + +.topology-panel__canvas-wrap { + position: relative; + background: var(--card-bg, #111827); + border: 1px solid var(--border, #1e293b); + border-radius: 12px; + min-height: 420px; + overflow: hidden; +} + +.topology-svg { + display: block; + width: 100%; + min-height: 420px; +} + +.topology-empty { + position: absolute; + inset: 0; + display: flex; + align-items: center; + justify-content: center; + color: var(--text-secondary, #94a3b8); + font-size: 0.85rem; + pointer-events: none; +} + +.topology-legend { + display: flex; + flex-wrap: wrap; + gap: 1rem; + margin-top: 0.75rem; + font-size: 0.68rem; + color: var(--text-muted); +} + +.topology-swatch { + display: inline-block; + width: 10px; + height: 10px; + border-radius: 50%; + margin-right: 0.3rem; + vertical-align: middle; +} + +.topology-swatch--mt { background: var(--accent-cyan); } +.topology-swatch--mc { background: var(--accent-purple); } +.topology-swatch--ni { background: var(--accent-cyan); } +.topology-swatch--tr { + background: transparent; + border: 1px dashed var(--accent-amber); + border-radius: 2px; + width: 14px; + height: 0; +} +.topology-swatch--weak { + background: transparent; + border: 1px dashed var(--text-muted); + border-radius: 2px; + width: 14px; + height: 0; +} + +@media (max-width: 768px) { + .topology-panel { padding: 0.85rem; } + .topology-panel__header { flex-direction: column; } +} diff --git a/frontend/index.html b/frontend/index.html index 239398cc..0d3092c8 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -21,6 +21,7 @@ + @@ -33,6 +34,7 @@ + @@ -102,6 +104,21 @@ Stats +
  • + + + + + Topology + +
  • @@ -391,6 +408,12 @@ + + @@ -679,6 +702,7 @@

    Service actions

    + @@ -693,6 +717,7 @@

    Service actions

    + @@ -747,6 +772,7 @@

    Service actions

    + diff --git a/frontend/js/app.js b/frontend/js/app.js index b6b237ed..d887db1e 100644 --- a/frontend/js/app.js +++ b/frontend/js/app.js @@ -41,7 +41,7 @@ document.addEventListener('DOMContentLoaded', async () => { const router = new Router({ defaultRoute: 'dashboard', allowedRoutes: [ - 'dashboard', 'stats', 'messages', 'radio', 'terminal', + 'dashboard', 'stats', 'topology', 'messages', 'radio', 'terminal', 'configuration/identity', 'configuration/radio', 'configuration/channels', 'configuration/transmit', 'configuration/mqtt', @@ -502,6 +502,7 @@ function _bootCommandPaletteAndKeymap(router) { const routeCommands = [ ['dashboard', 'Go to Dashboard', 'Pages'], ['stats', 'Go to Stats', 'Pages'], + ['topology', 'Go to Topology', 'Pages'], ['messages', 'Go to Messages', 'Pages'], ['radio', 'Go to Radio', 'Pages'], ['terminal', 'Go to Terminal', 'Pages'], diff --git a/frontend/js/components/node_map.js b/frontend/js/components/node_map.js index c8a88af8..9461daec 100644 --- a/frontend/js/components/node_map.js +++ b/frontend/js/components/node_map.js @@ -404,39 +404,73 @@ class NodeMap { }, 200); } + _showTopologyMapHint(show) { + let el = document.getElementById('map-topology-hint'); + if (!el && this._map) { + el = document.createElement('div'); + el.id = 'map-topology-hint'; + el.className = 'map-topology-hint'; + el.hidden = true; + this._map.getContainer().appendChild(el); + } + if (!el) return; + if (show) { + el.textContent = + 'Links need GPS on both endpoints. Open the Topology tab for the logical graph.'; + el.hidden = false; + } else { + el.hidden = true; + } + } + async _loadTopology() { try { - const res = await fetch('/api/analytics/topology'); - const links = await res.json(); + const res = await fetch('/api/analytics/topology?hours=24'); + const data = await res.json(); + const links = Array.isArray(data) ? data : (data.edges || []); this._topologyLayer.clearLayers(); + const root = getComputedStyle(document.documentElement); + const token = (name, fallback) => root.getPropertyValue(name).trim() || fallback; + const accentCyan = token('--accent-cyan', '#06b6d4'); + const accentAmber = token('--accent-amber', '#f59e0b'); + const textMuted = token('--text-muted', '#64748b'); + + let drawn = 0; for (const link of links) { const srcMarker = this._markers[link.source]; const tgtMarker = this._markers[link.target]; if (!srcMarker || !tgtMarker) continue; + const isNeighbor = link.edge_type === 'neighborinfo'; const line = L.polyline( [srcMarker.getLatLng(), tgtMarker.getLatLng()], { - color: '#f59e0b', - weight: 1.5, - opacity: 0.6, - dashArray: '4, 4', + color: link.weak ? textMuted : (isNeighbor ? accentCyan : accentAmber), + weight: isNeighbor ? 2 : 1.5, + opacity: link.weak ? 0.45 : 0.7, + dashArray: isNeighbor ? null : '6, 4', }, ); const rssiLabel = link.rssi != null ? `RSSI: ${link.rssi} dBm` : ''; const snrLabel = link.snr != null ? `SNR: ${link.snr} dB` : ''; + const typeLabel = link.edge_type ? `Type: ${link.edge_type}` : ''; const tooltip = [ `${link.source} ↔ ${link.target}`, - rssiLabel, snrLabel, + typeLabel, rssiLabel, snrLabel, ].filter(Boolean).join('
    '); line.bindTooltip(tooltip); this._topologyLayer.addLayer(line); + drawn += 1; } + this._showTopologyMapHint( + this._topologyVisible && links.length > 0 && drawn === 0, + ); } catch (e) { console.error('Topology load failed:', e); + this._showTopologyMapHint(false); } } diff --git a/frontend/js/status_strip.js b/frontend/js/status_strip.js new file mode 100644 index 00000000..afba91ea --- /dev/null +++ b/frontend/js/status_strip.js @@ -0,0 +1,65 @@ +/** + * Console-style operator status footer for analytics tabs. + * + * Renders a slim monospace strip at the bottom of a panel: + * TRAFFIC · concentrator · 8,247 pkts · updated 14s ago + */ +class StatusStrip { + constructor(hostEl, label) { + this._host = hostEl; + this._label = (label || 'STATUS').toUpperCase(); + this._root = null; + this._itemsEl = null; + this._updatedEl = null; + } + + mount() { + if (!this._host || this._root) return; + const root = document.createElement('footer'); + root.className = 'status-strip'; + root.setAttribute('role', 'status'); + root.setAttribute('aria-live', 'polite'); + root.innerHTML = ` + ${this._label} + + + + `; + this._host.appendChild(root); + this._root = root; + this._itemsEl = root.querySelector('.status-strip__items'); + this._updatedEl = root.querySelector('.status-strip__updated'); + } + + /** + * @param {string[]} items - dot-separated status fragments + * @param {Date|number|null} updatedAt - when underlying data was fetched + */ + update(items, updatedAt) { + if (!this._root) return; + const parts = (items || []).filter((item) => item != null && String(item).trim()); + this._itemsEl.textContent = parts.length ? parts.join(' · ') : 'waiting for data'; + this._root.classList.toggle('status-strip--quiet', parts.length === 0); + + if (this._updatedEl) { + if (updatedAt) { + this._updatedEl.textContent = ` · updated ${_formatRelative(updatedAt)}`; + } else { + this._updatedEl.textContent = ''; + } + } + } +} + +function _formatRelative(timestamp) { + const ms = timestamp instanceof Date ? timestamp.getTime() : Number(timestamp); + if (!ms) return 'just now'; + const seconds = Math.max(0, Math.floor((Date.now() - ms) / 1000)); + if (seconds < 30) return 'just now'; + if (seconds < 60) return `${seconds}s ago`; + if (seconds < 3600) return `${Math.floor(seconds / 60)}m ago`; + if (seconds < 86400) return `${Math.floor(seconds / 3600)}h ago`; + return `${Math.floor(seconds / 86400)}d ago`; +} + +window.StatusStrip = StatusStrip; diff --git a/frontend/js/topology_tab.js b/frontend/js/topology_tab.js new file mode 100644 index 00000000..6defdc3d --- /dev/null +++ b/frontend/js/topology_tab.js @@ -0,0 +1,277 @@ +/** + * Topology tab: D3 force-directed graph from NEIGHBORINFO, ROUTING, and TRACEROUTE. + */ +class TopologyTab { + constructor(containerId) { + this._container = document.getElementById(containerId); + this._hours = 24; + this._rendered = false; + this._simulation = null; + this._graph = { nodes: [], edges: [], routes: [] }; + this._selectedNode = null; + this._statusStrip = null; + this._fetchedAt = null; + } + + async refresh() { + if (!this._container || typeof d3 === 'undefined') return; + + try { + const res = await fetch(`/api/analytics/topology?hours=${this._hours}`); + if (!res.ok) return; + this._graph = await res.json(); + this._fetchedAt = Date.now(); + if (!this._rendered) { + this._buildLayout(); + this._rendered = true; + } + this._renderGraph(); + } catch (e) { + console.error('Topology refresh failed:', e); + } + } + + _buildLayout() { + this._container.innerHTML = ` +
    +
    +
    +

    Mesh topology

    +

    + Force-directed graph from NEIGHBORINFO, traceroute replies, and routing paths. + Click a node to highlight traced routes. +

    +
    +
    + + + +
    +
    +
    + 0 nodes + 0 links +
    +
    + + +
    +
    + Meshtastic + MeshCore + Neighbor link + Traced link + Weak (< -110 dBm) +
    +
    +
    + `; + + const stripHost = document.getElementById('topo-status-strip-host'); + if (stripHost && window.StatusStrip) { + this._statusStrip = new window.StatusStrip(stripHost, 'TOPOLOGY'); + this._statusStrip.mount(); + } + + this._svg = d3.select('#topo-svg'); + this._emptyEl = document.getElementById('topo-empty'); + this._container.querySelectorAll('[data-hours]').forEach((btn) => { + btn.addEventListener('click', () => { + this._hours = Number(btn.dataset.hours); + this._container.querySelectorAll('[data-hours]').forEach((b) => { + b.classList.toggle('topology-btn--active', b === btn); + }); + this.refresh(); + }); + }); + + window.addEventListener('resize', () => { + if (this._rendered) this._renderGraph(); + }); + } + + _cssToken(name, fallback) { + return getComputedStyle(document.documentElement) + .getPropertyValue(name).trim() || fallback; + } + + _ensureGraphNodes(nodes, edges) { + const byId = new Map(nodes.map((n) => [n.id, n])); + for (const edge of edges) { + for (const id of [edge.source, edge.target]) { + if (!id || byId.has(id)) continue; + const stub = { + id, + label: `!${String(id).slice(-4)}`, + protocol: 'meshtastic', + packet_count: 0, + latest_rssi: null, + }; + nodes.push(stub); + byId.set(id, stub); + } + } + return nodes; + } + + _renderGraph() { + const wrap = this._container.querySelector('.topology-panel__canvas-wrap'); + const width = wrap.clientWidth || 800; + const height = Math.max(420, Math.min(560, width * 0.55)); + + const accentCyan = this._cssToken('--accent-cyan', '#06b6d4'); + const accentPurple = this._cssToken('--accent-purple', '#a855f7'); + const accentAmber = this._cssToken('--accent-amber', '#f59e0b'); + const accentGreen = this._cssToken('--accent-green', '#00e5a0'); + const textMuted = this._cssToken('--text-muted', '#64748b'); + const textPrimary = this._cssToken('--text-primary', '#e2e8f0'); + + let nodes = (this._graph.nodes || []).map((n) => ({ ...n })); + const edges = (this._graph.edges || []).map((e) => ({ ...e })); + nodes = this._ensureGraphNodes(nodes, edges); + + const sources = (this._graph.edge_sources || []).join(', ') || 'none'; + const generated = this._graph.generated_at + ? new Date(this._graph.generated_at).toLocaleTimeString() + : 'now'; + document.getElementById('topo-node-count').textContent = + `${nodes.length} node${nodes.length === 1 ? '' : 's'}`; + document.getElementById('topo-edge-count').textContent = + `${edges.length} link${edges.length === 1 ? '' : 's'} · ${sources} · ${generated}`; + + const st = this._graph.stats || {}; + const ni = st.neighborinfo_packets ?? 0; + const tr = st.traceroute_packets ?? 0; + const rt = st.routing_packets ?? 0; + this._statusStrip?.update( + [ + `${nodes.length} nodes`, + `${edges.length} links`, + sources !== 'none' ? sources : 'awaiting edge data', + `${this._hours}h window`, + `pkts ${ni} NI / ${tr} trace / ${rt} route`, + ], + this._fetchedAt, + ); + + if (!nodes.length && !edges.length) { + this._svg.selectAll('*').remove(); + this._emptyEl.hidden = false; + this._emptyEl.textContent = + `No links in the last ${this._hours}h. Packets seen: ` + + `${ni} neighborinfo, ${tr} traceroute, ${rt} routing. ` + + 'Run a traceroute from the Meshtastic app or wait for the next neighbor broadcast.'; + return; + } + this._emptyEl.hidden = true; + + const svg = this._svg + .attr('viewBox', `0 0 ${width} ${height}`) + .attr('width', '100%') + .attr('height', height); + + svg.selectAll('*').remove(); + + const maxPackets = Math.max(1, ...nodes.map((n) => n.packet_count || 0)); + const radius = (count) => 6 + (count / maxPackets) * 14; + + const routeEdgeKeys = this._routeEdgeKeys(this._selectedNode); + + const link = svg.append('g') + .attr('stroke-opacity', 0.75) + .selectAll('line') + .data(edges) + .join('line') + .attr('stroke', (d) => { + if (routeEdgeKeys.has(this._edgeKey(d))) return accentGreen; + if (d.weak) return textMuted; + if (d.edge_type === 'neighborinfo') return accentCyan; + return accentAmber; + }) + .attr('stroke-width', (d) => (routeEdgeKeys.has(this._edgeKey(d)) ? 2.5 : d.weak ? 1 : 1.75)) + .attr('stroke-dasharray', (d) => { + if (d.weak) return '4 3'; + if (d.edge_type && d.edge_type !== 'neighborinfo') return '6 4'; + return null; + }); + + const node = svg.append('g') + .selectAll('circle') + .data(nodes) + .join('circle') + .attr('r', (d) => radius(d.packet_count || 0)) + .attr('fill', (d) => (d.protocol === 'meshcore' ? accentPurple : accentCyan)) + .attr('stroke', (d) => (d.id === this._selectedNode ? textPrimary : 'rgba(15,23,42,0.8)')) + .attr('stroke-width', (d) => (d.id === this._selectedNode ? 2 : 1)) + .style('cursor', 'pointer') + .call(this._drag(svg)); + + node.append('title').text((d) => { + const rssi = d.latest_rssi != null ? `${d.latest_rssi} dBm` : 'n/a'; + return `${d.label} (!${d.id})\n${d.packet_count || 0} pkts · ${rssi}`; + }); + + node.on('click', (_, d) => { + this._selectedNode = this._selectedNode === d.id ? null : d.id; + this._renderGraph(); + }); + + if (this._simulation) this._simulation.stop(); + this._simulation = d3.forceSimulation(nodes) + .force('link', d3.forceLink(edges).id((d) => d.id).distance(90).strength(0.5)) + .force('charge', d3.forceManyBody().strength(-180)) + .force('center', d3.forceCenter(width / 2, height / 2)) + .force('collide', d3.forceCollide().radius((d) => radius(d.packet_count || 0) + 4)) + .on('tick', () => { + link + .attr('x1', (d) => d.source.x) + .attr('y1', (d) => d.source.y) + .attr('x2', (d) => d.target.x) + .attr('y2', (d) => d.target.y); + node + .attr('cx', (d) => d.x) + .attr('cy', (d) => d.y); + }); + } + + _edgeKey(edge) { + const a = typeof edge.source === 'object' ? edge.source.id : edge.source; + const b = typeof edge.target === 'object' ? edge.target.id : edge.target; + return `${Math.min(a, b)}_${Math.max(a, b)}`; + } + + _routeEdgeKeys(nodeId) { + const keys = new Set(); + if (!nodeId) return keys; + for (const route of this._graph.routes || []) { + if (!route.route || !route.route.includes(nodeId)) continue; + for (let i = 0; i < route.route.length - 1; i++) { + const a = route.route[i]; + const b = route.route[i + 1]; + keys.add(`${Math.min(a, b)}_${Math.max(a, b)}`); + } + } + return keys; + } + + _drag(svg) { + const self = this; + function dragstarted(event, d) { + if (!event.active && self._simulation) self._simulation.alphaTarget(0.3).restart(); + d.fx = d.x; + d.fy = d.y; + } + function dragged(event, d) { + d.fx = event.x; + d.fy = event.y; + } + function dragended(event, d) { + if (!event.active && self._simulation) self._simulation.alphaTarget(0); + d.fx = null; + d.fy = null; + } + return d3.drag().on('start', dragstarted).on('drag', dragged).on('end', dragended); + } +} + +window.topologyTab = new TopologyTab('topology-panel'); diff --git a/frontend/sidebar/sidebar_controller.js b/frontend/sidebar/sidebar_controller.js index 800cc4ff..68983cf0 100644 --- a/frontend/sidebar/sidebar_controller.js +++ b/frontend/sidebar/sidebar_controller.js @@ -183,6 +183,9 @@ class SidebarController { if (root === 'stats' && window.statsTab) { window.statsTab.refresh(); } + if (root === 'topology' && window.topologyTab) { + window.topologyTab.refresh(); + } if (route === 'terminal' && window.terminalController) { window.terminalController.onActivated(); } diff --git a/src/api/routes/analytics.py b/src/api/routes/analytics.py index 69e0501a..96549f88 100644 --- a/src/api/routes/analytics.py +++ b/src/api/routes/analytics.py @@ -1,8 +1,6 @@ from __future__ import annotations -import json - -from fastapi import APIRouter +from fastapi import APIRouter, Query from src.analytics.signal_analyzer import SignalAnalyzer from src.analytics.traffic_monitor import TrafficMonitor @@ -52,43 +50,9 @@ async def signal_summary(): @router.get("/topology") -async def network_topology(): - """Extract node-to-node links from NEIGHBORINFO packets.""" +async def network_topology(hours: float = Query(24, ge=1, le=168)): + """Mesh graph data from NEIGHBORINFO edges and TRACEROUTE paths.""" if not _packet_repo: - return [] - - rows = await _packet_repo._db.fetch_all( - """ - SELECT source_id, decoded_payload, rssi, snr, timestamp - FROM packets - WHERE packet_type = 'neighborinfo' AND decoded_payload IS NOT NULL - ORDER BY timestamp DESC - """, - ) - - seen_links: dict[str, dict] = {} - for row in rows: - source = row["source_id"] - try: - payload = json.loads(row["decoded_payload"]) - except (json.JSONDecodeError, TypeError): - continue - - neighbors = payload.get("neighbors", []) - if isinstance(neighbors, list): - for neighbor in neighbors: - nid = neighbor.get("node_id") or neighbor.get("id") - if not nid: - continue - nid = str(nid) - link_key = f"{min(source, nid)}_{max(source, nid)}" - if link_key not in seen_links: - seen_links[link_key] = { - "source": source, - "target": nid, - "rssi": row.get("rssi"), - "snr": row.get("snr"), - "last_seen": row["timestamp"], - } - - return list(seen_links.values()) + return {"hours": hours, "nodes": [], "edges": [], "routes": []} + + return await _packet_repo.get_topology_graph(hours) diff --git a/src/storage/packet_repository.py b/src/storage/packet_repository.py index d6e847a7..9b335c6a 100644 --- a/src/storage/packet_repository.py +++ b/src/storage/packet_repository.py @@ -10,6 +10,12 @@ logger = logging.getLogger(__name__) +_EDGE_PRIORITY = { + "neighborinfo": 3, + "routing": 2, + "traceroute": 1, +} + class PacketRepository: """CRUD operations for captured mesh packets.""" @@ -96,15 +102,6 @@ async def get_signal_history( for row in rows ] - async def get_source_id_by_packet_id(self, packet_id: str) -> str: - if not packet_id: - return "" - row = await self._db.fetch_one( - "SELECT source_id FROM packets WHERE packet_id = ? LIMIT 1", - (packet_id,), - ) - return row["source_id"] if row else "" - async def get_by_source( self, source_id: str, limit: int = 100 ) -> list[Packet]: @@ -137,6 +134,432 @@ async def get_type_distribution(self) -> dict[str, int]: ) return {r["packet_type"]: r["cnt"] for r in rows} + async def get_hourly_traffic( + self, + hours: int = 24, + *, + default_sf: int = 11, + default_bw_khz: float = 250.0, + ) -> tuple[list[dict], dict[str, list[dict]]]: + """Hourly packet counts and modem buckets for ToA estimation. + + Returns: + (count_rows, modem_buckets_by_hour) where count_rows has keys + hour_start, meshtastic, meshcore, total; modem buckets are + grouped per hour_start with sf, bw, packet_count, avg_payload. + """ + hours = max(1, min(int(hours), 168)) + since = ( + datetime.now(timezone.utc) - timedelta(hours=hours) + ).isoformat() + + count_rows = await self._db.fetch_all( + """ + SELECT + strftime('%Y-%m-%dT%H:00:00Z', timestamp) AS hour_start, + SUM(CASE WHEN protocol = 'meshtastic' THEN 1 ELSE 0 END) AS meshtastic, + SUM(CASE WHEN protocol = 'meshcore' THEN 1 ELSE 0 END) AS meshcore, + COUNT(*) AS total + FROM packets + WHERE timestamp >= ? + GROUP BY hour_start + ORDER BY hour_start ASC + """, + (since,), + ) + + modem_rows = await self._db.fetch_all( + """ + SELECT + strftime('%Y-%m-%dT%H:00:00Z', timestamp) AS hour_start, + COALESCE(NULLIF(spreading_factor, 0), ?) AS sf, + COALESCE(NULLIF(bandwidth_khz, 0), ?) AS bw, + COUNT(*) AS packet_count, + AVG( + CASE + WHEN LENGTH(COALESCE(decoded_payload, '')) < 20 THEN 20 + ELSE LENGTH(COALESCE(decoded_payload, '')) + END + ) AS avg_payload + FROM packets + WHERE timestamp >= ? + GROUP BY hour_start, sf, bw + ORDER BY hour_start ASC + """, + (default_sf, default_bw_khz, since), + ) + + modem_by_hour: dict[str, list[dict]] = {} + for row in modem_rows: + hour = row["hour_start"] + modem_by_hour.setdefault(hour, []).append(row) + + return count_rows, modem_by_hour + + async def get_signal_buckets( + self, + source_id: str, + hours: float = 24, + bucket_minutes: int = 15, + ) -> list[dict]: + """15-minute RSSI/SNR buckets for node-card sparklines.""" + bucket_minutes = max(1, min(int(bucket_minutes), 60)) + since = ( + datetime.now(timezone.utc) - timedelta(hours=hours) + ).isoformat() + + rows = await self._db.fetch_all( + """ + SELECT + strftime('%Y-%m-%dT%H:', timestamp) || + printf('%02d:00Z', + (CAST(strftime('%M', timestamp) AS INTEGER) / ?) * ?) + AS bucket_start, + AVG(rssi) AS rssi_avg, + AVG(snr) AS snr_avg, + COUNT(*) AS packet_count + FROM packets + WHERE source_id = ? AND timestamp >= ? AND rssi IS NOT NULL + GROUP BY bucket_start + ORDER BY bucket_start ASC + """, + (bucket_minutes, bucket_minutes, source_id, since), + ) + return [ + { + "bucket": row["bucket_start"].replace("Z", "+00:00"), + "rssi_avg": ( + round(row["rssi_avg"], 1) + if row["rssi_avg"] is not None + else None + ), + "snr_avg": ( + round(row["snr_avg"], 1) + if row["snr_avg"] is not None + else None + ), + "packet_count": int(row["packet_count"] or 0), + } + for row in rows + ] + + @staticmethod + def _normalize_node_id(node_id: str | int | None) -> str | None: + if node_id is None: + return None + text = str(node_id).strip().lower().lstrip("!") + if not text: + return None + if len(text) > 8: + text = text[-8:] + return text.zfill(8) + + @staticmethod + def _edge_key(source: str, target: str) -> str: + a, b = sorted((source, target)) + return f"{a}_{b}" + + @staticmethod + def _upsert_topology_edge( + edges_by_key: dict[str, dict], + *, + source: str, + target: str, + edge_type: str, + last_seen: str, + rssi: float | None = None, + snr: float | None = None, + neighbor_snr: float | None = None, + ) -> None: + if source == target: + return + link_key = PacketRepository._edge_key(source, target) + confidence = "high" if edge_type == "neighborinfo" else "observed" + weak = rssi is not None and rssi < -110 + incoming = { + "source": source, + "target": target, + "edge_type": edge_type, + "confidence": confidence, + "rssi": round(rssi, 1) if rssi is not None else None, + "snr": round(snr, 1) if snr is not None else None, + "neighbor_snr": ( + round(neighbor_snr, 1) if neighbor_snr is not None else None + ), + "last_seen": last_seen, + "weak": weak, + } + existing = edges_by_key.get(link_key) + if existing is None: + edges_by_key[link_key] = incoming + return + new_rank = _EDGE_PRIORITY.get(edge_type, 0) + old_rank = _EDGE_PRIORITY.get(existing.get("edge_type", ""), 0) + if new_rank > old_rank: + edges_by_key[link_key] = incoming + elif new_rank == old_rank and last_seen > existing.get("last_seen", ""): + edges_by_key[link_key] = incoming + + @staticmethod + def _edges_from_route_path( + edges_by_key: dict[str, dict], + route: list[str], + *, + edge_type: str, + last_seen: str, + rssi: float | None = None, + snr: float | None = None, + snr_hops: list[float] | None = None, + ) -> None: + if len(route) < 2: + return + for idx in range(len(route) - 1): + hop_snr = None + if snr_hops and idx < len(snr_hops): + hop_snr = snr_hops[idx] + PacketRepository._upsert_topology_edge( + edges_by_key, + source=route[idx], + target=route[idx + 1], + edge_type=edge_type, + last_seen=last_seen, + rssi=rssi, + snr=hop_snr if hop_snr is not None else snr, + ) + + async def get_topology_graph( + self, + hours: float = 24, + *, + edge_limit: int = 1000, + route_limit: int = 200, + ) -> dict: + """Build nodes, mesh edges, and TRACEROUTE paths for the graph tab.""" + hours = max(1.0, min(float(hours), 168.0)) + since = ( + datetime.now(timezone.utc) - timedelta(hours=hours) + ).isoformat() + + edges_by_key: dict[str, dict] = {} + + ni_rows = await self._db.fetch_all( + """ + SELECT source_id, decoded_payload, rssi, snr, timestamp + FROM packets + WHERE packet_type = 'neighborinfo' + AND decoded_payload IS NOT NULL + AND timestamp >= ? + ORDER BY timestamp DESC + LIMIT ? + """, + (since, edge_limit), + ) + + for row in ni_rows: + source = self._normalize_node_id(row["source_id"]) + if not source: + continue + try: + payload = json.loads(row["decoded_payload"]) + except (json.JSONDecodeError, TypeError): + continue + + neighbors = payload.get("neighbors", []) + if not isinstance(neighbors, list): + continue + + for neighbor in neighbors: + nid = neighbor.get("node_id") or neighbor.get("id") + target = self._normalize_node_id(nid) + if not target: + continue + neighbor_snr = neighbor.get("snr") + self._upsert_topology_edge( + edges_by_key, + source=source, + target=target, + edge_type="neighborinfo", + last_seen=row["timestamp"], + rssi=row.get("rssi"), + snr=row.get("snr"), + neighbor_snr=neighbor_snr, + ) + + route_rows = await self._db.fetch_all( + """ + SELECT source_id, decoded_payload, rssi, snr, timestamp + FROM packets + WHERE packet_type = 'traceroute' + AND decoded_payload IS NOT NULL + AND timestamp >= ? + ORDER BY timestamp DESC + LIMIT ? + """, + (since, route_limit), + ) + + routes: list[dict] = [] + seen_routes: set[str] = set() + for row in route_rows: + try: + payload = json.loads(row["decoded_payload"]) + except (json.JSONDecodeError, TypeError): + continue + route_raw = payload.get("route") + if not isinstance(route_raw, list) or len(route_raw) < 2: + continue + normalized = [ + nid for n in route_raw + if (nid := self._normalize_node_id(n)) + ] + if len(normalized) < 2: + continue + key = f"{row['source_id']}:{'-'.join(normalized)}" + if key in seen_routes: + continue + seen_routes.add(key) + routes.append({ + "source": self._normalize_node_id(row["source_id"]) or row["source_id"], + "route": normalized, + "last_seen": row["timestamp"], + }) + snr_hops = payload.get("snr_towards") + if not isinstance(snr_hops, list): + snr_hops = None + self._edges_from_route_path( + edges_by_key, + normalized, + edge_type="traceroute", + last_seen=row["timestamp"], + rssi=row.get("rssi"), + snr=row.get("snr"), + snr_hops=snr_hops, + ) + + routing_rows = await self._db.fetch_all( + """ + SELECT source_id, decoded_payload, rssi, snr, timestamp + FROM packets + WHERE packet_type = 'routing' + AND decoded_payload IS NOT NULL + AND timestamp >= ? + ORDER BY timestamp DESC + LIMIT ? + """, + (since, route_limit), + ) + + for row in routing_rows: + try: + payload = json.loads(row["decoded_payload"]) + except (json.JSONDecodeError, TypeError): + continue + for field in ("route_reply", "route_request"): + route_raw = payload.get(field) + if not isinstance(route_raw, list) or len(route_raw) < 2: + continue + normalized = [ + nid for n in route_raw + if (nid := self._normalize_node_id(n)) + ] + if len(normalized) < 2: + continue + self._edges_from_route_path( + edges_by_key, + normalized, + edge_type="routing", + last_seen=row["timestamp"], + rssi=row.get("rssi"), + snr=row.get("snr"), + ) + + node_ids: set[str] = set() + for edge in edges_by_key.values(): + node_ids.add(edge["source"]) + node_ids.add(edge["target"]) + for route in routes: + node_ids.update(route["route"]) + + nodes: list[dict] = [] + if node_ids: + counts = await self._db.fetch_all( + f""" + SELECT source_id, protocol, COUNT(*) AS packet_count, + MAX(rssi) AS latest_rssi + FROM packets + WHERE timestamp >= ? AND source_id IN ({",".join("?" * len(node_ids))}) + GROUP BY source_id, protocol + """, + (since, *sorted(node_ids)), + ) + count_by_id: dict[str, dict] = {} + for row in counts: + nid = row["source_id"] + entry = count_by_id.setdefault(nid, { + "packet_count": 0, + "latest_rssi": row.get("latest_rssi"), + "protocol": row.get("protocol") or "meshtastic", + }) + entry["packet_count"] += int(row["packet_count"] or 0) + rssi = row.get("latest_rssi") + if rssi is not None and ( + entry["latest_rssi"] is None or rssi > entry["latest_rssi"] + ): + entry["latest_rssi"] = rssi + + meta_rows = await self._db.fetch_all( + f""" + SELECT node_id, long_name, short_name, protocol + FROM nodes + WHERE node_id IN ({",".join("?" * len(node_ids))}) + """, + tuple(sorted(node_ids)), + ) + meta_by_id = {r["node_id"]: r for r in meta_rows} + + for nid in sorted(node_ids): + meta = meta_by_id.get(nid, {}) + stats = count_by_id.get(nid, {}) + label = ( + meta.get("long_name") + or meta.get("short_name") + or f"!{nid[-4:]}" + ) + nodes.append({ + "id": nid, + "label": label, + "protocol": meta.get("protocol") + or stats.get("protocol") + or "meshtastic", + "packet_count": int(stats.get("packet_count") or 0), + "latest_rssi": stats.get("latest_rssi"), + }) + + edges = list(edges_by_key.values()) + edge_sources = sorted({e.get("edge_type", "unknown") for e in edges}) + by_type = {t: 0 for t in ("neighborinfo", "routing", "traceroute")} + for edge in edges: + et = edge.get("edge_type") + if et in by_type: + by_type[et] += 1 + + return { + "hours": hours, + "nodes": nodes, + "edges": edges, + "routes": routes, + "generated_at": datetime.now(timezone.utc).isoformat(), + "edge_sources": edge_sources, + "stats": { + "neighborinfo_packets": len(ni_rows), + "traceroute_packets": len(route_rows), + "routing_packets": len(routing_rows), + "edges_neighborinfo": by_type["neighborinfo"], + "edges_routing": by_type["routing"], + "edges_traceroute": by_type["traceroute"], + }, + } + async def cleanup_old(self, max_retained: int) -> int: total = await self.get_count() if total <= max_retained: diff --git a/tests/test_topology_api.py b/tests/test_topology_api.py new file mode 100644 index 00000000..7e148bd9 --- /dev/null +++ b/tests/test_topology_api.py @@ -0,0 +1,192 @@ +"""Topology graph aggregation from NEIGHBORINFO and TRACEROUTE packets.""" + +from __future__ import annotations + +import json +import unittest +from datetime import datetime, timedelta, timezone + +from src.storage.database import DatabaseManager +from src.storage.packet_repository import PacketRepository + + +class TestTopologyGraph(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.db = DatabaseManager(":memory:") + await self.db.connect() + self.repo = PacketRepository(self.db) + + async def asyncTearDown(self): + await self.db.disconnect() + + async def _insert_neighborinfo( + self, + packet_id: str, + source_id: str, + neighbors: list[dict], + *, + rssi: float = -95.0, + ts: datetime | None = None, + ) -> None: + ts = ts or datetime.now(timezone.utc) + payload = json.dumps({"neighbors": neighbors}) + await self.db.execute( + """ + INSERT INTO packets ( + packet_id, source_id, destination_id, protocol, + packet_type, hop_limit, hop_start, channel_hash, + want_ack, via_mqtt, relay_node, decoded_payload, decrypted, + rssi, snr, frequency_mhz, spreading_factor, + bandwidth_khz, capture_source, timestamp + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + packet_id, source_id, "ffffffff", "meshtastic", "neighborinfo", + 3, 3, 8, 0, 0, 0, payload, 1, + rssi, 5.0, 906.875, 11, 250, "concentrator", ts.isoformat(), + ), + ) + + async def _insert_routing( + self, + packet_id: str, + source_id: str, + route_reply: list[str], + ) -> None: + payload = json.dumps({"route_reply": route_reply}) + ts = datetime.now(timezone.utc).isoformat() + await self.db.execute( + """ + INSERT INTO packets ( + packet_id, source_id, destination_id, protocol, + packet_type, hop_limit, hop_start, channel_hash, + want_ack, via_mqtt, relay_node, decoded_payload, decrypted, + rssi, snr, frequency_mhz, spreading_factor, + bandwidth_khz, capture_source, timestamp + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + packet_id, source_id, "ffffffff", "meshtastic", "routing", + 3, 3, 8, 0, 0, 0, payload, 1, + -92.0, 6.0, 906.875, 11, 250, "concentrator", ts, + ), + ) + + async def _insert_traceroute( + self, + packet_id: str, + source_id: str, + route: list[str], + ) -> None: + payload = json.dumps({"route": route}) + ts = datetime.now(timezone.utc).isoformat() + await self.db.execute( + """ + INSERT INTO packets ( + packet_id, source_id, destination_id, protocol, + packet_type, hop_limit, hop_start, channel_hash, + want_ack, via_mqtt, relay_node, decoded_payload, decrypted, + rssi, snr, frequency_mhz, spreading_factor, + bandwidth_khz, capture_source, timestamp + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + packet_id, source_id, "ffffffff", "meshtastic", "traceroute", + 3, 3, 8, 0, 0, 0, payload, 1, + -90.0, 5.0, 906.875, 11, 250, "concentrator", ts, + ), + ) + + async def test_neighborinfo_edges_and_weak_flag(self): + await self._insert_neighborinfo( + "ni1", "aaaa0001", + [{"node_id": "bbbb0002", "snr": 4.5}], + rssi=-115.0, + ) + await self._insert_neighborinfo( + "ni2", "cccc0003", + [{"node_id": "dddd0004", "snr": 8.0}], + rssi=-88.0, + ) + await self.db.commit() + + graph = await self.repo.get_topology_graph(24) + self.assertEqual(len(graph["edges"]), 2) + weak = [e for e in graph["edges"] if e["weak"]] + strong = [e for e in graph["edges"] if not e["weak"]] + self.assertEqual(len(weak), 1) + self.assertEqual(len(strong), 1) + self.assertEqual(weak[0]["rssi"], -115.0) + + async def test_hours_window_excludes_old_packets(self): + old = datetime.now(timezone.utc) - timedelta(hours=30) + await self._insert_neighborinfo( + "old", "aaaa0001", + [{"node_id": "bbbb0002", "snr": 1.0}], + ts=old, + ) + await self.db.commit() + + graph = await self.repo.get_topology_graph(24) + self.assertEqual(graph["edges"], []) + + async def test_traceroute_paths_included(self): + await self._insert_neighborinfo( + "ni", "aaaa0001", [{"node_id": "bbbb0002", "snr": 3.0}], + ) + await self._insert_traceroute( + "tr", "aaaa0001", ["aaaa0001", "bbbb0002", "cccc0003"], + ) + await self.db.commit() + + graph = await self.repo.get_topology_graph(24) + self.assertEqual(len(graph["routes"]), 1) + self.assertEqual(graph["routes"][0]["route"][1], "bbbb0002") + node_ids = {n["id"] for n in graph["nodes"]} + self.assertIn("cccc0003", node_ids) + + async def test_traceroute_hops_become_edges(self): + await self._insert_traceroute( + "tr", "aaaa0001", ["aaaa0001", "bbbb0002", "cccc0003"], + ) + await self.db.commit() + + graph = await self.repo.get_topology_graph(24) + self.assertEqual(len(graph["edges"]), 2) + types = {e["edge_type"] for e in graph["edges"]} + self.assertEqual(types, {"traceroute"}) + self.assertIn("traceroute", graph["edge_sources"]) + + async def test_routing_route_reply_becomes_edges(self): + await self._insert_routing( + "rt", "dddd0005", + ["dddd0005", "eeee0006", "ffff0007"], + ) + await self.db.commit() + + graph = await self.repo.get_topology_graph(24) + self.assertEqual(len(graph["edges"]), 2) + self.assertTrue(all(e["edge_type"] == "routing" for e in graph["edges"])) + self.assertIn("routing", graph["edge_sources"]) + + async def test_neighborinfo_wins_over_traceroute_for_same_link(self): + await self._insert_traceroute( + "tr", "aaaa0001", ["aaaa0001", "bbbb0002"], + ) + await self._insert_neighborinfo( + "ni", "aaaa0001", [{"node_id": "bbbb0002", "snr": 4.0}], + ) + await self.db.commit() + + graph = await self.repo.get_topology_graph(24) + self.assertEqual(len(graph["edges"]), 1) + self.assertEqual(graph["edges"][0]["edge_type"], "neighborinfo") + self.assertEqual(graph["edges"][0]["confidence"], "high") + + async def test_normalize_node_id_formats(self): + self.assertEqual(PacketRepository._normalize_node_id("!AABB0001"), "aabb0001") + self.assertEqual(PacketRepository._normalize_node_id("aa1"), "00000aa1") + + +if __name__ == "__main__": + unittest.main()