GHI #118 - Switch to SSE (server-sent events) for API data linkage [integration/state-estims-telemetry]

src/hooks/SSE/sseCommon.ts

@@ -0,0 +1,16 @@
+/** 
+ * Common type for SSE messages
+ * 
+ * Origin: What service did this come from?
+ * Version: What version of the API or endpoint pushed this?
+ * Timestamp: When was the data recorded?
+ * MessageType: What kind of payload is enclosed?
+ * Payload: Should resolve based on messageType parameter -- actual contents
+ */
+export interface SSEMessage {
+  origin: string;
+  version: string;
+  timestamp: number;
+  messageType: string;
+  payload: unknown;
+}

src/hooks/SSE/sseFlight.ts

@@ -0,0 +1,122 @@
+import { SSEMessage } from "./sseCommon";
+import { SDEC_BASE_URL } from "@/utils/api"
+import type { RawSensorPacket } from "../useSensorData";
+import type { WirelessBoardInfo } from "@/components/widgets/BoardStatusWidget";
+import { altitudeHandler } from "@/utils/units/units";
+
+/**
+ * Constants for supported endpoint versions. 
+ */
+const expectedOrigin = "SDEC-API";
+const expectedVersion = "1.0";
+
+/**
+ * Structure to house an API stream and callbacks to access the data.
+ */
+export interface FlightSseListener {
+    onDashboardData?: (packet: RawSensorPacket) => void;
+    onVehicleId?: (boardInfo: WirelessBoardInfo) => void;
+}
+
+/** 
+ * Structure housing preset calibration data 
+ */
+interface CalibData {
+    imu_offset: [number, number, number, number, number, number];
+    baro_preset: [number, number];
+    qfe_reference: number;
+    servo_preset: [number, number, number, number];
+}
+
+const listeners = new Set<FlightSseListener>();
+let eventSource: EventSource | null = null;
+
+/**
+ * Determine if the value passed in is of type Record<>
+ * 
+ * @param value The object to type-check.
+ * @returns TRUE if the value is a json-style record.
+ */
+function isRecord(value: unknown): value is Record<string, unknown> {
+    return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+/**
+ * Open the SSE stream
+ */
+function startConnection(): void {
+    if (eventSource || typeof window === "undefined") return;
+
+    eventSource = new EventSource(`${SDEC_BASE_URL}/stream`);
+
+    eventSource.onmessage = (event: MessageEvent) => {
+        try {
+            const data: SSEMessage = JSON.parse(event.data);
+            // Uncomment for verbose diagnostics
+            //console.log("New message:", data.messageType);
+
+            if (data.origin !== expectedOrigin) {
+                console.error("SDEC SSE origin does not match expected.");
+                return;
+            }
+
+            if (data.version !== expectedVersion) {
+                console.error("SDEC SSE stream version mismatch.");
+                return;
+            }
+
+            if (!isRecord(data.payload)) {
+                console.error("SDEC SSE payload is not an object -- discarding.");
+                return;
+            }
+
+            switch (data.messageType) {
+                case "DASHBOARD_DATA":
+                    listeners.forEach((listener) => listener.onDashboardData?.(data.payload as RawSensorPacket));
+                    break;
+                case "VEHICLE_ID":
+                    listeners.forEach((listener) => listener.onVehicleId?.(data.payload as WirelessBoardInfo));
+                    break;
+                case "CALIBRATION":
+                    updateCalibrationData(data.payload as unknown as CalibData);
+                    break;
+                default:
+                    console.error("Unknown message type received -- discarding.");
+                    break;
+            }
+        } catch (error) {
+            console.error("Failed to parse event data:", error);
+        }
+     };
+
+    eventSource.onerror = (error) => {
+        console.error("SSE connection error:", error);
+    };
+}
+
+/**
+ * Subscribe to the SSE stream.
+ * 
+ * @param listener A structure containing callbacks to access data from the stream. 
+ */
+export function subscribeToFlightSse(listener: FlightSseListener): () => void {
+    listeners.add(listener);
+    startConnection();
+
+    return () => {
+        listeners.delete(listener);
+        if (listeners.size === 0) {
+        eventSource?.close();
+        eventSource = null;
+        }
+    };
+}
+
+/**
+ * A helper to use incoming calibration data to set up the rest of the system.
+ * @param data The calibration data to use for the update.
+ */
+function updateCalibrationData(data: CalibData) {
+    /* Only one step for now: update QFE reference */
+    altitudeHandler.referenceElevation = data.qfe_reference;
+}

src/hooks/useBoardConnection.ts

@@ -1,6 +1,7 @@
-import { useState, useEffect, useRef, useCallback } from "react";
+import { useState, useEffect, useCallback } from "react";
 import { api } from "@/utils/api";
 import type { BoardInfo, BoardSummary, WirelessBoardInfo } from "@/components/widgets/BoardStatusWidget";
+import { subscribeToFlightSse } from "./SSE/sseFlight";
 
 export interface UseBoardConnectionResult {
   boards: BoardSummary[];
@@ -20,8 +21,6 @@ interface ControllerPacket {
   status: string;
 }
 
-const WIRELESS_POLL_MS = 1500;
-
 const EMPTY_BOARD_INFO: BoardInfo = {
   firmware: "",
   name: "",
@@ -32,7 +31,7 @@ export const useBoardConnection = (reset: boolean): UseBoardConnectionResult =>
   const [activeComPort, setActiveComPort] = useState<string | null>(null);
   const [boardInfo, setBoardInfo] = useState<BoardInfo>(EMPTY_BOARD_INFO);
   const [wirelessBoardInfo, setWirelessBoardInfo] = useState<WirelessBoardInfo | null>(null);
-  const wirelessIntervalRef = useRef<ReturnType<typeof setInterval> | null>(null);
+  const [sseEnabled, setSseEnabled] = useState(false);
 
   // Fetch COM ports on reset
   useEffect(() => {
@@ -58,37 +57,6 @@ export const useBoardConnection = (reset: boolean): UseBoardConnectionResult =>
       });
   }, [reset]);
 
-  const fetchWirelessInfo = useCallback(async () => {
-    try {
-      const response = await api.getWirelessInfo();
-
-      if (response.status === 204 || !response.data) {
-        setWirelessBoardInfo(null);
-        return;
-      }
-
-      setWirelessBoardInfo(response.data as WirelessBoardInfo);
-    } catch (error) {
-      console.error("Error fetching wireless info:", error);
-      setWirelessBoardInfo(null);
-    }
-  }, []);
-
-  const startWirelessPolling = useCallback(() => {
-    if (wirelessIntervalRef.current) return; // already polling
-
-    fetchWirelessInfo();
-    wirelessIntervalRef.current = setInterval(fetchWirelessInfo, WIRELESS_POLL_MS);
-  }, [fetchWirelessInfo]);
-
-  const stopWirelessPolling = useCallback(() => {
-    if (wirelessIntervalRef.current) {
-      api.stopDashboardDump();
-      clearInterval(wirelessIntervalRef.current);
-      wirelessIntervalRef.current = null;
-    }
-  }, []);
-
   const connectToBoard = useCallback(
     (name: string, onConnect: (success: boolean) => void) => {
       api
@@ -101,8 +69,8 @@ export const useBoardConnection = (reset: boolean): UseBoardConnectionResult =>
             name: packet.controller.name,
           });
 
-          startWirelessPolling();
           api.startDashboardDump();
+          setSseEnabled(true);
 
           onConnect(true);
         })
