diff --git a/backend/src/api/routes.ts b/backend/src/api/routes.ts index 165c946..7b32af5 100644 --- a/backend/src/api/routes.ts +++ b/backend/src/api/routes.ts @@ -14,6 +14,35 @@ router.get('/nodes', async (_req, res) => { } }); +// GET /api/nodes/:id/links — ITM-viable neighbours for a node +router.get('/nodes/:id/links', async (req, res) => { + try { + const id = req.params['id']!; + const result = await query<{ + peer_id: string; peer_name: string | null; observed_count: number; + itm_path_loss_db: number | null; + count_this_to_peer: number; count_peer_to_this: number; + }>( + `SELECT + CASE WHEN node_a_id = $1 THEN node_b_id ELSE node_a_id END AS peer_id, + n.name AS peer_name, + observed_count, + itm_path_loss_db, + CASE WHEN node_a_id = $1 THEN count_a_to_b ELSE count_b_to_a END AS count_this_to_peer, + CASE WHEN node_a_id = $1 THEN count_b_to_a ELSE count_a_to_b END AS count_peer_to_this + FROM node_links + LEFT JOIN nodes n ON n.node_id = CASE WHEN node_a_id = $1 THEN node_b_id ELSE node_a_id END + WHERE (node_a_id = $1 OR node_b_id = $1) AND itm_viable = true + ORDER BY observed_count DESC`, + [id], + ); + res.json(result.rows); + } catch (err) { + console.error('[api] GET /nodes/:id/links', (err as Error).message); + res.status(500).json({ error: 'Internal server error' }); + } +}); + // GET /api/nodes/:id/history?hours=24 router.get('/nodes/:id/history', async (req, res) => { try { diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index 1958536..531fb3e 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -81,15 +81,17 @@ export async function insertPacket(p: { payload?: Record; rawHex: string; advertCount?: number; + pathHashes?: string[]; }): Promise { await pool.query( `INSERT INTO packets (time, packet_hash, rx_node_id, src_node_id, topic, packet_type, route_type, - hop_count, rssi, snr, payload, raw_hex, advert_count) - VALUES (NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`, + hop_count, rssi, snr, payload, raw_hex, advert_count, path_hashes) + VALUES (NOW(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)`, [p.packetHash, p.rxNodeId, p.srcNodeId, p.topic, p.packetType, p.routeType, p.hopCount, p.rssi, p.snr, - p.payload ? JSON.stringify(p.payload) : null, p.rawHex, p.advertCount ?? null] + p.payload ? JSON.stringify(p.payload) : null, p.rawHex, p.advertCount ?? null, + p.pathHashes ?? null] ); } @@ -130,7 +132,7 @@ export async function getLastNPackets(n: number) { const res = await pool.query( `SELECT * FROM ( SELECT DISTINCT ON (packet_hash) time, packet_hash, rx_node_id, src_node_id, - packet_type, hop_count, payload, advert_count + packet_type, hop_count, payload, advert_count, path_hashes FROM packets WHERE time > NOW() - INTERVAL '24 hours' ORDER BY packet_hash, time DESC @@ -141,4 +143,12 @@ export async function getLastNPackets(n: number) { return res.rows; } +/** Returns only viable link pairs — compact for sending in initial WebSocket state. */ +export async function getViableLinkPairs(): Promise<[string, string][]> { + const res = await pool.query<{ node_a_id: string; node_b_id: string }>( + `SELECT node_a_id, node_b_id FROM node_links WHERE itm_viable = true` + ); + return res.rows.map((r) => [r.node_a_id, r.node_b_id]); +} + export { pool }; diff --git a/backend/src/db/schema.sql b/backend/src/db/schema.sql index 1af2d36..b2cbbf5 100644 --- a/backend/src/db/schema.sql +++ b/backend/src/db/schema.sql @@ -72,6 +72,7 @@ SELECT add_retention_policy( ); ALTER TABLE packets ADD COLUMN IF NOT EXISTS advert_count INTEGER; +ALTER TABLE packets ADD COLUMN IF NOT EXISTS path_hashes TEXT[]; -- Indexes for common query patterns CREATE INDEX IF NOT EXISTS packets_hash_idx ON packets (packet_hash, time DESC); @@ -80,6 +81,28 @@ CREATE INDEX IF NOT EXISTS packets_src_idx ON packets (src_node_id, time DESC -- ─── Coverage polygons (one row per node, recalculated on position change) ─── +-- ─── Observed + ITM-validated RF links between nodes ───────────────────────── +-- Populated by the viewshed worker as real packets with path data arrive. +-- node_a_id < node_b_id (sorted) so each pair has exactly one row. + +CREATE TABLE IF NOT EXISTS node_links ( + node_a_id TEXT NOT NULL, + node_b_id TEXT NOT NULL, + observed_count INTEGER NOT NULL DEFAULT 1, + last_observed TIMESTAMPTZ NOT NULL DEFAULT NOW(), + itm_path_loss_db DOUBLE PRECISION, + itm_viable BOOLEAN, + itm_computed_at TIMESTAMPTZ, + count_a_to_b INTEGER NOT NULL DEFAULT 0, + count_b_to_a INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (node_a_id, node_b_id) +); +CREATE INDEX IF NOT EXISTS node_links_b_idx ON node_links(node_b_id); +ALTER TABLE node_links ADD COLUMN IF NOT EXISTS count_a_to_b INTEGER NOT NULL DEFAULT 0; +ALTER TABLE node_links ADD COLUMN IF NOT EXISTS count_b_to_a INTEGER NOT NULL DEFAULT 0; + +-- ─── Coverage polygons (one row per node, recalculated on position change) ─── + CREATE TABLE IF NOT EXISTS node_coverage ( node_id TEXT PRIMARY KEY, geom JSONB NOT NULL, -- GeoJSON Polygon or MultiPolygon diff --git a/backend/src/index.ts b/backend/src/index.ts index e7488cd..d742afd 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -6,8 +6,8 @@ import express from 'express'; import cors from 'cors'; import { rateLimit } from 'express-rate-limit'; import { initDb, query } from './db/index.js'; -import { startMqttClient, onPacket, onNodeSeen, onNodeUpsert } from './mqtt/client.js'; -import { initWebSocketServer, broadcastPacket, broadcastNodeUpdate, broadcastNodeUpsert, queueViewshedJob } from './ws/server.js'; +import { startMqttClient, onPacket, onNodeSeen, onNodeUpsert, backfillHistoricalLinks } from './mqtt/client.js'; +import { initWebSocketServer, broadcastPacket, broadcastNodeUpdate, broadcastNodeUpsert, queueViewshedJob, queueLinkJob } from './ws/server.js'; import apiRoutes from './api/routes.js'; const ALLOWED_ORIGINS = (process.env['ALLOWED_ORIGINS'] ?? '') @@ -45,7 +45,12 @@ async function main() { } // 2. Wire up MQTT → WS broadcast - onPacket((packet) => broadcastPacket(packet)); + onPacket((packet) => { + broadcastPacket(packet); + if (packet.path?.length && packet.rxNodeId) { + queueLinkJob(packet.rxNodeId, packet.srcNodeId, packet.path, packet.hopCount); + } + }); onNodeSeen((nodeId) => broadcastNodeUpdate(nodeId)); onNodeUpsert((node) => { broadcastNodeUpsert(node); @@ -105,6 +110,19 @@ async function main() { // 5. Start MQTT client startMqttClient(); + // 6. Backfill node_links from historical packets (once, if table is empty) + process.nextTick(async () => { + const { rows } = await query<{ count: string }>('SELECT COUNT(*) AS count FROM node_links'); + if (Number(rows[0]?.count ?? 0) === 0) { + console.log('[app] node_links empty — backfilling from historical packets…'); + await backfillHistoricalLinks((rxNodeId, srcNodeId, path, hopCount) => { + queueLinkJob(rxNodeId, srcNodeId, path, hopCount); + }); + } else { + console.log('[app] node_links already populated, skipping historical backfill'); + } + }); + httpServer.listen(PORT, '0.0.0.0', () => { console.log(`[app] listening on http://0.0.0.0:${PORT}`); }); diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index 7187eca..6ccbdff 100644 --- a/backend/src/mqtt/client.ts +++ b/backend/src/mqtt/client.ts @@ -4,7 +4,7 @@ import type { AdvertPayload, GroupTextPayload, TextMessagePayload, TracePayload, PathPayload, AckPayload, } from '@michaelhart/meshcore-decoder'; -import { insertPacket, upsertNode, incrementAdvertCount } from '../db/index.js'; +import { insertPacket, upsertNode, incrementAdvertCount, query } from '../db/index.js'; import type { LivePacket } from '../types/index.js'; type PacketCallback = (packet: LivePacket) => void; @@ -359,8 +359,41 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { payload: innerPayload ?? json, rawHex, advertCount, + pathHashes: path, }); } catch (err) { console.error('[mqtt] db insert failed', (err as Error).message); } } + +/** + * Decode all historical packets with raw_hex and queue link jobs for those with + * relay path data. Called once at startup when node_links is empty. + */ +export async function backfillHistoricalLinks( + queueFn: (rxNodeId: string, srcNodeId: string | undefined, path: string[], hopCount: number | undefined) => void, +): Promise { + const res = await query<{ + rx_node_id: string; src_node_id: string | null; hop_count: number | null; raw_hex: string; + }>( + `SELECT DISTINCT ON (packet_hash) + rx_node_id, src_node_id, hop_count, raw_hex + FROM packets + WHERE rx_node_id IS NOT NULL AND raw_hex IS NOT NULL AND raw_hex != '' + ORDER BY packet_hash, time DESC`, + ); + + let queued = 0; + for (const row of res.rows) { + try { + const decoded = MeshCoreDecoder.decode(row.raw_hex, { keyStore }); + if (decoded?.path && Array.isArray(decoded.path) && decoded.path.length > 0) { + queueFn(row.rx_node_id, row.src_node_id ?? undefined, decoded.path as string[], decoded.pathLength); + queued++; + } + } catch { + // Skip undecipherable packets + } + } + console.log(`[app] historical link backfill: queued ${queued} packets`); +} diff --git a/backend/src/ws/server.ts b/backend/src/ws/server.ts index f326f16..251a996 100644 --- a/backend/src/ws/server.ts +++ b/backend/src/ws/server.ts @@ -3,7 +3,7 @@ import type { IncomingMessage } from 'node:http'; import type { Server } from 'node:http'; import { Redis } from 'ioredis'; import type { WSMessage, LivePacket } from '../types/index.js'; -import { getNodes, getLastNPackets } from '../db/index.js'; +import { getNodes, getLastNPackets, getViableLinkPairs } from '../db/index.js'; const REDIS_CHANNEL = 'meshcore:live'; @@ -49,10 +49,12 @@ export function initWebSocketServer(httpServer: Server): WebSocketServer { // Send initial state: known nodes + last 5 minutes of packets try { - const [nodes, packets] = await Promise.all([getNodes(), getLastNPackets(10)]); + const [nodes, packets, viablePairs] = await Promise.all([ + getNodes(), getLastNPackets(10), getViableLinkPairs(), + ]); const initMsg: WSMessage = { type: 'initial_state', - data: { nodes, packets }, + data: { nodes, packets, viable_pairs: viablePairs }, ts: Date.now(), }; ws.send(JSON.stringify(initMsg)); @@ -101,3 +103,19 @@ export function broadcastNodeUpsert(node: Record): void { export function queueViewshedJob(nodeId: string, lat: number, lon: number): void { void pub.lpush('meshcore:viewshed_jobs', JSON.stringify({ node_id: nodeId, lat, lon })); } + +/** Push a link observation job for a received packet with relay path data. */ +export function queueLinkJob( + rxNodeId: string, + srcNodeId: string | undefined, + pathHashes: string[], + hopCount: number | undefined, +): void { + if (!pathHashes.length) return; + void pub.lpush('meshcore:link_jobs', JSON.stringify({ + rx_node_id: rxNodeId, + src_node_id: srcNodeId, + path_hashes: pathHashes, + hop_count: hopCount, + })); +} diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 4416d65..2fe13fd 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -6,7 +6,7 @@ import { FilterPanel, FILTER_ROWS, type Filters } from './components/FilterPanel import { StatsPanel } from './components/StatsPanel/StatsPanel.js'; import { PacketFeed } from './components/PacketFeed.js'; import { useWebSocket, type WSMessage, type WSReadyState } from './hooks/useWebSocket.js'; -import { useNodes, type LivePacketData, type MeshNode } from './hooks/useNodes.js'; +import { useNodes, type LivePacketData, type MeshNode, type AggregatedPacket } from './hooks/useNodes.js'; import { useCoverage, type NodeCoverage } from './hooks/useCoverage.js'; const DEFAULT_FILTERS: Filters = { @@ -16,6 +16,7 @@ const DEFAULT_FILTERS: Filters = { packetPaths: false, betaPaths: false, betaPathThreshold: 0.5, + links: false, }; // Connectivity indicator @@ -99,49 +100,42 @@ function resolvePathWaypoints( } // ── Beta path helpers ───────────────────────────────────────────────────────── -// Effective range for a node: uses the stored radio-horizon radius from the -// viewshed worker (computed from actual SRTM terrain elevation), clamped 50–80 km. + +const MAX_BETA_HOPS = 15; + +/** Canonical lookup key for a link between two nodes (order-independent). */ +function linkKey(a: string, b: string): string { + return a < b ? `${a}:${b}` : `${b}:${a}`; +} + +function distKm(a: MeshNode, b: MeshNode): number { + const midLat = ((a.lat! + b.lat!) / 2) * (Math.PI / 180); + const dlat = (a.lat! - b.lat!) * 111; + const dlon = (a.lon! - b.lon!) * 111 * Math.cos(midLat); + return Math.hypot(dlat, dlon); +} + +// Fallback range check (used when no ITM data exists for a pair). function nodeRange(nodeId: string, coverage: NodeCoverage[]): number { const cov = coverage.find((c) => c.node_id === nodeId); if (!cov?.radius_m) return 50; return Math.min(80, Math.max(50, cov.radius_m / 1000)); } - -// Returns true if two nodes are within range of each other. -// Threshold is the max of each node's elevation-derived range (50–80 km). function canReach(a: MeshNode, b: MeshNode, coverage: NodeCoverage[]): boolean { const threshold = Math.max(nodeRange(a.node_id, coverage), nodeRange(b.node_id, coverage)); - const midLat = ((a.lat! + b.lat!) / 2) * (Math.PI / 180); - const dlat = (a.lat! - b.lat!) * 111; - const dlon = (a.lon! - b.lon!) * 111 * Math.cos(midLat); - return Math.hypot(dlat, dlon) < threshold; -} - -function pickByInterpolation( - candidates: MeshNode[], - src: MeshNode | null, - rx: MeshNode, - hopIndex: number, - totalHops: number, -): MeshNode { - if (candidates.length === 1) return candidates[0]!; - if (!src?.lat || !src?.lon) return candidates[0]!; - const t = (hopIndex + 1) / (totalHops + 1); - const expLat = src.lat + t * (rx.lat! - src.lat); - const expLon = src.lon + t * (rx.lon! - src.lon); - return candidates.reduce((a, b) => - Math.hypot(a.lat! - expLat, a.lon! - expLon) <= Math.hypot(b.lat! - expLat, b.lon! - expLon) ? a : b, - ); + return distKm(a, b) < threshold; } /** - * Scores a beta path using per-hop confidence: - * - ambiguity factor: 1 / n_candidates (fewer matches → more confident) - * - distance factor: 1.0 if chosen hop is within 50 km of prev node - * 0.3 if out of range - * 0.7 if prev node unknown (first hop, src unseen) - * Last relay → rx is also distance-checked and penalised if out of range. - * Returns null if any hop has zero known candidates. + * Beta path resolver — backtracking DFS working backwards from the receiver. + * + * For each relay prefix (reversed, so we start anchored at a known position): + * 1. Try confirmed neighbours of prevNode first (real observed link data). + * 2. If none, try candidates within mutual radio range (coverage-radius based). + * 3. If a chosen candidate leads to dead ends at subsequent hops, backtrack + * and try the next candidate at this hop. + * 4. If no candidate leads to a valid continuation, skip this hop (up to + * maxSkips total). Skips are bounded so the path can't just drop all hops. */ function resolveBetaPath( pathHashes: string[], @@ -149,55 +143,78 @@ function resolveBetaPath( rx: MeshNode, allNodes: Map, coverage: NodeCoverage[], + linkPairs: Set, ): { path: [number, number][]; confidence: number } | null { if (!rx.lat || !rx.lon || pathHashes.length === 0) return null; + if (pathHashes.length >= MAX_BETA_HOPS) return null; - const resolvedNodes: MeshNode[] = []; - const hopConfidences: number[] = []; - let prevNode: MeshNode | null = src?.lat && src?.lon ? src : null; + type HopResult = { node: MeshNode; conf: number } | null; // null = skipped - for (let i = 0; i < pathHashes.length; i++) { - const prefix = pathHashes[i]!.slice(0, 2).toUpperCase(); - const candidates = Array.from(allNodes.values()).filter( + /** Ordered candidate list for a single hop: confirmed neighbours first, then reachable. */ + function getCandidates(prefix: string, prevNode: MeshNode): Array<{ node: MeshNode; conf: number }> { + const all = Array.from(allNodes.values()).filter( (n) => n.lat && n.lon && (n.role === undefined || n.role === 2) && !n.name?.includes('🚫') && n.node_id.toUpperCase().startsWith(prefix), ); - if (candidates.length === 0) continue; + const confirmedSet = new Set(); + const confirmed = all + .filter((c) => linkPairs.has(linkKey(c.node_id, prevNode.node_id))) + .sort((a, b) => distKm(a, prevNode) - distKm(b, prevNode)) + .map((c) => { confirmedSet.add(c.node_id); return { node: c, conf: 0.9 }; }); + const reachable = all + .filter((c) => !confirmedSet.has(c.node_id) && canReach(c, prevNode, coverage)) + .sort((a, b) => distKm(a, prevNode) - distKm(b, prevNode)) + .slice(0, 2) // cap to top-2 to bound the search tree + .map((c) => ({ node: c, conf: 0.3 / Math.max(1, all.length) })); + return [...confirmed, ...reachable]; + } - const ambiguityFactor = 1.0 / candidates.length; - let chosen: MeshNode; - let coverageFactor: number; + // Allow skipping at most 1/3 of hops — prevents degenerate all-skip paths. + const maxSkips = Math.ceil(pathHashes.length / 3); + // Hard limit on total DFS calls to prevent exponential blowup on ambiguous prefixes. + let budget = 300; - if (prevNode) { - const reachable = candidates.filter((c) => canReach(prevNode!, c, coverage)); - if (reachable.length > 0) { - chosen = pickByInterpolation(reachable, src, rx, i, pathHashes.length); - coverageFactor = 1.0; - } else { - chosen = pickByInterpolation(candidates, src, rx, i, pathHashes.length); - coverageFactor = 0.3; - } - } else { - // First hop, source unknown — can't check inbound coverage - chosen = pickByInterpolation(candidates, src, rx, i, pathHashes.length); - coverageFactor = 0.7; + /** + * Recursive DFS. Returns the hop results (rx-to-src order) or null if budget + * was exhausted before a valid path could be found. + */ + function solve(hopIdx: number, prevNode: MeshNode, skipsLeft: number, visited: Set): HopResult[] | null { + if (hopIdx < 0) return []; + if (--budget <= 0) return null; + + const prefix = pathHashes[hopIdx]!.slice(0, 2).toUpperCase(); + // Exclude nodes already used in this path — MeshCore nodes only relay a packet once. + const options = getCandidates(prefix, prevNode).filter((o) => !visited.has(o.node.node_id)); + + // Try each candidate. If it leads to a dead end, backtrack and try the next. + for (const opt of options) { + const nextVisited = new Set(visited); + nextVisited.add(opt.node.node_id); + const rest = solve(hopIdx - 1, opt.node, skipsLeft, nextVisited); + if (rest !== null) return [opt, ...rest]; } - hopConfidences.push(coverageFactor * ambiguityFactor); - resolvedNodes.push(chosen); - prevNode = chosen; + // No candidate produced a valid continuation — try skipping this hop. + if (skipsLeft > 0) { + const rest = solve(hopIdx - 1, prevNode, skipsLeft - 1, visited); + if (rest !== null) return [null, ...rest]; + } + + return null; // truly stuck — caller will try its next candidate } - // Penalise if the last relay can't reach rx - if (resolvedNodes.length > 0 && !canReach(resolvedNodes[resolvedNodes.length - 1]!, rx, coverage)) { - hopConfidences[hopConfidences.length - 1]! *= 0.3; - } + const raw = solve(pathHashes.length - 1, rx, maxSkips, new Set([rx.node_id])); + if (!raw) return null; - const confidence = hopConfidences.reduce((a, b) => a + b, 0) / hopConfidences.length; + // raw is in rx→src order; reverse to get src→rx order for rendering. + const hops = [...raw].reverse().filter((r): r is { node: MeshNode; conf: number } => r !== null); + if (hops.length === 0) return null; + + const confidence = hops.reduce((sum, h) => sum + h.conf, 0) / hops.length; const pathNodes: MeshNode[] = [ ...(src?.lat && src?.lon ? [src] : []), - ...resolvedNodes, + ...hops.map((h) => h.node), rx, ]; if (pathNodes.length < 2) return null; @@ -241,9 +258,13 @@ export const App: React.FC = () => { const [filters, setFilters] = useState(DEFAULT_FILTERS); const [stats, setStats] = useState({ mqttNodes: 0, staleNodes: 0, packetsDay: 0 }); const [map, setMap] = useState(null); + const [linkPairs, setLinkPairs] = useState>(new Set()); + const [viablePairsArr, setViablePairsArr] = useState<[string, string][]>([]); const [showDisclaimer, setShowDisclaimer] = useState(() => !localStorage.getItem(DISCLAIMER_KEY)); const [packetPath, setPacketPath] = useState<[number, number][] | null>(null); const [betaPacketPath, setBetaPacketPath] = useState<[number, number][] | null>(null); + const [pinnedPacketId, setPinnedPacketId] = useState(null); + const pinnedTimerRef = useRef | null>(null); const dismissDisclaimer = useCallback(() => { localStorage.setItem(DISCLAIMER_KEY, '1'); @@ -294,9 +315,10 @@ export const App: React.FC = () => { // Compute dotted path lines from most-recent packet's source → observer. // Clears after PATH_TTL ms (with a 1s fade), or immediately when the next - // distinct packet arrives. + // distinct packet arrives. Skipped while a packet is pinned by the user. const latestId = packets[0]?.id; useEffect(() => { + if (pinnedPacketId !== null) return; if (pathTimerRef.current) clearTimeout(pathTimerRef.current); if (pathFadeRef.current !== null) { cancelAnimationFrame(pathFadeRef.current); pathFadeRef.current = null; } @@ -318,10 +340,11 @@ export const App: React.FC = () => { // ── Beta path (unambiguous hops + coverage validation) ──────────────────── if (filters.betaPaths && latest?.rxNodeId && latest.path?.length && rx?.lat && rx?.lon) { const src = latest.srcNodeId ? (nodes.get(latest.srcNodeId) ?? null) : null; + const hops = latest.hopCount != null ? latest.path.slice(0, latest.hopCount) : latest.path; const result = resolveBetaPath( - latest.path, + hops, src?.lat && src?.lon ? src : null, - rx, nodes, coverage, + rx, nodes, coverage, linkPairs, ); setBetaPacketPath(result && result.confidence >= filters.betaPathThreshold ? result.path : null); } else { @@ -350,11 +373,77 @@ export const App: React.FC = () => { pathFadeRef.current = requestAnimationFrame(animate); }, PATH_TTL - 1_000); // eslint-disable-next-line react-hooks/exhaustive-deps - }, [latestId, filters.packetPaths, filters.betaPaths]); // eslint-disable-line react-hooks/exhaustive-deps + }, [latestId, filters.packetPaths, filters.betaPaths, pinnedPacketId]); // eslint-disable-line react-hooks/exhaustive-deps + + const handlePacketPin = useCallback((packet: AggregatedPacket) => { + // Toggle: clicking the already-pinned packet unpins it + if (pinnedPacketId === packet.id) { + setPinnedPacketId(null); + if (pinnedTimerRef.current) { clearTimeout(pinnedTimerRef.current); pinnedTimerRef.current = null; } + if (pathTimerRef.current) { clearTimeout(pathTimerRef.current); pathTimerRef.current = null; } + if (pathFadeRef.current !== null) { cancelAnimationFrame(pathFadeRef.current); pathFadeRef.current = null; } + setPacketPath(null); + setBetaPacketPath(null); + setPathOpacity(0.75); + return; + } + + // Clear any running auto timers + if (pathTimerRef.current) { clearTimeout(pathTimerRef.current); pathTimerRef.current = null; } + if (pathFadeRef.current !== null) { cancelAnimationFrame(pathFadeRef.current); pathFadeRef.current = null; } + if (pinnedTimerRef.current) { clearTimeout(pinnedTimerRef.current); pinnedTimerRef.current = null; } + + const rx = packet.rxNodeId ? nodes.get(packet.rxNodeId) : undefined; + + setPacketPath(null); + + if (packet.rxNodeId && packet.path?.length && rx?.lat && rx?.lon) { + const src = packet.srcNodeId ? (nodes.get(packet.srcNodeId) ?? null) : null; + const hops = packet.hopCount != null ? packet.path.slice(0, packet.hopCount) : packet.path; + const result = resolveBetaPath( + hops, src?.lat && src?.lon ? src : null, rx, nodes, coverage, linkPairs, + ); + setBetaPacketPath(result ? result.path : null); + } else { + setBetaPacketPath(null); + } + + setPathOpacity(0.75); + setPinnedPacketId(packet.id); + + // Auto-release after 30s with a 1s fade + pinnedTimerRef.current = setTimeout(() => { + const FADE_MS = 1_000; + const startTime = performance.now(); + const animate = (now: number) => { + const t = Math.min(1, (now - startTime) / FADE_MS); + setPathOpacity(0.75 * (1 - t)); + if (t < 1) { + pathFadeRef.current = requestAnimationFrame(animate); + } else { + pathFadeRef.current = null; + setPacketPath(null); + setBetaPacketPath(null); + setPathOpacity(0.75); + setPinnedPacketId(null); + pinnedTimerRef.current = null; + } + }; + pathFadeRef.current = requestAnimationFrame(animate); + }, 9_000); + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [pinnedPacketId, nodes, coverage, linkPairs, filters.betaPaths, filters.betaPathThreshold]); const handleMessage = useCallback((msg: WSMessage) => { if (msg.type === 'initial_state') { - handleInitialState(msg.data as Parameters[0]); + const data = msg.data as Parameters[0] & { + viable_pairs?: [string, string][]; + }; + handleInitialState(data); + if (data.viable_pairs) { + setLinkPairs(new Set(data.viable_pairs.map(([a, b]) => linkKey(a, b)))); + setViablePairsArr(data.viable_pairs); + } } else if (msg.type === 'packet') { handlePacket(msg.data as LivePacketData); } else if (msg.type === 'node_update') { @@ -436,9 +525,11 @@ export const App: React.FC = () => { showPackets={filters.livePackets} showCoverage={filters.coverage} showClientNodes={filters.clientNodes} + showLinks={filters.links} + viablePairsArr={viablePairsArr} packetPath={packetPath} betaPath={betaPacketPath} - showBetaPaths={filters.betaPaths} + showBetaPaths={filters.betaPaths || pinnedPacketId !== null} pathOpacity={pathOpacity} onMapReady={setMap} /> @@ -447,7 +538,14 @@ export const App: React.FC = () => { {/* ── Live Packet Feed ───────────────────────────────────────────── */} - {filters.livePackets && } + {filters.livePackets && ( + + )} {/* ── Disclaimer modal ───────────────────────────────────────────── */} {showDisclaimer && } diff --git a/frontend/src/components/FilterPanel/FilterPanel.tsx b/frontend/src/components/FilterPanel/FilterPanel.tsx index 658df9a..2fa98e7 100644 --- a/frontend/src/components/FilterPanel/FilterPanel.tsx +++ b/frontend/src/components/FilterPanel/FilterPanel.tsx @@ -7,6 +7,7 @@ export interface Filters { packetPaths: boolean; betaPaths: boolean; betaPathThreshold: number; // 0–1 + links: boolean; } interface FilterPanelProps { @@ -18,6 +19,7 @@ export const FILTER_ROWS: Array<{ key: keyof Filters; label: string; color: stri { key: 'livePackets', label: 'Live Feed', color: '#00c4ff' }, { key: 'packetPaths', label: 'Packet Paths', color: '#00c4ff', hollow: true }, { key: 'betaPaths', label: 'Paths (Beta)', color: '#a855f7', hollow: true }, + { key: 'links', label: 'Links', color: '#fbbf24' }, { key: 'coverage', label: 'Coverage', color: '#00e676' }, { key: 'clientNodes', label: 'Companion / Room', color: '#ff9800' }, ]; diff --git a/frontend/src/components/Map/MapView.tsx b/frontend/src/components/Map/MapView.tsx index 76d7b09..bff47da 100644 --- a/frontend/src/components/Map/MapView.tsx +++ b/frontend/src/components/Map/MapView.tsx @@ -82,6 +82,8 @@ interface MapViewProps { showPackets: boolean; showCoverage: boolean; showClientNodes: boolean; + showLinks: boolean; + viablePairsArr: [string, string][]; packetPath: [number, number][] | null; betaPath: [number, number][] | null; showBetaPaths: boolean; @@ -95,7 +97,7 @@ const DEFAULT_ZOOM = 11; export const MapView: React.FC = ({ nodes, arcs, activeNodes, coverage, showPackets, showCoverage, showClientNodes, - packetPath, betaPath, showBetaPaths, pathOpacity, onMapReady, + showLinks, viablePairsArr, packetPath, betaPath, showBetaPaths, pathOpacity, onMapReady, }) => { const [map, setMap] = useState(null); @@ -170,6 +172,20 @@ export const MapView: React.FC = ({ const coverageRings = useCoverageDisplayRings(coverage); + // Resolve viable link pairs to lat/lon polyline positions + const linkLines = useMemo(() => { + if (!showLinks || viablePairsArr.length === 0) return []; + const lines: [number, number][][] = []; + for (const [aId, bId] of viablePairsArr) { + const a = nodes.get(aId); + const b = nodes.get(bId); + if (a?.lat && a?.lon && b?.lat && b?.lon) { + lines.push([[a.lat, a.lon], [b.lat, b.lon]]); + } + } + return lines; + }, [showLinks, viablePairsArr, nodes]); + return (
@@ -212,6 +228,24 @@ export const MapView: React.FC = ({ )} + {/* Confirmed link lines — ITM-viable node pairs */} + {showLinks && linkLines.length > 0 && ( + + {linkLines.map((positions, i) => ( + + ))} + + )} + {/* Repeater markers — Leaflet default marker pane at zIndex 600 */} {nodesWithPos.map((node) => ( = React.memo(({ node, isActive, nodeCoverage }) => { const [showPreview, setShowPreview] = useState(false); + const [links, setLinks] = useState(null); const timerRef = useRef | null>(null); useEffect(() => () => { if (timerRef.current) clearTimeout(timerRef.current); }, []); @@ -110,7 +117,15 @@ export const NodeMarker: React.FC = React.memo(({ node, isActive, nodeCov position={[node.lat, node.lon]} icon={buildIcon(node.is_online, isActive, isStale, variant)} > - + { + if (links !== null) return; // already fetched + fetch(`https://app.teessidemesh.com/api/nodes/${node.node_id}/links`) + .then((r) => r.json()) + .then((data: NodeLink[]) => setLinks(data)) + .catch(() => setLinks([])); + }, + }}>
{node.name ?? `Unknown ${fallbackName}`}
{node.role !== undefined && node.role !== 2 && ( @@ -157,6 +172,26 @@ export const NodeMarker: React.FC = React.memo(({ node, isActive, nodeCov {showPreview ? 'Showing coverage…' : 'Preview coverage'} )} + {links === null &&
Loading neighbours…
} + {links !== null && links.length > 0 && ( +
+
Confirmed neighbours
+ {links.map((lk) => { + const tx = lk.count_this_to_peer > 0; + const rx = lk.count_peer_to_this > 0; + const arrow = tx && rx ? '↔' : tx ? '→' : '←'; + return ( +
+ {arrow} {lk.peer_name ?? lk.peer_id.slice(0, 8)} + + {lk.observed_count}× seen + {lk.itm_path_loss_db != null && <> · {Math.round(lk.itm_path_loss_db)} dB} + +
+ ); + })} +
+ )}
diff --git a/frontend/src/components/PacketFeed.tsx b/frontend/src/components/PacketFeed.tsx index 5fc8573..5bee68b 100644 --- a/frontend/src/components/PacketFeed.tsx +++ b/frontend/src/components/PacketFeed.tsx @@ -15,11 +15,13 @@ const TYPE_LABELS: Record = { }; interface Props { - packets: AggregatedPacket[]; - nodes: Map; + packets: AggregatedPacket[]; + nodes: Map; + onPacketClick?: (packet: AggregatedPacket) => void; + pinnedPacketId?: string | null; } -export const PacketFeed: React.FC = React.memo(({ packets, nodes }) => ( +export const PacketFeed: React.FC = React.memo(({ packets, nodes, onPacketClick, pinnedPacketId }) => (
= 7 ? ' packet-feed--overflow' : ''}`}> {packets.slice(0, 7).map((p) => { const typeLabel = p.packetType !== undefined @@ -37,8 +39,17 @@ export const PacketFeed: React.FC = React.memo(({ packets, nodes }) => ( ? (p.advertCount === 1 ? 'NEW' : `${p.advertCount}`) : undefined; + const isPinned = pinnedPacketId === p.id; + return ( -
+
onPacketClick?.(p)} + role="button" + tabIndex={0} + onKeyDown={(e) => e.key === 'Enter' && onPacketClick?.(p)} + > {typeLabel} {advertBadge && ( {advertBadge} @@ -53,6 +64,7 @@ export const PacketFeed: React.FC = React.memo(({ packets, nodes }) => ( {p.rxCount > 0 && {p.rxCount}rx} {p.txCount > 0 && {p.txCount}tx} + {isPinned && ●}
); })} diff --git a/frontend/src/hooks/useNodes.ts b/frontend/src/hooks/useNodes.ts index 0c95e71..bda5bb9 100644 --- a/frontend/src/hooks/useNodes.ts +++ b/frontend/src/hooks/useNodes.ts @@ -91,6 +91,7 @@ export function useNodes() { hop_count?: number; payload?: Record; advert_count?: number | null; + path_hashes?: string[] | null; }>; }) => { const nodeMap = new Map(); @@ -115,6 +116,7 @@ export function useNodes() { srcNodeId: row.src_node_id, summary, hopCount: row.hop_count, + path: row.path_hashes ?? undefined, rxCount: 1, txCount: 0, ts: new Date(row.time).getTime(), diff --git a/frontend/src/pages/HomePage.tsx b/frontend/src/pages/HomePage.tsx index d15f97a..c5204f3 100644 --- a/frontend/src/pages/HomePage.tsx +++ b/frontend/src/pages/HomePage.tsx @@ -43,7 +43,7 @@ export const HomePage: React.FC = () => { useEffect(() => { const fetch_ = () => - fetch('/api/stats') + fetch('https://app.teessidemesh.com/api/stats') .then(r => r.json()) .then(d => setStats({ packetsDay: d.packetsDay, totalNodes: d.totalNodes, longestHop: d.longestHop, longestHopHash: d.longestHopHash ?? null })) .catch(() => {}); diff --git a/frontend/src/pages/StatsPage.tsx b/frontend/src/pages/StatsPage.tsx index 1f27d4b..e4a7bc8 100644 --- a/frontend/src/pages/StatsPage.tsx +++ b/frontend/src/pages/StatsPage.tsx @@ -94,7 +94,7 @@ export const StatsPage: React.FC = () => { const [lastUpdate, setLastUpdate] = useState(null); const load = () => { - fetch('/api/stats/charts') + fetch('https://app.teessidemesh.com/api/stats/charts') .then(r => r.json()) .then((d: ChartData) => { setData(d); setLoading(false); setLastUpdate(new Date()); }) .catch(() => setLoading(false)); diff --git a/frontend/src/styles/globals.css b/frontend/src/styles/globals.css index 7abe58b..b2681b2 100644 --- a/frontend/src/styles/globals.css +++ b/frontend/src/styles/globals.css @@ -464,6 +464,44 @@ html, body, #root { border-color: #1ec850; color: #1ec850; } +.node-popup__neighbours-loading { + margin-top: 8px; + font-size: 10px; + color: var(--text-muted); +} +.node-popup__neighbours { + margin-top: 10px; + border-top: 1px solid rgba(255,255,255,0.08); + padding-top: 8px; +} +.node-popup__neighbours-title { + font-size: 9px; + text-transform: uppercase; + letter-spacing: 0.08em; + color: var(--text-muted); + margin-bottom: 5px; +} +.node-popup__neighbour-row { + display: flex; + justify-content: space-between; + align-items: baseline; + gap: 8px; + padding: 2px 0; +} +.node-popup__neighbour-name { + font-size: 11px; + color: var(--text-primary); + white-space: nowrap; + overflow: hidden; + text-overflow: ellipsis; + max-width: 120px; +} +.node-popup__neighbour-meta { + font-size: 10px; + color: var(--text-muted); + white-space: nowrap; + flex-shrink: 0; +} /* ─── Live packet feed (bottom left) ────────────────────────────────────── */ .packet-feed { @@ -480,6 +518,11 @@ html, body, #root { pointer-events: none; } +/* Re-enable pointer events on individual items so they're clickable */ +.packet-item { + pointer-events: auto; +} + /* Fade the oldest (topmost) item only when the feed is full */ .packet-feed--overflow { -webkit-mask-image: linear-gradient(to bottom, transparent 0%, black 40px, black 100%); @@ -505,6 +548,37 @@ html, body, #root { opacity: 0; } +.packet-item--clickable { + cursor: pointer; +} + +.packet-item--clickable:hover { + background: rgba(255, 255, 255, 0.04); + border-color: rgba(255, 255, 255, 0.15); +} + +.packet-item--pinned { + background: rgba(0, 196, 255, 0.08); + border-color: rgba(0, 196, 255, 0.4); +} + +.packet-item--pinned:hover { + background: rgba(0, 196, 255, 0.12); +} + +.packet-item__pin { + font-size: 7px; + color: var(--accent); + opacity: 0.7; + flex-shrink: 0; + animation: pin-pulse 2s ease-in-out infinite; +} + +@keyframes pin-pulse { + 0%, 100% { opacity: 0.7; } + 50% { opacity: 0.3; } +} + @keyframes slide-in { from { transform: translateY(8px); opacity: 0; } to { transform: translateY(0); opacity: 1; } diff --git a/viewshed-worker/worker.py b/viewshed-worker/worker.py index ca7944a..5e35085 100644 --- a/viewshed-worker/worker.py +++ b/viewshed-worker/worker.py @@ -40,8 +40,9 @@ SRTM_DIR = Path(os.environ.get('SRTM_DIR', '/data/srtm')) REDIS_URL = os.environ.get('REDIS_URL', 'redis://redis:6379') DATABASE_URL = os.environ.get('DATABASE_URL') -JOB_QUEUE = 'meshcore:viewshed_jobs' -LIVE_CHANNEL = 'meshcore:live' +JOB_QUEUE = 'meshcore:viewshed_jobs' +LINK_JOB_QUEUE = 'meshcore:link_jobs' +LIVE_CHANNEL = 'meshcore:live' ANTENNA_HEIGHT_M = 5 # observer height above ground (m) — fixed 5 m antenna MAX_RADIUS_M = 100_000 # absolute cap on viewshed radius (m) @@ -53,6 +54,111 @@ STEP_M = 50.0 # ray step size in metres K_FACTOR = 4 / 3 # effective Earth radius multiplier (standard troposphere) R_EARTH_M = 6_371_000 # mean Earth radius (m) +# ── RF propagation model parameters (LoRa 868 MHz) ─────────────────────────── + +FREQ_MHZ = 868.0 +LAMBDA_M = 3e8 / (FREQ_MHZ * 1e6) # wavelength ~0.345 m +LINK_BUDGET_DB = 148.0 # 17 dBm TX + ~130 dBm RX sensitivity (LoRa SF10 BW125) +1 dB SRTM DSM correction +FADE_MARGIN_DB = 10.0 # safety / link margin +PROFILE_STEP_M = 250.0 # terrain profile sample spacing (m) + +def compute_path_loss(lat1: float, lon1: float, elev1: float, + lat2: float, lon2: float, elev2: float, + vrt_path: str) -> tuple[float, bool]: + """Estimate RF path loss (dB) between two points. + + Uses free-space path loss plus ITU-R P.526 single knife-edge diffraction + over the dominant terrain obstruction, corrected for Earth curvature. + Returns (path_loss_db, is_viable). + """ + cos_mid = math.cos(math.radians((lat1 + lat2) / 2)) + dlat = (lat2 - lat1) * 111_320 + dlon = (lon2 - lon1) * 111_320 * cos_mid + d_total = math.sqrt(dlat ** 2 + dlon ** 2) + + if d_total < 1.0: + return 0.0, True + + # Free-space path loss (dB) + fspl = 20 * math.log10(4 * math.pi * d_total / LAMBDA_M) + + # Terrain profile: N evenly-spaced samples along the path + N = max(20, min(200, int(d_total / PROFILE_STEP_M))) + + ds = gdal.Open(vrt_path) + if ds is None: + viable = fspl < LINK_BUDGET_DB + return fspl, viable + + gt = ds.GetGeoTransform() + inv_gt = gdal.InvGeoTransform(gt) + band = ds.GetRasterBand(1) + + heights: list[float] = [] + dists: list[float] = [] + for i in range(N + 1): + t = i / N + la = lat1 + t * (lat2 - lat1) + lo = lon1 + t * (lon2 - lon1) + px, py = gdal.ApplyGeoTransform(inv_gt, lo, la) + px = int(np.clip(px, 0, ds.RasterXSize - 1)) + py = int(np.clip(py, 0, ds.RasterYSize - 1)) + data = band.ReadAsArray(px, py, 1, 1) + h = max(0.0, float(data[0][0])) if data is not None else 0.0 + heights.append(h) + dists.append(t * d_total) + ds = None + + h_tx = elev1 + ANTENNA_HEIGHT_M # transmitter height ASL + antenna + h_rx = elev2 + ANTENNA_HEIGHT_M + + # Find dominant obstruction via Fresnel-Kirchhoff diffraction parameter + max_v = -999.0 + for i in range(1, N): + d1 = dists[i] + d2 = d_total - dists[i] + if d1 <= 0 or d2 <= 0: + continue + los_h = h_tx + (h_rx - h_tx) * (d1 / d_total) + earth_bulge = (d1 * d2) / (2 * K_FACTOR * R_EARTH_M) + excess_h = heights[i] + earth_bulge - los_h + v = excess_h * math.sqrt(2 * (d1 + d2) / (LAMBDA_M * d1 * d2)) + max_v = max(max_v, v) + + # ITU-R P.526 knife-edge diffraction loss (dB) + if max_v <= -0.78: + diff_loss = 0.0 + else: + diff_loss = max(0.0, 6.9 + 20 * math.log10( + math.sqrt((max_v - 0.1) ** 2 + 1) + max_v - 0.1 + )) + + total_loss = fspl + diff_loss + viable = total_loss < LINK_BUDGET_DB + return total_loss, viable + + +def build_link_vrt(lat1: float, lon1: float, lat2: float, lon2: float, + tmp_dir: str) -> Optional[str]: + """Build a GDAL VRT from already-cached SRTM tiles covering the path. + Returns None if no tiles are available (will be retried later once + nearby viewsheds have triggered tile downloads).""" + min_lat = math.floor(min(lat1, lat2)) + max_lat = math.floor(max(lat1, lat2)) + min_lon = math.floor(min(lon1, lon2)) + max_lon = math.floor(max(lon1, lon2)) + paths = [ + str(SRTM_DIR / f'{tile_name(lt, ln)}.hgt') + for lt in range(min_lat, max_lat + 1) + for ln in range(min_lon, max_lon + 1) + if (SRTM_DIR / f'{tile_name(lt, ln)}.hgt').exists() + ] + if not paths: + return None + vrt = f'{tmp_dir}/link.vrt' + r = subprocess.run(['gdalbuildvrt', vrt] + paths, capture_output=True, text=True) + return vrt if r.returncode == 0 else None + # ── UK mainland polygon (loaded once at startup for ocean clipping) ─────────── def _load_uk_mainland(): @@ -360,6 +466,146 @@ def backfill_elevations(db): log.info(f' {node_id[:12]}…: elevation={elevation_m:.0f} m ASL (from radius {radius_m/1000:.1f} km)') db.commit() +def process_link_job(db, r_client, job: dict): + """Resolve relay path prefixes to known nodes (backwards from the receiver), + record observations in node_links, and compute RF path loss for new pairs.""" + rx_node_id = job.get('rx_node_id') + src_node_id = job.get('src_node_id') + path_hashes = job.get('path_hashes', []) + + if not rx_node_id or not path_hashes: + return + + # Load all positioned nodes + with db.cursor() as cur: + cur.execute( + 'SELECT node_id, lat, lon, elevation_m, name, role FROM nodes ' + 'WHERE lat IS NOT NULL AND lon IS NOT NULL' + ) + all_nodes = { + row[0]: {'lat': row[1], 'lon': row[2], 'elevation_m': row[3] or 0.0, + 'name': row[4], 'role': row[5]} + for row in cur.fetchall() + } + + rx = all_nodes.get(rx_node_id) + if not rx: + return + + def node_dist(a: dict, b: dict) -> float: + cos_m = math.cos(math.radians((a['lat'] + b['lat']) / 2)) + return math.sqrt( + ((a['lat'] - b['lat']) * 111.32) ** 2 + + ((a['lon'] - b['lon']) * 111.32 * cos_m) ** 2 + ) + + # Resolve path working backwards from rx (known position anchor) + resolved: list[tuple[str, dict]] = [] # built in reverse, then flipped + prev = rx + for prefix in reversed(path_hashes): + prefix = prefix[:2].upper() + candidates = [ + (nid, nd) for nid, nd in all_nodes.items() + if nid.upper().startswith(prefix) + and (nd['role'] is None or nd['role'] == 2) + and nd['name'] and '🚫' not in nd['name'] + ] + if not candidates: + continue + candidates.sort(key=lambda x: node_dist(x[1], prev)) + best_id, best = candidates[0] + resolved.insert(0, (best_id, best)) + prev = best + + # Build adjacency list: src → relays → rx + full: list[tuple[str, dict]] = [] + if src_node_id and src_node_id in all_nodes: + full.append((src_node_id, all_nodes[src_node_id])) + full.extend(resolved) + full.append((rx_node_id, rx)) + + if len(full) < 2: + return + + # Upsert observations and compute path loss for each adjacent pair. + # full[i] → full[i+1] means full[i] transmitted, full[i+1] received. + with tempfile.TemporaryDirectory() as tmp: + for i in range(len(full) - 1): + src_id, src = full[i] # transmitted + dst_id, dst = full[i + 1] # received + if not src['lat'] or not dst['lat']: + continue + + # Canonical ordering (lower ID first) → unique primary key + if src_id < dst_id: + a_id, a, b_id, b = src_id, src, dst_id, dst + inc_atob, inc_btoa = 1, 0 # src==a transmitted to dst==b + else: + a_id, a, b_id, b = dst_id, dst, src_id, src + inc_atob, inc_btoa = 0, 1 # src==b transmitted to dst==a + + # Upsert observation with directional counts; check whether ITM already computed + with db.cursor() as cur: + cur.execute( + '''INSERT INTO node_links + (node_a_id, node_b_id, observed_count, last_observed, + count_a_to_b, count_b_to_a) + VALUES (%s, %s, 1, NOW(), %s, %s) + ON CONFLICT (node_a_id, node_b_id) DO UPDATE + SET observed_count = node_links.observed_count + 1, + last_observed = NOW(), + count_a_to_b = node_links.count_a_to_b + %s, + count_b_to_a = node_links.count_b_to_a + %s + RETURNING observed_count, itm_computed_at''', + (a_id, b_id, inc_atob, inc_btoa, inc_atob, inc_btoa), + ) + row = cur.fetchone() + obs_count = row[0] if row else 1 + itm_computed = row[1] if row else None + + # Compute ITM path loss if not yet done and tiles are cached + path_loss_db: Optional[float] = None + itm_viable: Optional[bool] = None + if itm_computed is None: + vrt = build_link_vrt(a['lat'], a['lon'], b['lat'], b['lon'], tmp) + if vrt: + try: + path_loss_db, itm_viable = compute_path_loss( + a['lat'], a['lon'], a['elevation_m'], + b['lat'], b['lon'], b['elevation_m'], + vrt, + ) + with db.cursor() as cur: + cur.execute( + '''UPDATE node_links + SET itm_path_loss_db = %s, + itm_viable = %s, + itm_computed_at = NOW() + WHERE node_a_id = %s AND node_b_id = %s''', + (round(path_loss_db, 1), itm_viable, a_id, b_id), + ) + log.info( + f'Link {a_id[:8]}…↔{b_id[:8]}…: ' + f'{path_loss_db:.1f} dB {"✓" if itm_viable else "✗"} ' + f'(obs={obs_count})' + ) + except Exception as exc: + log.warning(f'Path loss computation failed: {exc}') + + # Notify frontend + r_client.publish(LIVE_CHANNEL, json.dumps({ + 'type': 'link_update', + 'data': { + 'node_a_id': a_id, + 'node_b_id': b_id, + 'observed_count': obs_count, + 'itm_path_loss_db': path_loss_db, + 'itm_viable': itm_viable, + }, + 'ts': int(time.time() * 1000), + })) + + def enqueue_uncovered(db, r_client): """On startup, queue all nodes that have a position but no coverage yet.""" # Remove any coverage that was previously computed for hidden or non-repeater nodes. @@ -458,11 +704,22 @@ def worker_loop(): while True: try: - item = r_client.brpop(JOB_QUEUE, timeout=60) + # Drain any pending link jobs first (fast) before blocking on viewshed + while True: + raw = r_client.rpop(LINK_JOB_QUEUE) + if raw is None: + break + process_link_job(db, r_client, json.loads(raw)) + + # Block-wait for a viewshed or link job (viewshed has priority) + item = r_client.brpop([JOB_QUEUE, LINK_JOB_QUEUE], timeout=30) if item is None: continue - _, raw = item - process_job(db, r_client, json.loads(raw)) + queue_name, raw = item + if queue_name == LINK_JOB_QUEUE: + process_link_job(db, r_client, json.loads(raw)) + else: + process_job(db, r_client, json.loads(raw)) except psycopg2.OperationalError: log.warning(f'{name}: DB connection lost — reconnecting') db = wait_for_db()