diff --git a/app.py b/app.py index 307c11c..af7819a 100644 --- a/app.py +++ b/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 diff --git a/stream.py b/stream.py new file mode 100644 index 0000000..fe5a442 --- /dev/null +++ b/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 \ No newline at end of file