From 05c954a72d31df925c33b3c604b5b14683f001c1 Mon Sep 17 00:00:00 2001 From: etsells Date: Sat, 18 Jul 2026 13:53:19 -0700 Subject: [PATCH 1/6] Implement SSE based stream of data to clients (initial/pre-integration) --- app.py | 32 +++++++++++++++++++++++++++++-- stream.py | 57 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 2 deletions(-) create mode 100644 stream.py diff --git a/app.py b/app.py index 307c11c..4753959 100644 --- a/app.py +++ b/app.py @@ -5,11 +5,12 @@ 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 threads import poll_dashboard_dump +from stream import Stream +from threads import poll_dashboard_dump, sse_stream_callback 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,31 @@ 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" + ) + @app.route("/dashboard-dump", methods=["GET", "POST"]) def dashboard_dump(): global dashboard_dump_thread diff --git a/stream.py b/stream.py new file mode 100644 index 0000000..901fefe --- /dev/null +++ b/stream.py @@ -0,0 +1,57 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2025 Sun Devil Rocketry + +import threading +import time + +from typing import Dict, Any +from util import make_safe_number + +# CONSTANTS +ORIGIN_STRING = "SDEC-API" +STREAM_VERSION_STRING = "1.0" + + +class Stream: + def __init__(self): + self.signal_broadcast_event = threading.Event() + self.stream_mutex = threading.Lock() + 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_mutex: + self.data = { + "origin": ORIGIN_STRING, + "version": STREAM_VERSION_STRING, + "timestamp": timestamp, + "messageType": messageType, + "payload": msg + } + self.sequence_number += 1 + self.signal_broadcast_event() # Only signal the events after releasing the mutex + + + def sse_broadcast_callback(self): + """ + Yields messages for the client with this request open. + """ + last_seq_num = 0 # scoped to this instance + + # Send a message immediately on connect if one exists + with self.stream_mutex: + if self.data is not None: + last_seq_num = self.sequence_number + yield self.data + + # Wait for new messages to be put + while not self.terminate: + self.signal_broadcast_event.wait() + with self.stream_mutex: + if self.data is not None and self.sequence_number != last_seq_num: + last_seq_num = self.sequence_number + yield self.data \ No newline at end of file From 9561e4d5a1e8db403e23676097f370319b5cc096 Mon Sep 17 00:00:00 2001 From: etsells Date: Sat, 18 Jul 2026 14:28:36 -0700 Subject: [PATCH 2/6] Fix SSE producer --- app.py | 4 ++-- stream.py | 6 ++++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/app.py b/app.py index 4753959..4990eb7 100644 --- a/app.py +++ b/app.py @@ -10,7 +10,7 @@ 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, sse_stream_callback +from threads import poll_dashboard_dump from typing import List, Dict from util import make_safe_number @@ -174,7 +174,7 @@ def stream(): not require upversioning. """ return Response( - stream_with_context(stream_obj.sse_broadcast_callback), + stream_with_context(stream_obj.sse_broadcast_callback()), mimetype="text/event-stream" ) diff --git a/stream.py b/stream.py index 901fefe..1059855 100644 --- a/stream.py +++ b/stream.py @@ -25,15 +25,17 @@ def sse_put_callback(self, messageType: str, timestamp: float, msg: dict[str, An Set up a message for all clients to broadcast. """ with self.stream_mutex: - self.data = { + response = { "origin": ORIGIN_STRING, "version": STREAM_VERSION_STRING, "timestamp": timestamp, "messageType": messageType, "payload": msg } + self.data = str("data: " + str(response) + "\n\n").encode("utf-8") self.sequence_number += 1 - self.signal_broadcast_event() # Only signal the events after releasing the mutex + self.signal_broadcast_event.set() # Only signal the events after releasing the mutex + self.signal_broadcast_event.clear() def sse_broadcast_callback(self): From 7b39ed3b263cba15630f8dcafcee6e55f0dc83a5 Mon Sep 17 00:00:00 2001 From: etsells <147210601+ETSells@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:35:38 -0700 Subject: [PATCH 3/6] Addressing: Add headers to the SSE event stream Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- app.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/app.py b/app.py index 4990eb7..af7819a 100644 --- a/app.py +++ b/app.py @@ -175,7 +175,12 @@ def stream(): """ return Response( stream_with_context(stream_obj.sse_broadcast_callback()), - mimetype="text/event-stream" + mimetype="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no" + } ) @app.route("/dashboard-dump", methods=["GET", "POST"]) From a3de9be1610b7a60ec3a8b3c8692d1ca7c80bfaa Mon Sep 17 00:00:00 2001 From: etsells <147210601+ETSells@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:38:56 -0700 Subject: [PATCH 4/6] Addressing: Switch to a single Condition object rather than discrete mutexes and events Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- stream.py | 36 ++++++++++++++++++++++-------------- 1 file changed, 22 insertions(+), 14 deletions(-) diff --git a/stream.py b/stream.py index 1059855..ebd4d8d 100644 --- a/stream.py +++ b/stream.py @@ -13,9 +13,8 @@ class Stream: - def __init__(self): - self.signal_broadcast_event = threading.Event() - self.stream_mutex = threading.Lock() + def __init__(self) -> None: + self.stream_cond = threading.Condition() self.sequence_number = 0 self.terminate = False self.data = None @@ -24,7 +23,7 @@ def sse_put_callback(self, messageType: str, timestamp: float, msg: dict[str, An """ Set up a message for all clients to broadcast. """ - with self.stream_mutex: + with self.stream_cond: response = { "origin": ORIGIN_STRING, "version": STREAM_VERSION_STRING, @@ -32,10 +31,9 @@ def sse_put_callback(self, messageType: str, timestamp: float, msg: dict[str, An "messageType": messageType, "payload": msg } - self.data = str("data: " + str(response) + "\n\n").encode("utf-8") + self.data = f"data: {json.dumps(response)}\n\n".encode("utf-8") self.sequence_number += 1 - self.signal_broadcast_event.set() # Only signal the events after releasing the mutex - self.signal_broadcast_event.clear() + self.stream_cond.notify_all() def sse_broadcast_callback(self): @@ -45,15 +43,25 @@ def sse_broadcast_callback(self): last_seq_num = 0 # scoped to this instance # Send a message immediately on connect if one exists - with self.stream_mutex: + with self.stream_cond: if self.data is not None: last_seq_num = self.sequence_number - yield self.data + 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: - self.signal_broadcast_event.wait() - with self.stream_mutex: - if self.data is not None and self.sequence_number != last_seq_num: - last_seq_num = self.sequence_number - yield self.data \ No newline at end of file + 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() + if self.terminate: + break + last_seq_num = self.sequence_number + data_to_yield = self.data + + yield data_to_yield \ No newline at end of file From bec76e789c4b156c3e970b61b5afda9776f50392 Mon Sep 17 00:00:00 2001 From: ETSsound <147210601+ETSsound@users.noreply.github.com> Date: Sat, 18 Jul 2026 17:18:21 -0700 Subject: [PATCH 5/6] Fix: Implement a rate limiter for the broadcast function to slow down the rate of request broadcasts --- stream.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/stream.py b/stream.py index ebd4d8d..5e63453 100644 --- a/stream.py +++ b/stream.py @@ -3,6 +3,7 @@ import threading import time +import json from typing import Dict, Any from util import make_safe_number @@ -10,6 +11,8 @@ # CONSTANTS ORIGIN_STRING = "SDEC-API" STREAM_VERSION_STRING = "1.0" +RATE_LIMITER_HZ = 30 +RATE_LIMITER_PERIOD = 1.0 / RATE_LIMITER_HZ class Stream: @@ -23,6 +26,7 @@ def sse_put_callback(self, messageType: str, timestamp: float, msg: dict[str, An """ Set up a message for all clients to broadcast. """ + with self.stream_cond: response = { "origin": ORIGIN_STRING, @@ -41,6 +45,7 @@ def sse_broadcast_callback(self): Yields messages for the client with this request open. """ last_seq_num = 0 # scoped to this instance + last_time = 0 # Send a message immediately on connect if one exists with self.stream_cond: @@ -59,6 +64,16 @@ def sse_broadcast_callback(self): # 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.time() + if RATE_LIMITER_PERIOD + last_time < curr_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() + time.sleep( RATE_LIMITER_PERIOD + last_time - curr_time ) + self.stream_cond.acquire() + # end the broadcast entirely if the termination flag is if self.terminate: break last_seq_num = self.sequence_number From 97c5eb3b73295686175f5cf4c3986115257913ce Mon Sep 17 00:00:00 2001 From: etsells <147210601+ETSells@users.noreply.github.com> Date: Sat, 18 Jul 2026 17:25:16 -0700 Subject: [PATCH 6/6] Addressing: Bad timeout logic & switching to monotonic timer Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- stream.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/stream.py b/stream.py index 5e63453..fe5a442 100644 --- a/stream.py +++ b/stream.py @@ -45,7 +45,7 @@ def sse_broadcast_callback(self): Yields messages for the client with this request open. """ last_seq_num = 0 # scoped to this instance - last_time = 0 + last_time = time.monotonic() # Send a message immediately on connect if one exists with self.stream_cond: @@ -66,17 +66,20 @@ def sse_broadcast_callback(self): 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.time() - if RATE_LIMITER_PERIOD + last_time < curr_time: + 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() - time.sleep( RATE_LIMITER_PERIOD + last_time - curr_time ) - self.stream_cond.acquire() - # end the broadcast entirely if the termination flag is + 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 \ No newline at end of file