@@ -111,29 +79,29 @@ export const useBoardConnection = (reset: boolean): UseBoardConnectionResult =>
           onConnect(false);
         });
     },
-    [startWirelessPolling],
+    [],
   );
 
   const disconnectBoard = useCallback(
     (onDisconnect: (connected: boolean) => void) => {
       api
         .disconnectBoard()
         .then(() => {
-          stopWirelessPolling();
           api.stopDashboardDump();
+          setSseEnabled(false);
+          setWirelessBoardInfo(null);
           onDisconnect(false);
         })
         .catch(() => onDisconnect(true));
     },
-    [stopWirelessPolling],
+    [],
   );
 
-  // Cleanup polling on unmount
   useEffect(() => {
-    return () => {
-      stopWirelessPolling();
-    };
-  }, [stopWirelessPolling]);
+    if (!sseEnabled) return;
+
+    return subscribeToFlightSse({ onVehicleId: setWirelessBoardInfo });
+  }, [sseEnabled]);
 
   return {
     boards,

src/hooks/useSensorData.ts

@@ -2,6 +2,7 @@ import { useState, useEffect, useCallback } from "react";
 import { api } from "@/utils/api";
 import { MockFlight } from "@/utils/mock";
 import type { SensorData } from "@/components/widgets/SensorReadingWidget";
+import { subscribeToFlightSse } from "./SSE/sseFlight";
 
 /** Used for renderers to determine the fastest possible update rate */
 export const POLLING_INTERVAL_MS = 40;
@@ -10,7 +11,7 @@ export const POLLING_INTERVAL_MS = 40;
  * Raw sensor payload shape coming from the backend / mock flight source.
  * Field names mirror the device's wire format before conversion to the
  */
-interface RawSensorPacket {
+export interface RawSensorPacket {
   quat_w?: number;
   quat_x?: number;
   quat_y?: number;
@@ -95,11 +96,19 @@ export const useSensorData = (
   }, [mock, rowCount, onConnectionLost]);
 
   useEffect(() => {
-    if (!connected && !mock) return;
+    if (!mock) return;
 
     const interval = setInterval(fetchData, POLLING_INTERVAL_MS);
     return () => clearInterval(interval);
-  }, [connected, mock, fetchData]);
+  }, [mock, fetchData]);
+
+  useEffect(() => {
+    if (!connected || mock) return;
+
+    return subscribeToFlightSse({
+      onDashboardData: (packet) => setSensorData((previous) => parseSensorData(packet, previous)),
+    });
+  }, [connected, mock]);
 
   return sensorData;
 };

