diff --git a/PROGRESS.md b/PROGRESS.md index dc89e60..7c70dfc 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -12,11 +12,80 @@ Goal: Bare-minimum loop from ESP32 to browser. Zero-config with passive radar an | Passive radar support | **Done** | BSSID filter in csi.c | | BLE scanning | **Done** | Core 0 concurrent with WiFi | | Mothership WebSocket ingestion | **Done** | See iteration 1 below | -| Dashboard skeleton | Not started | | +| Dashboard skeleton | **Done** | See iteration 3 below | | Docker packaging | Not started | | ### Iteration Log +#### Iteration 3 — 2026-03-26 + +**Completed:** Dashboard skeleton with 3D visualization + +Implemented the web dashboard in vanilla JS + Three.js with: + +- **Static file structure:** `dashboard/` directory with HTML/JS + - `index.html` — Dark theme UI with status bar, node/link panels, amplitude chart + - `js/app.js` — Main application (~350 lines) + +- **3D Scene (Three.js):** + - Ground grid (10×10m, 20 divisions) + - OrbitControls for pan/zoom/rotate with damping + - Axes helper for orientation + - Responsive canvas with devicePixelRatio support + - FPS counter in status bar + +- **WebSocket connection (`/ws/dashboard`):** + - Auto-reconnect with 3s backoff + - JSON message handling for state, node events, link events + - Binary CSI frame parsing (24-byte header + I/Q payload) + - Connection status indicator (green/red dot) + +- **Node/Link panels:** + - Live list of connected nodes with MAC, firmware, chip + - Active links with node:peer MAC pairs + - Click-to-select for amplitude chart display + +- **Amplitude chart (Canvas 2D):** + - 64-bar visualization of subcarrier amplitudes + - Real-time update from selected link's CSI frames + - Color gradient based on amplitude intensity + - Channel and RSSI display overlay + +- **Dashboard package (Go):** + - `internal/dashboard/hub.go` — Hub for client management and broadcasting + - `internal/dashboard/server.go` — WebSocket handler with ping/pong keepalive + - `internal/dashboard/hub_test.go` — 5 tests for hub operations + - IngestionState interface for querying node/link state + - CSIBroadcaster interface implementation + +- **Main.go updates:** + - Static file serving with SPA fallback + - Dashboard WebSocket endpoint at `/ws/dashboard` + - Auto-discovery of dashboard directory + - Wiring between ingestion and dashboard packages + +**Dependencies used (frontend):** +- Three.js r128 (CDN) +- OrbitControls (CDN) + +**Tests:** 5 new tests for dashboard hub. All 27 tests pass. + +**Files created:** +``` +dashboard/ +├── index.html +└── js/ + └── app.js + +mothership/internal/dashboard/ +├── hub.go +├── hub_test.go +└── server.go +``` + +**Remaining for Phase 1:** +- Docker packaging + #### Iteration 2 — 2026-03-26 **Completed:** ESP32-S3 firmware skeleton diff --git a/dashboard/index.html b/dashboard/index.html new file mode 100644 index 0000000..aeee685 --- /dev/null +++ b/dashboard/index.html @@ -0,0 +1,240 @@ + + + + + + Spaxel Dashboard + + + + +
+
+
+ Disconnected +
+
+ Nodes: 0 +
+
+ Links: 0 +
+
+ FPS: 0 +
+
+ + +
+ + +
+

Nodes

+
+
No nodes connected
+
+
+

Links

