Initial

test_server.py

@@ -0,0 +1,122 @@
+"""Synthetic telemetry WS server for dashboard development.
+
+Emits packets in the exact shape the stratoflights backend broadcasts
+(see stratoflights_api/consumers.py), simulating a full flight profile:
+ascent at ~5 m/s to 30 km, burst, parachute descent. Point the dashboard at
+ws://localhost:8765/ and press "Подключиться".
+
+Usage:
+  pip install websockets
+  python3 test_server.py [--interval 1.0] [--speedup 20]
+
+--speedup compresses flight time: 20 means one wall-clock second advances the
+flight by 20 s (a 3 h flight replays in ~9 min).
+"""
+
+import argparse
+import asyncio
+import json
+import math
+import random
+import time
+
+import websockets
+
+BURST_ALT = 30000.0
+ASCENT_MS = 5.0
+GROUND_ALT = 200.0
+BASE_LAT, BASE_LON = 61.66, 129.38
+START_WALL = time.time()
+
+
+def sample(flight_s: float) -> dict:
+    ascent_end = (BURST_ALT - GROUND_ALT) / ASCENT_MS
+    if flight_s < ascent_end:
+        alt = GROUND_ALT + ASCENT_MS * flight_s
+        vs_phase = "ascent"
+    else:
+        # parachute descent, faster in thin air
+        t = flight_s - ascent_end
+        alt = BURST_ALT
+        dt, step = 0.0, 5.0
+        while dt < t and alt > GROUND_ALT:
+            rate = 5.0 + 45.0 * (alt / BURST_ALT) ** 2
+            alt -= rate * step
+            dt += step
+        alt = max(alt, GROUND_ALT)
+        vs_phase = "descent" if alt > GROUND_ALT else "landed"
+
+    alt_noise = alt + random.uniform(-8, 8)
+    pressure = 101325.0 * math.exp(-alt / 8400.0)
+    temp_ext = 15.0 - 6.5 * min(alt, 11000) / 1000.0 + (0.002 * max(alt - 20000, 0))
+    temp_int = max(temp_ext + 25.0, -15.0)
+    hum = max(5.0, 60.0 - alt / 400.0)
+    vbat = 3900 - flight_s * 0.06 + random.uniform(-5, 5)
+
+    flags = ["GPS_FIX", "GPS_TIME_VALID", "SD_OK", "MS5611_OK", "SHT45_OK", "PWR_GNSS", "PWR_SD"]
+    if temp_int < 0:
+        flags.append("HEATER1_ON")
+    if vbat < 3300:  # firmware BATT_LOW_MV (board.h)
+        flags.append("LOW_BATT")
+
+    utc = time.gmtime()
+    return {
+        "id": "test",
+        # flight-time timestamp so derived vertical speed is realistic under --speedup
+        "timestamp": int(START_WALL + flight_s),
+        "lat": BASE_LAT + flight_s * 1.2e-5 + random.uniform(-1e-5, 1e-5),
+        "lon": BASE_LON + flight_s * 6e-5 + random.uniform(-1e-5, 1e-5),
+        "alt": round(alt_noise, 1),
+        "payload": {
+            "utc": time.strftime("%H:%M:%S", utc) + ".00",
+            "callsign": "YKSA-1",
+            "uptime_s": int(flight_s) + 600,
+            "sats": random.randint(7, 12) if vs_phase != "landed" else random.randint(4, 8),
+            "vbat_mv": int(vbat),
+            "pressure_pa": int(pressure),
+            "baro_alt": round(alt_noise + random.uniform(-30, 30), 1) if alt < 25000 else None,
+            "baro_temp_c": round(temp_int, 2),
+            "humidity_pct": round(hum, 1),
+            "sht_temp_c": round(temp_int - 3.0, 2),
+            "tmp_ext_c": round(temp_ext, 2),
+            "flags": flags,
+        },
+        "raw_data": {},
+    }
+
+
+async def handler(ws, interval: float, speedup: float, start_s: float = 0.0):
+    print(f"client connected: {ws.remote_address}")
+    start = time.time()
+    n = 0
+    try:
+        while True:
+            flight_s = start_s + (time.time() - start) * speedup
+            packet = sample(flight_s)
+            await ws.send(json.dumps(packet))
+            n += 1
+            if n % 25 == 0:
+                print(f"  sent {n} packets, flight time {flight_s/60:.1f} min, alt {packet['alt']:.0f} m")
+            await asyncio.sleep(interval)
+    except websockets.ConnectionClosed:
+        print(f"client disconnected after {n} packets")
+
+
+async def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--port", type=int, default=8765)
+    ap.add_argument("--interval", type=float, default=1.0, help="seconds between packets")
+    ap.add_argument("--speedup", type=float, default=20.0, help="flight seconds per wall second")
+    ap.add_argument("--start", type=float, default=0.0,
+                    help="flight seconds to skip at connect (5900 ≈ just before burst)")
+    args = ap.parse_args()
+
+    async with websockets.serve(
+        lambda ws: handler(ws, args.interval, args.speedup, args.start), "localhost", args.port
+    ):
+        print(f"synthetic telemetry on ws://localhost:{args.port}/  (speedup x{args.speedup})")
+        await asyncio.Future()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())