src/utils/api.ts

@@ -1,6 +1,6 @@
 import axios, { type AxiosResponse } from "axios";
 
-const BASE_URL = "http://127.0.0.1:5000";
+export const SDEC_BASE_URL = "http://127.0.0.1:5000";
 // const BASE_URL = 'http://localhost:5000';
 
 export interface ConnectBoardPacket {
@@ -44,26 +44,26 @@ export interface RawSensorPacket {
 
 export const api = {
   // Backend status
-  checkBackend: (): Promise<AxiosResponse<unknown>> => axios.get(`${BASE_URL}/`),
-  ping: (): Promise<AxiosResponse<unknown>> => axios.get(`${BASE_URL}/ping`),
+  checkBackend: (): Promise<AxiosResponse<unknown>> => axios.get(`${SDEC_BASE_URL}/`),
+  ping: (): Promise<AxiosResponse<unknown>> => axios.get(`${SDEC_BASE_URL}/ping`),
 
   // Board management
   connectBoard: (comport: string): Promise<AxiosResponse<ConnectBoardPacket>> => {
     console.log("Connecting to board with comport:", comport);
-    return axios.post(`${BASE_URL}/connect`, { comport: comport.toString() });
+    return axios.post(`${SDEC_BASE_URL}/connect`, { comport: comport.toString() });
   },
-  getComPorts: (): Promise<AxiosResponse<ComPortsMap>> => axios.get(`${BASE_URL}/comports`),
+  getComPorts: (): Promise<AxiosResponse<ComPortsMap>> => axios.get(`${SDEC_BASE_URL}/comports`),
   getActiveComPort: (): Promise<AxiosResponse<string | null>> =>
-    axios.get(`${BASE_URL}/comports/active`),
-  disconnectBoard: (): Promise<AxiosResponse<unknown>> => axios.get(`${BASE_URL}/disconnect`),
+    axios.get(`${SDEC_BASE_URL}/comports/active`),
+  disconnectBoard: (): Promise<AxiosResponse<unknown>> => axios.get(`${SDEC_BASE_URL}/disconnect`),
   getWirelessInfo: (): Promise<AxiosResponse<WirelessInfoResponse | null>> =>
-    axios.get(`${BASE_URL}/wireless-stats`),
+    axios.get(`${SDEC_BASE_URL}/wireless-stats`),
 
   // Sensor data
   startDashboardDump: (): Promise<AxiosResponse<unknown>> =>
-    axios.post(`${BASE_URL}/dashboard-dump`, { start: true }),
+    axios.post(`${SDEC_BASE_URL}/dashboard-dump`, { start: true }),
   stopDashboardDump: (): Promise<AxiosResponse<unknown>> =>
-    axios.post(`${BASE_URL}/dashboard-dump`, { stop: true }),
+    axios.post(`${SDEC_BASE_URL}/dashboard-dump`, { stop: true }),
   getSensorData: (): Promise<AxiosResponse<RawSensorPacket>> =>
-    axios.get(`${BASE_URL}/dashboard-dump`),
+    axios.get(`${SDEC_BASE_URL}/dashboard-dump`),
 };