+ +
+
+ + +
+
Amplitude — no link selected
+ +
+ + + + + + + + + + diff --git a/dashboard/js/app.js b/dashboard/js/app.js new file mode 100644 index 0000000..7a6854b --- /dev/null +++ b/dashboard/js/app.js @@ -0,0 +1,530 @@ +/** + * Spaxel Dashboard - Main Application + * + * Phase 1 skeleton: 3D scene with ground grid, OrbitControls, + * WebSocket connection, and amplitude bar chart visualization. + */ + +(function() { + 'use strict'; + + // ============================================ + // Configuration + // ============================================ + const CONFIG = { + wsReconnectDelay: 3000, + gridWidth: 10, // meters + gridDepth: 10, // meters + gridDivisions: 20, + cameraFov: 60, + cameraNear: 0.1, + cameraFar: 1000, + cameraInitial: { x: 8, y: 8, z: 8 }, + chartBars: 64, // number of subcarriers + chartUpdateMs: 100 // update chart at 10 Hz max + }; + + // ============================================ + // State + // ============================================ + const state = { + ws: null, + wsConnected: false, + nodes: new Map(), // MAC -> { mac, firmware, chip, lastSeen } + links: new Map(), // linkID -> { nodeMAC, peerMAC, lastFrame, lastCSI } + selectedLinkID: null, + lastChartUpdate: 0, + frameCount: 0, + lastFpsTime: performance.now() + }; + + // ============================================ + // Three.js Scene Setup + // ============================================ + let scene, camera, renderer, controls, gridHelper, axesHelper; + + function initScene() { + const container = document.getElementById('scene-container'); + + // Scene + scene = new THREE.Scene(); + scene.background = new THREE.Color(0x1a1a2e); + + // Camera + camera = new THREE.PerspectiveCamera( + CONFIG.cameraFov, + window.innerWidth / window.innerHeight, + CONFIG.cameraNear, + CONFIG.cameraFar + ); + camera.position.set( + CONFIG.cameraInitial.x, + CONFIG.cameraInitial.y, + CONFIG.cameraInitial.z + ); + + // Renderer + renderer = new THREE.WebGLRenderer({ antialias: true }); + renderer.setSize(window.innerWidth, window.innerHeight); + renderer.setPixelRatio(window.devicePixelRatio); + container.appendChild(renderer.domElement); + + // OrbitControls + controls = new THREE.OrbitControls(camera, renderer.domElement); + controls.enableDamping = true; + controls.dampingFactor = 0.05; + controls.screenSpacePanning = true; + controls.minDistance = 2; + controls.maxDistance = 50; + controls.maxPolarAngle = Math.PI / 2 + 0.3; // Allow slight below-ground view + + // Grid helper (XZ plane, Y-up) + gridHelper = new THREE.GridHelper( + CONFIG.gridWidth, + CONFIG.gridDivisions, + 0x444466, // center line color + 0x333344 // grid line color + ); + scene.add(gridHelper); + + // Axes helper for orientation + axesHelper = new THREE.AxesHelper(2); + axesHelper.position.set(-CONFIG.gridWidth / 2, 0.01, -CONFIG.gridDepth / 2); + scene.add(axesHelper); + + // Ambient light + const ambientLight = new THREE.AmbientLight(0xffffff, 0.6); + scene.add(ambientLight); + + // Directional light + const directionalLight = new THREE.DirectionalLight(0xffffff, 0.4); + directionalLight.position.set(5, 10, 5); + scene.add(directionalLight); + + // Handle window resize + window.addEventListener('resize', onWindowResize); + + console.log('[Spaxel] Scene initialized'); + } + + function onWindowResize() { + camera.aspect = window.innerWidth / window.innerHeight; + camera.updateProjectionMatrix(); + renderer.setSize(window.innerWidth, window.innerHeight); + } + + function animate() { + requestAnimationFrame(animate); + controls.update(); + renderer.render(scene, camera); + updateFPS(); + } + + function updateFPS() { + state.frameCount++; + const now = performance.now(); + const elapsed = now - state.lastFpsTime; + if (elapsed >= 1000) { + const fps = Math.round(state.frameCount * 1000 / elapsed); + document.getElementById('fps-counter').textContent = fps; + state.frameCount = 0; + state.lastFpsTime = now; + } + } + + // ============================================ + // WebSocket Connection + // ============================================ + function connectWebSocket() { + const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const wsURL = `${wsProtocol}//${window.location.host}/ws/dashboard`; + + console.log('[Spaxel] Connecting to', wsURL); + + state.ws = new WebSocket(wsURL); + state.ws.binaryType = 'arraybuffer'; + + state.ws.onopen = function() { + console.log('[Spaxel] WebSocket connected'); + state.wsConnected = true; + updateConnectionStatus(true); + }; + + state.ws.onclose = function(event) { + console.log('[Spaxel] WebSocket closed:', event.code, event.reason); + state.wsConnected = false; + updateConnectionStatus(false); + scheduleReconnect(); + }; + + state.ws.onerror = function(error) { + console.error('[Spaxel] WebSocket error:', error); + }; + + state.ws.onmessage = function(event) { + handleMessage(event.data); + }; + } + + function scheduleReconnect() { + console.log('[Spaxel] Reconnecting in', CONFIG.wsReconnectDelay, 'ms'); + setTimeout(connectWebSocket, CONFIG.wsReconnectDelay); + } + + function updateConnectionStatus(connected) { + const dot = document.getElementById('ws-status'); + const text = document.getElementById('ws-status-text'); + + if (connected) { + dot.classList.remove('disconnected'); + dot.classList.add('connected'); + text.textContent = 'Connected'; + } else { + dot.classList.remove('connected'); + dot.classList.add('disconnected'); + text.textContent = 'Disconnected'; + } + } + + // ============================================ + // Message Handling + // ============================================ + function handleMessage(data) { + if (typeof data === 'string') { + // JSON message + try { + const msg = JSON.parse(data); + handleJSONMessage(msg); + } catch (e) { + console.error('[Spaxel] Failed to parse JSON:', e); + } + } else if (data instanceof ArrayBuffer) { + // Binary CSI frame + handleBinaryFrame(data); + } + } + + function handleJSONMessage(msg) { + switch (msg.type) { + case 'state': + // Initial state dump + if (msg.nodes) { + msg.nodes.forEach(node => { + state.nodes.set(node.mac, { + mac: node.mac, + firmware: node.firmware_version, + chip: node.chip, + lastSeen: Date.now() + }); + }); + } + if (msg.links) { + msg.links.forEach(link => { + state.links.set(link.id, { + nodeMAC: link.node_mac, + peerMAC: link.peer_mac, + lastFrame: Date.now(), + lastCSI: null + }); + }); + } + updateNodeList(); + updateLinkList(); + break; + + case 'node_connected': + state.nodes.set(msg.mac, { + mac: msg.mac, + firmware: msg.firmware_version, + chip: msg.chip, + lastSeen: Date.now() + }); + updateNodeList(); + break; + + case 'node_disconnected': + state.nodes.delete(msg.mac); + updateNodeList(); + break; + + case 'link_active': + if (!state.links.has(msg.id)) { + state.links.set(msg.id, { + nodeMAC: msg.node_mac, + peerMAC: msg.peer_mac, + lastFrame: Date.now(), + lastCSI: null + }); + updateLinkList(); + } + break; + + default: + // Ignore unknown types (forward-compatible) + } + } + + function handleBinaryFrame(buffer) { + const frame = parseCSIFrame(buffer); + if (!frame) return; + + const linkID = frame.linkID; + + // Update link state + let link = state.links.get(linkID); + if (!link) { + link = { + nodeMAC: frame.nodeMAC, + peerMAC: frame.peerMAC, + lastFrame: Date.now(), + lastCSI: null + }; + state.links.set(linkID, link); + updateLinkList(); + } else { + link.lastFrame = Date.now(); + } + + // Store CSI for chart rendering + link.lastCSI = frame; + + // Update chart if this is the selected link + if (state.selectedLinkID === linkID) { + const now = performance.now(); + if (now - state.lastChartUpdate >= CONFIG.chartUpdateMs) { + drawAmplitudeChart(frame); + state.lastChartUpdate = now; + } + } + } + + // ============================================ + // CSI Frame Parsing (matches Go binary format) + // ============================================ + function parseCSIFrame(buffer) { + const view = new DataView(buffer); + const bytes = new Uint8Array(buffer); + + if (bytes.length < 24) { + return null; // Header too short + } + + const nodeMAC = formatMAC(bytes, 0); + const peerMAC = formatMAC(bytes, 6); + const timestampUS = view.getBigUint64(12, true); // little-endian + const rssi = view.getInt8(20); + const noiseFloor = view.getInt8(21); + const channel = bytes[22]; + const nSub = bytes[23]; + + if (channel === 0 || channel > 14) { + return null; // Invalid channel + } + + const expectedLen = 24 + nSub * 2; + if (bytes.length !== expectedLen) { + return null; // Payload length mismatch + } + + // Extract I/Q pairs and compute amplitude + const subcarriers = []; + for (let i = 0; i < nSub; i++) { + const offset = 24 + i * 2; + const iVal = bytes[offset]; + const qVal = bytes[offset + 1]; + // Convert from unsigned to signed (JavaScript quirk) + const I = iVal > 127 ? iVal - 256 : iVal; + const Q = qVal > 127 ? qVal - 256 : qVal; + const amplitude = Math.sqrt(I * I + Q * Q); + subcarriers.push({ I, Q, amplitude }); + } + + return { + nodeMAC, + peerMAC, + linkID: `${nodeMAC}:${peerMAC}`, + timestampUS: Number(timestampUS), + rssi, + noiseFloor, + channel, + nSub, + subcarriers + }; + } + + function formatMAC(bytes, offset) { + const parts = []; + for (let i = 0; i < 6; i++) { + parts.push(bytes[offset + i].toString(16).padStart(2, '0').toUpperCase()); + } + return parts.join(':'); + } + + // ============================================ + // UI Updates + // ============================================ + function updateNodeList() { + const container = document.getElementById('node-list'); + document.getElementById('node-count').textContent = state.nodes.size; + + if (state.nodes.size === 0) { + container.innerHTML = '
No nodes connected
'; + return; + } + + let html = ''; + state.nodes.forEach((node, mac) => { + const isOnline = Date.now() - node.lastSeen < 30000; + html += ` +
+ ${mac} + + ${isOnline ? 'Online' : 'Offline'} + +
+ `; + }); + container.innerHTML = html; + } + + function updateLinkList() { + const container = document.getElementById('link-list'); + document.getElementById('link-count').textContent = state.links.size; + + if (state.links.size === 0) { + container.innerHTML = '
No links active
'; + return; + } + + let html = ''; + state.links.forEach((link, id) => { + const selected = state.selectedLinkID === id ? 'selected' : ''; + const shortID = id.split(':').map(p => p.split(':').slice(-1)[0]).join('→'); + html += ` +
+ ${shortID} + ${link.nSub || 64} sub +
+ `; + }); + container.innerHTML = html; + + // Add click handlers + container.querySelectorAll('.link-item').forEach(el => { + el.addEventListener('click', () => selectLink(el.dataset.linkId)); + }); + } + + function selectLink(linkID) { + state.selectedLinkID = linkID; + updateLinkList(); + + // Update chart title + document.querySelector('#chart-title .link-id').textContent = linkID || 'no link selected'; + + // Draw current data immediately if available + const link = state.links.get(linkID); + if (link && link.lastCSI) { + drawAmplitudeChart(link.lastCSI); + } + } + + // ============================================ + // Amplitude Chart (Canvas 2D) + // ============================================ + let chartCanvas, chartCtx; + + function initChart() { + chartCanvas = document.getElementById('amplitude-chart'); + chartCtx = chartCanvas.getContext('2d'); + + // Set canvas resolution + const rect = chartCanvas.getBoundingClientRect(); + chartCanvas.width = rect.width * window.devicePixelRatio; + chartCanvas.height = rect.height * window.devicePixelRatio; + chartCtx.scale(window.devicePixelRatio, window.devicePixelRatio); + + // Draw empty state + drawEmptyChart(); + } + + function drawEmptyChart() { + const width = chartCanvas.width / window.devicePixelRatio; + const height = chartCanvas.height / window.devicePixelRatio; + + chartCtx.fillStyle = '#1a1a2e'; + chartCtx.fillRect(0, 0, width, height); + + chartCtx.fillStyle = '#444'; + chartCtx.font = '12px sans-serif'; + chartCtx.textAlign = 'center'; + chartCtx.fillText('No data', width / 2, height / 2); + } + + function drawAmplitudeChart(frame) { + const width = chartCanvas.width / window.devicePixelRatio; + const height = chartCanvas.height / window.devicePixelRatio; + + // Clear + chartCtx.fillStyle = '#1a1a2e'; + chartCtx.fillRect(0, 0, width, height); + + const subcarriers = frame.subcarriers; + const nSub = subcarriers.length; + if (nSub === 0) return; + + const barWidth = width / nSub; + const padding = 1; + + // Find max amplitude for scaling + let maxAmp = 0; + subcarriers.forEach(s => { + if (s.amplitude > maxAmp) maxAmp = s.amplitude; + }); + if (maxAmp === 0) maxAmp = 1; + + // Draw bars + for (let i = 0; i < nSub; i++) { + const amp = subcarriers[i].amplitude; + const barHeight = (amp / maxAmp) * (height - 10); + const x = i * barWidth + padding; + const y = height - barHeight; + + // Gradient color based on amplitude + const intensity = amp / maxAmp; + const r = Math.floor(79 + intensity * (255 - 79)); + const g = Math.floor(195 - intensity * 100); + const b = Math.floor(247 - intensity * 150); + chartCtx.fillStyle = `rgb(${r}, ${g}, ${b})`; + + chartCtx.fillRect(x, y, barWidth - padding * 2, barHeight); + } + + // Draw channel/rssi info + chartCtx.fillStyle = '#666'; + chartCtx.font = '10px monospace'; + chartCtx.textAlign = 'left'; + chartCtx.fillText(`CH${frame.channel} RSSI:${frame.rssi}dBm`, 4, height - 4); + } + + // ============================================ + // Initialization + // ============================================ + function init() { + console.log('[Spaxel] Dashboard initializing...'); + + initScene(); + initChart(); + connectWebSocket(); + animate(); + + console.log('[Spaxel] Dashboard ready'); + } + + // Start when DOM is ready + if (document.readyState === 'loading') { + document.addEventListener('DOMContentLoaded', init); + } else { + init(); + } +})(); diff --git a/mothership/cmd/mothership/main.go b/mothership/cmd/mothership/main.go index 159f2d5..fcbafb7 100644 --- a/mothership/cmd/mothership/main.go +++ b/mothership/cmd/mothership/main.go @@ -9,12 +9,14 @@ import ( "net/http" "os" "os/signal" + "path/filepath" "syscall" "time" "github.com/go-chi/chi" "github.com/go-chi/chi/middleware" "github.com/hashicorp/mdns" + "github.com/spaxel/mothership/internal/dashboard" "github.com/spaxel/mothership/internal/ingestion" ) @@ -25,6 +27,7 @@ var version = "dev" type Config struct { BindAddr string DataDir string + StaticDir string MDNSName string MDNSEnabled bool LogLevel string @@ -33,7 +36,7 @@ type Config struct { func main() { cfg := parseConfig() log.Printf("[INFO] Spaxel mothership v%s starting", version) - log.Printf("[DEBUG] Config: bind=%s data=%s mdns=%s", cfg.BindAddr, cfg.DataDir, cfg.MDNSName) + log.Printf("[DEBUG] Config: bind=%s data=%s static=%s mdns=%s", cfg.BindAddr, cfg.DataDir, cfg.StaticDir, cfg.MDNSName) // Create context with cancellation for graceful shutdown _, cancel := context.WithCancel(context.Background()) @@ -59,6 +62,61 @@ func main() { ingestSrv := ingestion.NewServer() r.HandleFunc("/ws/node", ingestSrv.HandleNodeWS) + // Create dashboard hub and server + dashboardHub := dashboard.NewHub() + dashboardSrv := dashboard.NewServer(dashboardHub) + + // Connect ingestion to dashboard (for state queries) + dashboardHub.SetIngestionState(ingestSrv) + + // Start dashboard hub in background + go dashboardHub.Run() + + // Dashboard WebSocket endpoint + r.HandleFunc("/ws/dashboard", dashboardSrv.HandleDashboardWS) + + // Serve dashboard static files + staticDir := cfg.StaticDir + if staticDir == "" { + // Default: look for dashboard directory relative to binary or cwd + staticDir = findDashboardDir() + } + + if staticDir != "" { + // Check if directory exists + if _, err := os.Stat(staticDir); err == nil { + log.Printf("[INFO] Serving dashboard from %s", staticDir) + r.Get("/*", func(w http.ResponseWriter, r *http.Request) { + // Try to serve static file, fall back to index.html for SPA routing + path := filepath.Join(staticDir, r.URL.Path) + + // If path is a directory, serve index.html + if info, err := os.Stat(path); err == nil && info.IsDir() { + path = filepath.Join(path, "index.html") + } + + // If file exists, serve it + if _, err := os.Stat(path); err == nil { + http.ServeFile(w, r, path) + return + } + + // Fall back to index.html for SPA routing (except for /js/* paths) + if filepath.Ext(r.URL.Path) == "" { + http.ServeFile(w, r, filepath.Join(staticDir, "index.html")) + return + } + + // File not found + http.NotFound(w, r) + }) + } else { + log.Printf("[WARN] Dashboard directory not found: %s", staticDir) + } + } else { + log.Printf("[WARN] No dashboard directory found, static files not served") + } + // Start mDNS advertisement var mdnsServer *mdns.Server if cfg.MDNSEnabled { @@ -69,7 +127,7 @@ func main() { "", 8080, nil, - []string{"version=1", "ws=/ws/node", "api=/api"}, + []string{"version=1", "ws=/ws/node", "dashboard=/ws/dashboard"}, ) if err != nil { log.Printf("[ERROR] Failed to create mDNS service: %v", err) @@ -88,7 +146,7 @@ func main() { Addr: cfg.BindAddr, Handler: r, ReadTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, + WriteTimeout: 30 * time.Second, // Longer for WebSocket } // Run server in goroutine @@ -127,12 +185,14 @@ func main() { func parseConfig() Config { bindAddr := getEnv("SPAXEL_BIND_ADDR", "0.0.0.0:8080") dataDir := getEnv("SPAXEL_DATA_DIR", "/data") + staticDir := getEnv("SPAXEL_STATIC_DIR", "") mdnsName := getEnv("SPAXEL_MDNS_NAME", "spaxel") mdnsEnabled := getEnvBool("SPAXEL_MDNS_ENABLED", true) logLevel := getEnv("SPAXEL_LOG_LEVEL", "info") flag.StringVar(&bindAddr, "bind", bindAddr, "Listen address") flag.StringVar(&dataDir, "data", dataDir, "Data directory") + flag.StringVar(&staticDir, "static", staticDir, "Static files directory (dashboard)") flag.StringVar(&mdnsName, "mdns-name", mdnsName, "mDNS service name") flag.BoolVar(&mdnsEnabled, "mdns", mdnsEnabled, "Enable mDNS advertisement") flag.StringVar(&logLevel, "log-level", logLevel, "Log level (debug, info, warn, error)") @@ -141,6 +201,7 @@ func parseConfig() Config { return Config{ BindAddr: bindAddr, DataDir: dataDir, + StaticDir: staticDir, MDNSName: mdnsName, MDNSEnabled: mdnsEnabled, LogLevel: logLevel, @@ -160,3 +221,26 @@ func getEnvBool(key string, defaultVal bool) bool { } return defaultVal } + +// findDashboardDir attempts to locate the dashboard directory +func findDashboardDir() string { + // Try common locations + candidates := []string{ + "dashboard", // When running from repo root + "../dashboard", // When running from mothership/ + "../../dashboard", // When running from mothership/cmd/mothership/ + "/app/dashboard", // Docker container location + } + + for _, dir := range candidates { + absPath, err := filepath.Abs(dir) + if err != nil { + continue + } + if _, err := os.Stat(filepath.Join(absPath, "index.html")); err == nil { + return absPath + } + } + + return "" +} diff --git a/mothership/internal/dashboard/hub.go b/mothership/internal/dashboard/hub.go new file mode 100644 index 0000000..6c0c121 --- /dev/null +++ b/mothership/internal/dashboard/hub.go @@ -0,0 +1,235 @@ +// Package dashboard handles WebSocket connections from dashboard clients +// and broadcasts CSI data from the ingestion server. +package dashboard + +import ( + "encoding/json" + "log" + "sync" + "time" + + "github.com/spaxel/mothership/internal/ingestion" +) + +// Hub manages all dashboard client connections and broadcasts +type Hub struct { + mu sync.RWMutex + clients map[*Client]struct{} + broadcast chan []byte + register chan *Client + unregister chan *Client + + // Reference to ingestion server for state queries + ingestionState IngestionState +} + +// IngestionState is an interface to query node/link state from ingestion +type IngestionState interface { + GetConnectedNodesInfo() []ingestion.NodeInfo + GetAllLinksInfo() []ingestion.LinkInfo +} + +// Client represents a dashboard WebSocket client +type Client struct { + hub *Hub + send chan []byte +} + +// NewHub creates a new dashboard hub +func NewHub() *Hub { + return &Hub{ + clients: make(map[*Client]struct{}), + broadcast: make(chan []byte, 256), + register: make(chan *Client), + unregister: make(chan *Client), + } +} + +// SetIngestionState sets the ingestion state provider +func (h *Hub) SetIngestionState(state IngestionState) { + h.mu.Lock() + h.ingestionState = state + h.mu.Unlock() +} + +// Run starts the hub's main loop +func (h *Hub) Run() { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + + for { + select { + case client := <-h.register: + h.mu.Lock() + h.clients[client] = struct{}{} + h.mu.Unlock() + log.Printf("[INFO] Dashboard client connected (total: %d)", len(h.clients)) + + // Send initial state + h.sendInitialState(client) + + case client := <-h.unregister: + h.mu.Lock() + if _, ok := h.clients[client]; ok { + delete(h.clients, client) + close(client.send) + } + h.mu.Unlock() + log.Printf("[INFO] Dashboard client disconnected (total: %d)", len(h.clients)) + + case message := <-h.broadcast: + h.mu.RLock() + for client := range h.clients { + select { + case client.send <- message: + default: + // Client buffer full, skip + } + } + h.mu.RUnlock() + + case <-ticker.C: + // Periodic state broadcast + h.broadcastState() + } + } +} + +// Register registers a new dashboard client +func (h *Hub) Register(client *Client) { + h.register <- client +} + +// Unregister unregisters a dashboard client +func (h *Hub) Unregister(client *Client) { + h.unregister <- client +} + +// Broadcast sends a message to all connected dashboard clients +func (h *Hub) Broadcast(message []byte) { + select { + case h.broadcast <- message: + default: + // Channel full, drop message + } +} + +// BroadcastCSI broadcasts a CSI frame to all dashboard clients +func (h *Hub) BroadcastCSI(nodeMAC, peerMAC string, data []byte) { + // For now, just forward the raw binary frame + // Dashboard clients will parse it + h.Broadcast(data) +} + +// BroadcastNodeConnected notifies dashboards of a new node +func (h *Hub) BroadcastNodeConnected(mac, firmware, chip string) { + msg := map[string]interface{}{ + "type": "node_connected", + "mac": mac, + "firmware_version": firmware, + "chip": chip, + } + data, _ := json.Marshal(msg) + h.Broadcast(data) +} + +// BroadcastNodeDisconnected notifies dashboards of a node leaving +func (h *Hub) BroadcastNodeDisconnected(mac string) { + msg := map[string]interface{}{ + "type": "node_disconnected", + "mac": mac, + } + data, _ := json.Marshal(msg) + h.Broadcast(data) +} + +// BroadcastLinkActive notifies dashboards of an active link +func (h *Hub) BroadcastLinkActive(linkID, nodeMAC, peerMAC string) { + msg := map[string]interface{}{ + "type": "link_active", + "id": linkID, + "node_mac": nodeMAC, + "peer_mac": peerMAC, + } + data, _ := json.Marshal(msg) + h.Broadcast(data) +} + +// BroadcastLinkInactive notifies dashboards of an inactive link +func (h *Hub) BroadcastLinkInactive(linkID string) { + msg := map[string]interface{}{ + "type": "link_inactive", + "id": linkID, + } + data, _ := json.Marshal(msg) + h.Broadcast(data) +} + +func (h *Hub) sendInitialState(client *Client) { + h.mu.RLock() + state := h.ingestionState + h.mu.RUnlock() + + if state == nil { + return + } + + // Build state message + msg := map[string]interface{}{ + "type": "state", + } + + nodes := state.GetConnectedNodesInfo() + if nodes != nil { + msg["nodes"] = nodes + } + + links := state.GetAllLinksInfo() + if links != nil { + msg["links"] = links + } + + data, _ := json.Marshal(msg) + + select { + case client.send <- data: + default: + // Buffer full, skip + } +} + +func (h *Hub) broadcastState() { + h.mu.RLock() + state := h.ingestionState + clientCount := len(h.clients) + h.mu.RUnlock() + + if state == nil || clientCount == 0 { + return + } + + // Build state message + msg := map[string]interface{}{ + "type": "state", + } + + nodes := state.GetConnectedNodesInfo() + if nodes != nil { + msg["nodes"] = nodes + } + + links := state.GetAllLinksInfo() + if links != nil { + msg["links"] = links + } + + data, _ := json.Marshal(msg) + h.Broadcast(data) +} + +// ClientCount returns the number of connected dashboard clients +func (h *Hub) ClientCount() int { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.clients) +} diff --git a/mothership/internal/dashboard/hub_test.go b/mothership/internal/dashboard/hub_test.go new file mode 100644 index 0000000..ab002e8 --- /dev/null +++ b/mothership/internal/dashboard/hub_test.go @@ -0,0 +1,188 @@ +package dashboard + +import ( + "testing" + "time" + + "github.com/spaxel/mothership/internal/ingestion" +) + +func TestHub_RegisterUnregister(t *testing.T) { + hub := NewHub() + go hub.Run() + defer func() { + // Hub has no shutdown method in Phase 1, just let it run + }() + + client := &Client{ + hub: hub, + send: make(chan []byte, 10), + } + + // Register + hub.Register(client) + time.Sleep(10 * time.Millisecond) + + if hub.ClientCount() != 1 { + t.Errorf("expected 1 client, got %d", hub.ClientCount()) + } + + // Unregister + hub.Unregister(client) + time.Sleep(10 * time.Millisecond) + + if hub.ClientCount() != 0 { + t.Errorf("expected 0 clients, got %d", hub.ClientCount()) + } +} + +func TestHub_Broadcast(t *testing.T) { + hub := NewHub() + go hub.Run() + + client := &Client{ + hub: hub, + send: make(chan []byte, 10), + } + + hub.Register(client) + time.Sleep(10 * time.Millisecond) + + // Broadcast a message + testMsg := []byte(`{"type":"test"}`) + hub.Broadcast(testMsg) + + // Client should receive it + select { + case msg := <-client.send: + if string(msg) != string(testMsg) { + t.Errorf("expected %s, got %s", testMsg, msg) + } + case <-time.After(100 * time.Millisecond): + t.Error("expected to receive broadcast message") + } +} + +func TestHub_BroadcastCSI(t *testing.T) { + hub := NewHub() + go hub.Run() + + client := &Client{ + hub: hub, + send: make(chan []byte, 10), + } + + hub.Register(client) + time.Sleep(10 * time.Millisecond) + + // Broadcast CSI (raw binary data) + testData := []byte{0x01, 0x02, 0x03, 0x04} + hub.BroadcastCSI("AA:BB:CC:DD:EE:FF", "11:22:33:44:55:66", testData) + + select { + case msg := <-client.send: + if string(msg) != string(testData) { + t.Errorf("expected %v, got %v", testData, msg) + } + case <-time.After(100 * time.Millisecond): + t.Error("expected to receive CSI broadcast") + } +} + +func TestHub_NodeEvents(t *testing.T) { + hub := NewHub() + go hub.Run() + + client := &Client{ + hub: hub, + send: make(chan []byte, 10), + } + + hub.Register(client) + time.Sleep(10 * time.Millisecond) + + // Test node connected event + hub.BroadcastNodeConnected("AA:BB:CC:DD:EE:FF", "1.0.0", "ESP32-S3") + + msg := <-client.send + expected := `{"chip":"ESP32-S3","firmware_version":"1.0.0","mac":"AA:BB:CC:DD:EE:FF","type":"node_connected"}` + if string(msg) != expected { + t.Errorf("expected %s, got %s", expected, msg) + } + + // Test node disconnected event + hub.BroadcastNodeDisconnected("AA:BB:CC:DD:EE:FF") + + msg = <-client.send + expected = `{"mac":"AA:BB:CC:DD:EE:FF","type":"node_disconnected"}` + if string(msg) != expected { + t.Errorf("expected %s, got %s", expected, msg) + } +} + +func TestHub_LinkEvents(t *testing.T) { + hub := NewHub() + go hub.Run() + + client := &Client{ + hub: hub, + send: make(chan []byte, 10), + } + + hub.Register(client) + time.Sleep(10 * time.Millisecond) + + // Test link active event + hub.BroadcastLinkActive("AA:BB:CC:DD:EE:FF:11:22:33:44:55:66", "AA:BB:CC:DD:EE:FF", "11:22:33:44:55:66") + + msg := <-client.send + expected := `{"id":"AA:BB:CC:DD:EE:FF:11:22:33:44:55:66","node_mac":"AA:BB:CC:DD:EE:FF","peer_mac":"11:22:33:44:55:66","type":"link_active"}` + if string(msg) != expected { + t.Errorf("expected %s, got %s", expected, msg) + } +} + +// MockIngestionState implements IngestionState for testing +type MockIngestionState struct { + nodes []ingestion.NodeInfo + links []ingestion.LinkInfo +} + +func (m *MockIngestionState) GetConnectedNodesInfo() []ingestion.NodeInfo { + return m.nodes +} + +func (m *MockIngestionState) GetAllLinksInfo() []ingestion.LinkInfo { + return m.links +} + +func TestHub_InitialState(t *testing.T) { + hub := NewHub() + go hub.Run() + + // Set mock ingestion state + mock := &MockIngestionState{ + nodes: []ingestion.NodeInfo{ + {MAC: "AA:BB:CC:DD:EE:FF", FirmwareVersion: "1.0.0", Chip: "ESP32-S3"}, + }, + links: []ingestion.LinkInfo{ + {ID: "AA:BB:CC:DD:EE:FF:11:22:33:44:55:66", NodeMAC: "AA:BB:CC:DD:EE:FF", PeerMAC: "11:22:33:44:55:66"}, + }, + } + hub.SetIngestionState(mock) + + client := &Client{ + hub: hub, + send: make(chan []byte, 10), + } + + hub.Register(client) + time.Sleep(10 * time.Millisecond) + + // Should receive initial state + msg := <-client.send + // Just check it's a valid JSON with type "state" + if len(msg) == 0 || msg[0] != '{' { + t.Errorf("expected JSON state message, got %v", msg) + } +} diff --git a/mothership/internal/dashboard/server.go b/mothership/internal/dashboard/server.go new file mode 100644 index 0000000..57e3817 --- /dev/null +++ b/mothership/internal/dashboard/server.go @@ -0,0 +1,146 @@ +package dashboard + +import ( + "log" + "net/http" + "time" + + "github.com/gorilla/websocket" +) + +const ( + // Dashboard ping/pong timing + dashboardPingInterval = 30 * time.Second + dashboardReadDeadline = 60 * time.Second + + // Send buffer size per client + sendBufferSize = 1024 +) + +// Server handles WebSocket connections from dashboard clients +type Server struct { + hub *Hub + upgrader websocket.Upgrader +} + +// NewServer creates a new dashboard server +func NewServer(hub *Hub) *Server { + return &Server{ + hub: hub, + upgrader: websocket.Upgrader{ + // Allow all origins for development + CheckOrigin: func(r *http.Request) bool { + return true + }, + ReadBufferSize: 256, + WriteBufferSize: 4096, + }, + } +} + +// HandleDashboardWS handles WebSocket connections at /ws/dashboard +func (s *Server) HandleDashboardWS(w http.ResponseWriter, r *http.Request) { + // Upgrade HTTP connection to WebSocket + conn, err := s.upgrader.Upgrade(w, r, nil) + if err != nil { + log.Printf("[WARN] Dashboard WebSocket upgrade failed: %v", err) + return + } + + // Create client + client := &Client{ + hub: s.hub, + send: make(chan []byte, sendBufferSize), + } + + // Register with hub + s.hub.Register(client) + + // Start write goroutine + go s.writePump(conn, client) + + // Run read pump in this goroutine + s.readPump(conn, client) +} + +// readPump handles incoming messages from the dashboard client +func (s *Server) readPump(conn *websocket.Conn, client *Client) { + defer func() { + conn.Close() + s.hub.Unregister(client) + }() + + // Set read deadline + conn.SetReadDeadline(time.Now().Add(dashboardReadDeadline)) + + // Set pong handler to reset deadline + conn.SetPongHandler(func(string) error { + conn.SetReadDeadline(time.Now().Add(dashboardReadDeadline)) + return nil + }) + + for { + _, _, err := conn.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + log.Printf("[WARN] Dashboard read error: %v", err) + } + break + } + // Dashboard clients don't send meaningful messages in Phase 1 + // Just keep the connection alive + } +} + +// writePump handles outgoing messages to the dashboard client +func (s *Server) writePump(conn *websocket.Conn, client *Client) { + ticker := time.NewTicker(dashboardPingInterval) + defer func() { + ticker.Stop() + conn.Close() + }() + + for { + select { + case message, ok := <-client.send: + conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if !ok { + // Hub closed the channel + conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + + // Determine message type (binary vs text) + if len(message) > 0 && message[0] == '{' { + // Looks like JSON, send as text + err := conn.WriteMessage(websocket.TextMessage, message) + if err != nil { + log.Printf("[WARN] Dashboard write error: %v", err) + return + } + } else { + // Binary CSI frame + err := conn.WriteMessage(websocket.BinaryMessage, message) + if err != nil { + log.Printf("[WARN] Dashboard write error: %v", err) + return + } + } + + case <-ticker.C: + conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} + +// Hub returns the server's hub for external use +func (s *Server) Hub() *Hub { + return s.hub +} + +// Client represents a dashboard WebSocket client +// (redeclared here for documentation; defined in hub.go) +type dashboardClient = Client diff --git a/mothership/internal/ingestion/server.go b/mothership/internal/ingestion/server.go index 88367cc..2c315ee 100644 --- a/mothership/internal/ingestion/server.go +++ b/mothership/internal/ingestion/server.go @@ -11,6 +11,14 @@ import ( "github.com/gorilla/websocket" ) +// CSIBroadcaster is a callback for broadcasting CSI frames to dashboard +type CSIBroadcaster interface { + BroadcastCSI(nodeMAC, peerMAC string, data []byte) + BroadcastNodeConnected(mac, firmware, chip string) + BroadcastNodeDisconnected(mac string) + BroadcastLinkActive(linkID, nodeMAC, peerMAC string) +} + // Server manages WebSocket connections from ESP32 nodes type Server struct { mu sync.RWMutex @@ -25,6 +33,9 @@ type Server struct { // Shutdown state shutdown bool + + // Dashboard broadcaster (optional) + dashboardBroadcaster CSIBroadcaster } // NodeConnection tracks state for a connected node @@ -75,6 +86,13 @@ func NewServer() *Server { } } +// SetDashboardBroadcaster sets the callback for broadcasting CSI frames +func (s *Server) SetDashboardBroadcaster(broadcaster CSIBroadcaster) { + s.mu.Lock() + s.dashboardBroadcaster = broadcaster + s.mu.Unlock() +} + // HandleNodeWS handles WebSocket connections at /ws/node func (s *Server) HandleNodeWS(w http.ResponseWriter, r *http.Request) { // Upgrade HTTP connection to WebSocket @@ -127,11 +145,17 @@ func (s *Server) HandleNodeWS(w http.ResponseWriter, r *http.Request) { } s.connections[hello.MAC] = nc s.malformedCounts[hello.MAC] = &malformedCounter{} + broadcaster := s.dashboardBroadcaster s.mu.Unlock() log.Printf("[INFO] Node connected: MAC=%s firmware=%s chip=%s", hello.MAC, hello.FirmwareVersion, hello.Chip) + // Broadcast node connected to dashboard + if broadcaster != nil { + broadcaster.BroadcastNodeConnected(hello.MAC, hello.FirmwareVersion, hello.Chip) + } + // Send initial role and config s.sendRole(nc, "rx", "") s.sendConfig(nc, 20, 0, 0) // 20 Hz default @@ -150,8 +174,15 @@ func (s *Server) handleMessages(nc *NodeConnection) { s.mu.Lock() delete(s.connections, nc.MAC) delete(s.malformedCounts, nc.MAC) + broadcaster := s.dashboardBroadcaster s.mu.Unlock() + log.Printf("[INFO] Node disconnected: MAC=%s", nc.MAC) + + // Broadcast node disconnected to dashboard + if broadcaster != nil { + broadcaster.BroadcastNodeDisconnected(nc.MAC) + } }() for { @@ -189,14 +220,27 @@ func (s *Server) handleBinaryFrame(nc *NodeConnection, data []byte) { linkID := frame.LinkID() s.mu.Lock() ring, exists := s.links[linkID] + isNewLink := !exists if !exists { ring = NewRingBuffer() s.links[linkID] = ring } + broadcaster := s.dashboardBroadcaster s.mu.Unlock() // Push frame to ring buffer ring.Push(frame, time.Now()) + + // Broadcast to dashboard clients + if broadcaster != nil { + // Forward raw binary frame to dashboard + broadcaster.BroadcastCSI(frame.MACString(), frame.PeerMACString(), data) + + // Notify of new link + if isNewLink { + broadcaster.BroadcastLinkActive(linkID, frame.MACString(), frame.PeerMACString()) + } + } } // handleJSONMessage processes a JSON control message @@ -349,6 +393,56 @@ func (s *Server) GetConnectedNodes() []string { return macs } +// NodeInfo represents a connected node's state for dashboard +type NodeInfo struct { + MAC string `json:"mac"` + FirmwareVersion string `json:"firmware_version,omitempty"` + Chip string `json:"chip,omitempty"` +} + +// GetConnectedNodesInfo returns detailed info about connected nodes +func (s *Server) GetConnectedNodesInfo() []NodeInfo { + s.mu.RLock() + defer s.mu.RUnlock() + + nodes := make([]NodeInfo, 0, len(s.connections)) + for mac, nc := range s.connections { + info := NodeInfo{MAC: mac} + if nc.Hello != nil { + info.FirmwareVersion = nc.Hello.FirmwareVersion + info.Chip = nc.Hello.Chip + } + nodes = append(nodes, info) + } + return nodes +} + +// LinkInfo represents a link with its endpoints +type LinkInfo struct { + ID string `json:"id"` + NodeMAC string `json:"node_mac"` + PeerMAC string `json:"peer_mac"` +} + +// GetAllLinksInfo returns detailed info about all active links +func (s *Server) GetAllLinksInfo() []LinkInfo { + s.mu.RLock() + defer s.mu.RUnlock() + + links := make([]LinkInfo, 0, len(s.links)) + for linkID := range s.links { + // Parse linkID format "nodeMAC:peerMAC" (17 chars + 1 colon + 17 chars) + if len(linkID) >= 35 { + links = append(links, LinkInfo{ + ID: linkID, + NodeMAC: linkID[:17], + PeerMAC: linkID[18:], + }) + } + } + return links +} + // GetLinkBuffer returns the ring buffer for a specific link func (s *Server) GetLinkBuffer(nodeMAC, peerMAC string) *RingBuffer { linkID := nodeMAC + ":" + peerMAC