API: Server-Side Event Stream

app.py

@@ -5,10 +5,11 @@
 import threading
 import json
 
-from flask import Flask, request, Response, jsonify
+from flask import Flask, request, Response, jsonify, stream_with_context
 from flask_cors import CORS
 from hardware import serial_connection, sensor_sentry, serial_lock, background_lifecycle_lock, firmware # Objects from hardware.py
 from serial import SerialException
+from stream import Stream
 from threads import poll_dashboard_dump
 from typing import List, Dict
 from util import make_safe_number
@@ -29,7 +30,9 @@
 
 # Globals for polling dashboard dump
 stop_event = threading.Event()
+stream_obj = Stream()
 telemetry_obj = Telemetry()
+telemetry_obj.register_rx_callback(stream_obj.sse_put_callback)
 dashboard_dump_thread = threading.Thread(
     target=poll_dashboard_dump,
     args=(serial_connection, stop_event, telemetry_obj),
@@ -150,6 +153,36 @@ def wireless_stats():
     else:
         return Response("No wireless target available.", status=204)
 
+@app.route("/stream", methods=["GET"])
+def stream():
+    """
+    Subscribes to a server-sent event stream that is populated by all 
+    incoming messages.
+
+    Works for both the serial (USB) and over-the-air (LoRa) paths.
+
+    Integration Concerns: 
+        SDEC-API <-> SDEC - App initialization needs to provide a 
+        message_received callback to the telemetry object. SDEC needs
+        to invoke this callback upon receiving a new message. Null
+        or uninitialized callbacks must be handled.
+
+        SDEC-API <-> Client - Dashboard Dump must be initialized with
+        a POST request prior to subscribing to the stream, else no events
+        will be sent. The version number shall be incremented every time
+        an **existing** message type is changed. New message types do
+        not require upversioning.
+    """
+    return Response(
+        stream_with_context(stream_obj.sse_broadcast_callback()),
+        mimetype="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no"
+        }
+    )
+
 @app.route("/dashboard-dump", methods=["GET", "POST"])
 def dashboard_dump():
     global dashboard_dump_thread

stream.py

@@ -0,0 +1,85 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2025 Sun Devil Rocketry
+
+import threading
+import time
+import json
+
+from typing import Dict, Any
+from util import make_safe_number
+
+# CONSTANTS
+ORIGIN_STRING = "SDEC-API"
+STREAM_VERSION_STRING = "1.0"
+RATE_LIMITER_HZ = 30
+RATE_LIMITER_PERIOD = 1.0 / RATE_LIMITER_HZ
+
+
+class Stream:
+    def __init__(self) -> None:
+        self.stream_cond = threading.Condition()
+        self.sequence_number = 0
+        self.terminate = False
+        self.data = None
+
+    def sse_put_callback(self, messageType: str, timestamp: float, msg: dict[str, Any]):
+        """
+        Set up a message for all clients to broadcast.
+        """
+
+        with self.stream_cond:
+            response = {
+                "origin": ORIGIN_STRING,
+                "version": STREAM_VERSION_STRING,
+                "timestamp": timestamp,
+                "messageType": messageType,
+                "payload": msg
+            }
+            self.data = f"data: {json.dumps(response)}\n\n".encode("utf-8")
+            self.sequence_number += 1
+            self.stream_cond.notify_all()
+
+
+    def sse_broadcast_callback(self):
+        """
+        Yields messages for the client with this request open.
+        """
+        last_seq_num = 0 # scoped to this instance
+        last_time = time.monotonic()
+
+        # Send a message immediately on connect if one exists
+        with self.stream_cond:
+            if self.data is not None:
+                last_seq_num = self.sequence_number
+                data_to_yield = self.data
+            else:
+                data_to_yield = None
+
+        if data_to_yield is not None:
+            yield data_to_yield
+
+        # Wait for new messages to be put
+        while not self.terminate:
+            with self.stream_cond:
+                # Wait for sequence_number to change to avoid spurious wakeups
+                while self.sequence_number == last_seq_num and not self.terminate:
+                    self.stream_cond.wait()
+                # delay the broadcast (yielding to other threads) if rate limiter
+                # indicates that messages are coming in too fast
+                curr_time = time.monotonic()
+                if curr_time < RATE_LIMITER_PERIOD + last_time:
+                    # Release the lock if we have to wait for the timeout,
+                    # then reacquire once we're back on the target rate.
+                    self.stream_cond.release()
+                    try:
+                        time.sleep( RATE_LIMITER_PERIOD + last_time - curr_time )
+                    finally:
+                        self.stream_cond.acquire()
+                # end the broadcast entirely if the termination flag is set
+                if self.terminate:
+                    break
+                last_seq_num = self.sequence_number
+                data_to_yield = self.data
+                last_time = time.monotonic()
+
+            yield data_to_yield