app.py

"""
Flask web server to render telemetry overlay.
"""

import subprocess
import threading
from typing import Any

from flask import Flask

import altos
from config import ALTOS_COMMAND

app = Flask(__name__)


@app.route("/")
def root():
    return app.send_static_file("main.html")


@app.route("/data/<key>")
def data(key):
    p = TelemetryWatcher.last_packet
    d = p["sensor"]
    d["tick"] = p["preamble"]["tick"]
    if TelemetryWatcher.launch_at > -1:
        met_tick = d["tick"] - TelemetryWatcher.launch_at
        d["met"] = (
            f"T+{(met_tick // 6000):02d}:{met_tick // 100 % 60:02d}.{met_tick % 100:02d}"
        )
    else:
        d["met"] = "T+00:00.000"

    state_map = {
        0: "On Pad",
        1: "On Pad",
        2: "On Pad",
        3: "Boost Phase",
        4: "Fast Phase",
        5: "Coast Phase",
        6: "Drogue Deployed",
        7: "Main Deployed",
        8: "Landed",
    }
    d["state_c"] = state_map.get(d["state"], "Unknown State")
    return str(d.get(key, "ERROR"))


class TelemetryWatcher(threading.Thread):
    last_packet: dict[str, Any]
    launch_at: int = -1

    def run(self):
        p = subprocess.Popen(ALTOS_COMMAND, stdout=subprocess.PIPE)
        out = p.stdout
        if out is None:
            raise IOError

        while True:
            line = out.readline().decode(encoding="utf-8")
            try:
                packet = altos.parse_serial_line(line)
                if packet["preamble"]["type"] == 0x0A:  # sensor data
                    TelemetryWatcher.last_packet = packet
                    if (
                        packet["sensor"]["state"] > 0
                        and TelemetryWatcher.launch_at == -1
                    ):
                        TelemetryWatcher.launch_at = packet["preamble"]["tick"]
            except ValueError:
                pass


if __name__ == "__main__":
    watcher_thread = TelemetryWatcher()
    watcher_thread.start()
    app.run()