أسطول طائرات من مصنعين مختلفين على لوحة مراقبة واحدة

# Drone operations lab: 2 x PX4 + 1 x ArduPilot in SITL, a vendor X dock over MQTT, an edge gateway,
# NATS JetStream, TimescaleDB, a core service with a live map, Grafana and MediaMTX.
#   docker compose -p lab-dops up -d --build
name: lab-dops

x-drones: &drones '{"10.251.31.11": "px4-a", "10.251.31.12": "px4-b", "10.251.31.13": "ap-c"}'

networks:
  air:      # the radio side: autopilots and the gateway
    name: lab-dops-air
    ipam: { config: [{ subnet: 10.251.31.0/24 }] }
  cloud:    # the command-center side
    name: lab-dops-cloud
    ipam: { config: [{ subnet: 10.251.30.0/24 }] }

volumes:
  lab-dops-nats: { name: lab-dops-nats }
  lab-dops-db: { name: lab-dops-db }
  lab-dops-buf: { name: lab-dops-buf }
  lab-dops-data: { name: lab-dops-data }
  lab-dops-rec: { name: lab-dops-rec }

services:
  px4a:
    image: jonasvautherin/px4-gazebo-headless:1.17.0
    container_name: lab-dops-px4a
    init: true                    # an init process, so the container stops cleanly (PX4 leaves zombies otherwise)
    command: ["10.251.31.20", "10.251.31.20"]   # API link (UDP 14540) and GCS link (UDP 14550)
    stdin_open: true
    tty: true
    networks: { air: { ipv4_address: 10.251.31.11 } }

  px4b:
    image: jonasvautherin/px4-gazebo-headless:1.17.0
    container_name: lab-dops-px4b
    init: true
    profiles: [second-px4]        # a second Gazebo starved this 8-core machine; start it with --profile second-px4
    command: ["10.251.31.20", "10.251.31.20"]   # API link (UDP 14540) and GCS link (UDP 14550)
    stdin_open: true
    tty: true
    networks: { air: { ipv4_address: 10.251.31.12 } }

  ap:
    build: { context: ., dockerfile: Dockerfile.ardupilot }
    image: lab-dops-ardupilot:1
    container_name: lab-dops-ap
    command: ["-I0", "--home", "47.3972,8.5432,488,0",
              "--serial0", "udpclient:10.251.31.20:14560", "--serial1", "udpclient:10.251.31.20:14561"]
    networks: { air: { ipv4_address: 10.251.31.13 } }

  gw:
    build: { context: ., dockerfile: Dockerfile.py }
    image: lab-dops-py:1
    container_name: lab-dops-gw
    command: ["python", "-u", "gateway.py"]
    environment: { DRONES: *drones, NATS_URL: "nats://lab-dops-nats:4222" }
    cap_add: [NET_ADMIN]          # only for the tc netem break step
    volumes: [lab-dops-buf:/buf]
    networks:
      air: { ipv4_address: 10.251.31.20 }
      cloud: { ipv4_address: 10.251.30.20 }
    depends_on: [nats]

  pilot:
    image: lab-dops-py:1
    container_name: lab-dops-pilot
    command: ["python", "-u", "pilot.py"]
    environment: { DRONES: *drones }
    network_mode: "service:gw"    # same address as the gateway: PX4 sends its API link there on UDP 14540
    depends_on: [gw]

  vendorx:
    image: lab-dops-py:1
    container_name: lab-dops-vendorx
    command: ["python", "-u", "vendorx.py"]
    environment: { FIRMWARE: "${FIRMWARE:-1}" }
    networks: [cloud]
    depends_on: [nats]

  nats:
    image: nats:2.15.0-alpine
    container_name: lab-dops-nats
    command: ["-c", "/etc/nats/nats.conf"]
    volumes: ["./nats.conf:/etc/nats/nats.conf:ro", "lab-dops-nats:/data"]
    ports: ["127.0.0.1:38222:8222"]
    networks: { cloud: { ipv4_address: 10.251.30.10 } }

  db:
    image: timescale/timescaledb:2.30.2-pg18
    container_name: lab-dops-db
    environment: { POSTGRES_PASSWORD: lab, POSTGRES_DB: fleet }
    volumes: [lab-dops-db:/var/lib/postgresql]
    ports: ["127.0.0.1:35432:5432"]
    networks: [cloud]

  core:
    image: lab-dops-py:1
    container_name: lab-dops-core
    command: ["python", "-u", "core.py"]
    environment: { VX_ADAPTER: "${VX_ADAPTER:-1}" }
    volumes: [lab-dops-data:/data]
    ports: ["127.0.0.1:38080:8080"]
    networks: [cloud]
    depends_on: [nats, db]

  grafana:
    image: grafana/grafana:13.2.3
    container_name: lab-dops-grafana
    environment:
      GF_AUTH_ANONYMOUS_ENABLED: "true"
      GF_AUTH_ANONYMOUS_ORG_ROLE: Viewer
      GF_SECURITY_ADMIN_PASSWORD: lab-only-password
    volumes: ["./grafana/provisioning:/etc/grafana/provisioning:ro", "./grafana/dashboards:/var/lib/grafana/dashboards:ro"]
    ports: ["127.0.0.1:33000:3000"]
    networks: [cloud]

  mtx:
    image: bluenviron/mediamtx:1.21.2
    container_name: lab-dops-mtx
    volumes: ["./mediamtx.yml:/mediamtx.yml:ro", "lab-dops-rec:/recordings"]
    ports: ["127.0.0.1:38889:8889", "127.0.0.1:38189:38189/udp"]
    networks: { cloud: { ipv4_address: 10.251.30.30 } }

"""Edge gateway: MAVLink in (PX4 and ArduPilot), normalized telemetry out to NATS JetStream.

  - one MAVLink parser per source address, because two PX4 vehicles can both say "I am system 1"
  - event time from the autopilot clock (SYSTEM_TIME), not the time the packet arrived
  - 2 messages per second per drone to the cloud (the full rate stays in the flight log)
  - store-and-forward: when the cloud link is down, messages wait in a SQLite outbox on disk
    and are sent later in order, each with a Nats-Msg-Id so a resend is not stored twice
"""
import asyncio
import json
import os
import socket
import sqlite3
import time

import nats
from pymavlink.dialects.v20 import ardupilotmega as mavlink2

from schema import iso, make

NATS_URL = os.environ.get("NATS_URL", "nats://lab-dops-nats:4222")
# who is who: the radio link (here a UDP port and a source address) decides the identity, not sysid
DRONES = json.loads(os.environ.get("DRONES", "{}"))     # {"10.251.31.11": "px4-a", ...}
RATE_HZ = 2.0
PX4_MODES = {(1, 0): "MANUAL", (2, 0): "ALTCTL", (3, 0): "POSCTL", (4, 2): "TAKEOFF", (4, 3): "HOLD",
             (4, 4): "MISSION", (4, 5): "RTL", (4, 6): "LAND", (6, 0): "OFFBOARD"}
AP_MODES = {0: "STABILIZE", 2: "ALT_HOLD", 3: "AUTO", 4: "GUIDED", 5: "LOITER", 6: "RTL", 9: "LAND", 16: "POSHOLD"}


def log(msg):
    print(f"{time.strftime('%H:%M:%S')} gateway {msg}", flush=True)


class Drone:
    def __init__(self, drone_id, addr, sock):
        self.id, self.addr, self.sock = drone_id, addr, sock
        self.mav = mavlink2.MAVLink(None, srcSystem=255, srcComponent=190)
        self.st = {}
        self.boot_offset = None        # unix seconds at boot = time_unix - time_boot
        self.vendor = None
        self.seq = 0
        self.last_pub = 0.0
        self.requested = False
        self.types = {}
        self.last_unix = None

    def send(self, m):
        self.sock.sendto(m.pack(self.mav), self.addr)


class Gateway:
    def __init__(self):
        self.drones = {}
        self.out = asyncio.Queue()
        self.db = sqlite3.connect("/buf/outbox.db")
        self.db.execute("CREATE TABLE IF NOT EXISTS outbox (id INTEGER PRIMARY KEY, subject TEXT, msg_id TEXT, body BLOB)")
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.commit()
        self.pending = []               # rows for the outbox, written to disk together twice a second
        self.down_until = 0.0           # simple circuit breaker after a failed publish
        self.stats = {"sent": 0, "buffered": 0, "flushed": 0}

    # ---------- MAVLink side ----------
    def on_mavlink(self, sock, data, addr):
        d = self.drones.get(addr)
        if d is None:
            if addr[0] not in DRONES:
                return
            d = self.drones[addr] = Drone(DRONES[addr[0]], addr, sock)
            log(f"new link {addr[0]}:{addr[1]} -> {d.id}")
        try:
            msgs = d.mav.parse_buffer(data) or []
        except mavlink2.MAVError:
            return
        for m in msgs:
            self.on_msg(d, m)

    def on_msg(self, d, m):
        t = m.get_type()
        d.types[t] = d.types.get(t, 0) + 1
        s = d.st
        if t == "HEARTBEAT" and m.type != mavlink2.MAV_TYPE_GCS:
            d.vendor = "px4" if m.autopilot == mavlink2.MAV_AUTOPILOT_PX4 else "ardupilot"
            s["armed"] = bool(m.base_mode & mavlink2.MAV_MODE_FLAG_SAFETY_ARMED)
            if not d.requested:
                self.request_streams(d)
            if d.vendor == "px4":
                main, sub = (m.custom_mode >> 16) & 0xFF, (m.custom_mode >> 24) & 0xFF
                s["mode"] = PX4_MODES.get((main, sub), f"{main}.{sub}")
            else:
                s["mode"] = AP_MODES.get(m.custom_mode, str(m.custom_mode))
        elif t == "SYSTEM_TIME":
            d.last_unix = m.time_unix_usec
            if m.time_unix_usec < 1.6e15:        # 0 (or 1970) until the autopilot has a real date from GNSS
                return
            d.boot_offset = m.time_unix_usec / 1e6 - m.time_boot_ms / 1e3
        elif t == "SYS_STATUS":
            s["battery_pct"] = m.battery_remaining if m.battery_remaining >= 0 else None
        elif t == "GPS_RAW_INT":
            s.update(gnss_fix=m.fix_type, gnss_sats=m.satellites_visible,
                     gnss_eph_m=(m.h_acc / 1000) if getattr(m, "h_acc", 0) else None)
        elif t == "VFR_HUD":
            s["gs_ms"] = m.groundspeed
        elif t == "GLOBAL_POSITION_INT":
            s.update(lat=m.lat / 1e7, lon=m.lon / 1e7, alt_amsl_m=m.alt / 1000, alt_rel_m=m.relative_alt / 1000,
                     vs_ms=-m.vz / 100, heading_deg=None if m.hdg == 65535 else m.hdg / 100)
            now = time.time()
            if d.boot_offset is None or now - d.last_pub < 1 / RATE_HZ:
                return
            d.last_pub = now
            d.seq += 1
            msg = make(drone_id=d.id, vendor=d.vendor, seq=d.seq, ts=iso(d.boot_offset + m.time_boot_ms / 1e3),
                       rx_ts=iso(now), **s)
            self.out.put_nowait((f"tlm.{d.id}", f"{d.id}-{d.seq}", json.dumps(msg).encode()))

    def request_streams(self, d):
        # ArduPilot sends little on a new link until asked: ask for the messages we normalize at 5 Hz.
        # PX4 streams them by default, but SYSTEM_TIME (id 2) has to be asked for, here once a second.
        wanted = [(1, 200000), (2, 1000000), (24, 200000), (33, 200000), (74, 200000)]
        if d.vendor == "px4":
            wanted = [(2, 1000000)]
        for msg_id, interval_us in wanted:
            d.send(mavlink2.MAVLink_command_long_message(1, 1 if d.vendor == "px4" else 0,
                                                         mavlink2.MAV_CMD_SET_MESSAGE_INTERVAL, 0,
                                                         msg_id, interval_us, 0, 0, 0, 0, 0))
        d.requested = True

    async def heartbeats(self):
        while True:
            for d in list(self.drones.values()):
                d.send(mavlink2.MAVLink_heartbeat_message(mavlink2.MAV_TYPE_GCS, mavlink2.MAV_AUTOPILOT_INVALID,
                                                          0, 0, 0, 3))
            await asyncio.sleep(1)

    # ---------- cloud side ----------
    async def publisher(self, js, nc):
        while True:
            subject, msg_id, body = await self.out.get()
            backlog = len(self.pending) or self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
            if backlog == 0 and nc.is_connected and time.time() > self.down_until:
                try:
                    await js.publish(subject, body, headers={"Nats-Msg-Id": msg_id}, timeout=3)
                    self.stats["sent"] += 1
                    continue
                except Exception as e:
                    self.down_until = time.time() + 3
                    log(f"publish failed ({type(e).__name__}), buffering on disk")
            # one disk write per message blocked this loop for up to seconds on a busy disk, and the MAVLink
            # packets that arrived meanwhile were dropped by the kernel: so collect here, write in batches
            self.pending.append((subject, msg_id, body))
            self.stats["buffered"] += 1

    async def flusher(self, js, nc):
        # the backlog must drain faster than new data arrives, so send a batch at once and wait for all the acks
        # (the first version sent one message, waited for its ack, then the next: under load it never caught up)
        while True:
            await asyncio.sleep(0.5)
            if self.pending:
                rows, self.pending = self.pending, []
                await asyncio.to_thread(self.persist, rows)
            if not nc.is_connected or time.time() < self.down_until:
                continue
            rows = self.db.execute("SELECT id, subject, msg_id, body FROM outbox ORDER BY id LIMIT 100").fetchall()
            if not rows:
                continue
            results = await asyncio.gather(*[js.publish(subj, body, headers={"Nats-Msg-Id": mid}, timeout=5)
                                             for _, subj, mid, body in rows], return_exceptions=True)
            done = [(r[0],) for r, res in zip(rows, results) if not isinstance(res, Exception)]
            self.db.executemany("DELETE FROM outbox WHERE id = ?", done)
            self.db.commit()
            self.stats["flushed"] += len(done)
            if len(done) < len(rows):
                self.down_until = time.time() + 3

    def persist(self, rows):
        db = sqlite3.connect("/buf/outbox.db")
        db.executemany("INSERT INTO outbox (subject, msg_id, body) VALUES (?, ?, ?)", rows)
        db.commit()
        db.close()

    async def report(self):
        while True:
            await asyncio.sleep(5)
            backlog = self.db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
            for d in self.drones.values():
                if d.boot_offset is None:
                    log(f"{d.id}: no SYSTEM_TIME with a real date yet, holding its messages "
                        f"(last time_unix_usec {d.last_unix})")
            log(f"sent {self.stats['sent']}, buffered {self.stats['buffered']}, flushed {self.stats['flushed']}, "
                f"waiting on disk {backlog}")

    async def run(self):
        loop = asyncio.get_running_loop()
        for port in (14550, 14560):     # 14550: PX4 GCS link, 14560: ArduPilot serial0
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            sock.bind(("0.0.0.0", port))
            sock.setblocking(False)
            loop.add_reader(sock, lambda s=sock: self.on_mavlink(s, *s.recvfrom(65535)))
        async def down():
            log("NATS disconnected")

        async def up():
            log("NATS reconnected")

        async def err(e):
            pass                         # reconnect attempts fail while the link is down; down() already said so

        nc = await nats.connect(NATS_URL, ping_interval=2, max_outstanding_pings=2, max_reconnect_attempts=-1,
                                reconnect_time_wait=1, connect_timeout=2, disconnected_cb=down, reconnected_cb=up,
                                error_cb=err)
        js = nc.jetstream()
        log(f"connected to {NATS_URL}, waiting for {len(DRONES)} drones")
        await asyncio.gather(self.heartbeats(), self.publisher(js, nc), self.flusher(js, nc), self.report())


if __name__ == "__main__":
    asyncio.run(Gateway().run())

"""Command-center core: vendor X adapter, TimescaleDB writer, live state and alerts over WebSocket,
and the post-flight log upload API.

  GET  /                 the live map page          GET /ws          live snapshots twice a second
  GET  /api/state        the same snapshot as JSON  GET /api/time    server clock (for latency tests)
  POST /api/logs         multipart: drone_id, file (.ulg or .bin) -> parsed summary, stored in flight_logs
"""
import asyncio
import json
import os
import time
from datetime import datetime, timezone

import asyncpg
import nats
from aiohttp import web, WSMsgType
from nats.js.api import StreamConfig

from schema import iso, make, parse_iso

NATS_URL = os.environ.get("NATS_URL", "nats://lab-dops-nats:4222")
PG = os.environ.get("PG", "postgresql://postgres:lab@lab-dops-db:5432/fleet")
VX_ADAPTER = int(os.environ.get("VX_ADAPTER", "1"))    # 1 = trusts the vendor document, 2 = checks units and clock
LINK_LOST_S, LOW_BATT, CEILING_M, SKEW_S = 5, 30, 120, 5

DDL = """
CREATE TABLE IF NOT EXISTS telemetry (
  ts timestamptz NOT NULL, drone_id text NOT NULL, seq bigint, vendor text, rx_ts timestamptz,
  lat double precision, lon double precision, alt_amsl_m real, alt_rel_m real, gs_ms real, vs_ms real,
  heading_deg real, battery_pct real, mode text, armed boolean,
  gnss_fix smallint, gnss_sats smallint, gnss_eph_m real, stored_at timestamptz DEFAULT now(),
  PRIMARY KEY (drone_id, ts));
SELECT create_hypertable('telemetry', by_range('ts', INTERVAL '1 day'), if_not_exists => TRUE);
CREATE TABLE IF NOT EXISTS flight_logs (
  id bigserial PRIMARY KEY, drone_id text NOT NULL, kind text NOT NULL, file_name text, size_bytes bigint,
  started_at timestamptz, duration_s real, summary jsonb, uploaded_at timestamptz DEFAULT now());
"""
COLS = ("ts", "drone_id", "seq", "vendor", "rx_ts", "lat", "lon", "alt_amsl_m", "alt_rel_m", "gs_ms", "vs_ms",
        "heading_deg", "battery_pct", "mode", "armed", "gnss_fix", "gnss_sats", "gnss_eph_m")
TYPES = ("timestamptz", "text", "int8", "text", "timestamptz", "float8", "float8", "float4", "float4", "float4",
         "float4", "float4", "float4", "text", "bool", "int2", "int2", "float4")
# one statement per batch; RETURNING tells how many rows were new (a replayed duplicate returns nothing)
INSERT = (f"INSERT INTO telemetry ({', '.join(COLS)}) SELECT * FROM unnest("
          + ", ".join(f"${i + 1}::{t}[]" for i, t in enumerate(TYPES))
          + ") ON CONFLICT (drone_id, ts) DO NOTHING RETURNING 1")


def log(msg):
    print(f"{time.strftime('%H:%M:%S')} core {msg}", flush=True)


class Core:
    def __init__(self):
        self.latest = {}      # drone_id -> last accepted message (by event time)
        self.last_rx = {}     # drone_id -> local time we last heard anything
        self.counters = {"late": 0, "future": 0, "written": 0, "duplicates": 0, "quarantined": 0,
                         "clock_replaced": 0, "rejected": 0}
        self.clients = set()

    # ---------- vendor X adapter ----------
    async def vendor_x(self, nc, js):
        async def on(m):
            raw = json.loads(m.data)
            rx = time.time()
            vts = datetime.strptime(raw["time"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp()
            alt = raw["alt"]
            if VX_ADAPTER >= 2:
                unit = raw.get("alt_unit", "m")
                if unit not in ("m", "ft"):
                    self.counters["quarantined"] += 1
                    await nc.publish("quarantine.vendorx", m.data)
                    return
                alt = alt * 0.3048 if unit == "ft" else alt
                if abs(vts - rx) > 30:           # a zone-less or drifting vendor clock: keep our receive time
                    self.counters["clock_replaced"] += 1
                    vts = rx
            msg = make(drone_id=raw["sn"].lower(), vendor="vendor-x", seq=None, ts=iso(vts), rx_ts=iso(rx),
                       lat=raw["pos"]["lat"], lon=raw["pos"]["lng"], alt_rel_m=alt, gs_ms=raw["spd_kmh"] / 3.6,
                       heading_deg=raw["hdg"], battery_pct=raw["batt"] * 100, mode=raw["state"], armed=True)
            try:
                await js.publish(f"tlm.{msg['drone_id']}", json.dumps(msg).encode(),
                                 headers={"Nats-Msg-Id": f"{msg['drone_id']}-{raw['time']}"})
            except Exception as e:
                log(f"vendor X publish failed: {type(e).__name__}")
        await nc.subscribe("vendorx.*.telemetry", cb=on)

    # ---------- storage ----------
    async def writer(self, js, pool):
        sub = await js.pull_subscribe("tlm.>", durable="db-writer", stream="TLM")
        while True:
            try:
                batch = await sub.fetch(200, timeout=1)
            except (nats.errors.TimeoutError, asyncio.TimeoutError):
                continue
            rows = []
            for m in batch:
                try:
                    d = json.loads(m.data)
                    assert d.get("v") == 1 and d.get("ts") and d.get("drone_id")
                except Exception:
                    self.counters["rejected"] += 1      # not schema v1: keep it out of the table
                    continue
                rows.append(tuple(datetime.fromisoformat(d[c].replace("Z", "+00:00")) if c in ("ts", "rx_ts")
                                  else d.get(c) for c in COLS))
            if not rows:
                for m in batch:
                    await m.ack()
                continue
            async with pool.acquire() as con:
                new = len(await con.fetch(INSERT, *[list(col) for col in zip(*rows)]))
            self.counters["written"] += new
            self.counters["duplicates"] += len(rows) - new
            for m in batch:
                await m.ack()

    # ---------- live state ----------
    async def live(self, nc):
        async def on(m):
            try:
                d = json.loads(m.data)
                did = d["drone_id"]
                ts = parse_iso(d["ts"])
            except Exception:
                return
            now = time.time()
            self.last_rx[did] = now
            if ts > now + SKEW_S:
                self.counters["future"] += 1
                if VX_ADAPTER >= 2:              # never let a future clock decide what "latest" is
                    return
            cur = self.latest.get(did)
            if cur and parse_iso(cur["ts"]) >= ts:
                self.counters["late"] += 1       # an old message (replayed after an outage): store it, don't draw it
                return
            self.latest[did] = d
        await nc.subscribe("tlm.>", cb=on)

    def snapshot(self):
        now = time.time()
        alerts = []
        for did, d in sorted(self.latest.items()):
            age = now - self.last_rx.get(did, 0)
            skew = parse_iso(d["rx_ts"]) - parse_iso(d["ts"])
            if age > LINK_LOST_S:
                alerts.append({"drone": did, "type": "link", "text": f"no data for {age:.0f} s"})
            if d.get("battery_pct") is not None and d["battery_pct"] < LOW_BATT:
                alerts.append({"drone": did, "type": "battery", "text": f"battery {d['battery_pct']:.0f}%"})
            if d.get("alt_rel_m") and d["alt_rel_m"] > CEILING_M:
                alerts.append({"drone": did, "type": "ceiling", "text": f"altitude {d['alt_rel_m']:.0f} m"})
            if abs(skew) > SKEW_S:
                alerts.append({"drone": did, "type": "clock", "text": f"clock skew {skew:+.0f} s"})
        return {"server_time": iso(now), "drones": list(self.latest.values()), "alerts": alerts,
                "counters": self.counters}

    async def push(self):
        while True:
            await asyncio.sleep(0.5)
            if self.clients:
                data = json.dumps(self.snapshot())
                for ws in list(self.clients):
                    try:
                        await ws.send_str(data)
                    except Exception:
                        self.clients.discard(ws)

    # ---------- HTTP ----------
    async def ws(self, request):
        ws = web.WebSocketResponse(heartbeat=10)
        await ws.prepare(request)
        self.clients.add(ws)
        async for msg in ws:
            if msg.type == WSMsgType.ERROR:
                break
        self.clients.discard(ws)
        return ws

    async def upload(self, request):
        from logparse import parse
        form = await request.post()
        f = form["file"]
        data = f.file.read()
        path = f"/data/logs/{int(time.time())}-{f.filename}"
        os.makedirs("/data/logs", exist_ok=True)
        with open(path, "wb") as out:
            out.write(data)
        t0 = time.time()
        summary = await asyncio.to_thread(parse, path)
        summary["parse_seconds"] = round(time.time() - t0, 2)
        started = datetime.fromtimestamp(summary["start_unix"], timezone.utc) if summary.get("start_unix") else None
        async with request.app["pool"].acquire() as con:
            row_id = await con.fetchval(
                "INSERT INTO flight_logs (drone_id, kind, file_name, size_bytes, started_at, duration_s, summary) "
                "VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id", form["drone_id"], summary["kind"], f.filename,
                len(data), started, summary.get("duration_s"), json.dumps(summary))
        return web.json_response({"id": row_id, **summary})

    async def run(self):
        while True:                       # the database may take a minute on first start
            try:
                pool = await asyncpg.create_pool(PG, min_size=1, max_size=4)
                break
            except Exception as e:
                log(f"database not ready: {e}")
                await asyncio.sleep(2)
        async with pool.acquire() as con:
            await con.execute(DDL)
        nc = await nats.connect(NATS_URL, max_reconnect_attempts=-1)
        js = nc.jetstream()
        await js.add_stream(StreamConfig(name="TLM", subjects=["tlm.>"], max_age=7 * 86400, duplicate_window=120))
        await self.vendor_x(nc, js)
        await self.live(nc)
        app = web.Application(client_max_size=512 * 1024 * 1024)
        app["pool"] = pool
        app.add_routes([web.get("/ws", self.ws),
                        web.get("/api/state", lambda r: web.json_response(self.snapshot())),
                        web.get("/api/time", lambda r: web.json_response({"t": time.time() * 1000})),
                        web.post("/api/logs", self.upload),
                        web.get("/", lambda r: web.FileResponse("web/index.html")),
                        web.static("/web", "web")])
        runner = web.AppRunner(app)
        await runner.setup()
        await web.TCPSite(runner, "0.0.0.0", 8080).start()
        log(f"listening on 8080, vendor X adapter v{VX_ADAPTER}")
        await asyncio.gather(self.writer(js, pool), self.push())


if __name__ == "__main__":
    asyncio.run(Core().run())

"""A simulated drone dock from "vendor X" that speaks JSON over MQTT, the way many vendor clouds do.

It is not a real product. Its documented payload (firmware 1.x):
  {"sn", "fw", "time": "YYYY-MM-DD HH:MM:SS" (UTC), "pos": {"lat","lng"}, "alt": metres above take-off,
   "spd_kmh", "hdg", "batt": 0..1, "state"}
FIRMWARE=2 is the "update" the integrator was not told about: altitude in feet with "alt_unit": "ft",
and the time in the dock's local time (UTC+3) without any zone.
"""
import json
import math
import os
import time
from datetime import datetime, timedelta, timezone

import paho.mqtt.client as mqtt

SN = os.environ.get("VX_SN", "VX-1001")
FW = int(os.environ.get("FIRMWARE", "1"))
HOME = (47.3990, 8.5440)

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=SN)
c.connect(os.environ.get("MQTT_HOST", "lab-dops-nats"), 1883)
c.loop_start()
t0 = time.time()
batt = 0.95
print(f"vendor X dock {SN} firmware {FW}.0 publishing to vendorx/{SN}/telemetry", flush=True)
while True:
    t = time.time() - t0
    a = t / 30                                   # one circle every ~3 minutes
    lat = HOME[0] + 90 / 111320 * math.cos(a)
    lng = HOME[1] + 90 / (111320 * math.cos(math.radians(HOME[0]))) * math.sin(a)
    alt_m = 45.0
    batt = max(0.05, batt - 0.0004)
    now = datetime.now(timezone.utc)
    msg = {"sn": SN, "fw": f"{FW}.0", "pos": {"lat": round(lat, 7), "lng": round(lng, 7)},
           "spd_kmh": round(90 / 30 * 3.6, 1), "hdg": round((math.degrees(a) + 90) % 360, 1),
           "batt": round(batt, 3), "state": "PATROL"}
    if FW == 1:
        msg.update(time=now.strftime("%Y-%m-%d %H:%M:%S"), alt=alt_m)
    else:
        msg.update(time=(now + timedelta(hours=3)).strftime("%Y-%m-%d %H:%M:%S"), alt=round(alt_m * 3.28084, 1),
                   alt_unit="ft")
    c.publish(f"vendorx/{SN}/telemetry", json.dumps(msg), qos=1)
    time.sleep(1)

"""Post-flight log parsing: PX4 ULog (.ulg) with pyulog, ArduPilot DataFlash (.bin) with pymavlink.
Both end in the same small summary, so the ops platform can list flights from any autopilot together.

  python logparse.py <file>
"""
import json
import sys
from collections import Counter

PX4_NAV = {0: "MANUAL", 2: "POSCTL", 3: "MISSION", 4: "LOITER", 5: "RTL", 14: "OFFBOARD", 17: "TAKEOFF", 18: "LAND"}


def parse_ulog(path):
    from pyulog import ULog
    # only the topics the summary needs: a long SITL log holds hundreds of MB of others
    u = ULog(path, message_name_filter_list=["vehicle_status", "vehicle_global_position", "sensor_combined",
                                             "vehicle_attitude", "vehicle_gps_position"])
    dur = (u.last_timestamp - u.start_timestamp) / 1e6
    boot_utc = u.msg_info_dict.get("boot_time_utc_us")
    topics = {f"{d.name}/{d.multi_id}": d for d in u.data_list}

    def rate(name):
        d = topics.get(f"{name}/0")
        return round(len(d.data["timestamp"]) / dur, 1) if d else None

    gp = topics.get("vehicle_global_position/0")
    st = topics.get("vehicle_status/0")
    modes, last = [], None
    if st:
        for t, nav in zip(st.data["timestamp"], st.data["nav_state"]):
            if nav != last:
                modes.append([round((int(t) - u.start_timestamp) / 1e6, 1), PX4_NAV.get(int(nav), int(nav))])
                last = nav
    return {
        "kind": "px4-ulog", "software": u.get_version_info_str(),
        "start_unix": (boot_utc + u.start_timestamp) / 1e6 if boot_utc else None, "duration_s": round(dur, 1),
        "topics_read": len(u.data_list), "dropouts": len(u.dropouts),
        "rate_hz": {"sensor_combined": rate("sensor_combined"), "vehicle_attitude": rate("vehicle_attitude"),
                    "vehicle_gps_position": rate("vehicle_gps_position")},
        "max_alt_amsl_m": round(float(max(gp.data["alt"])), 1) if gp else None,
        "modes": modes,
        "warnings": [m.message for m in u.logged_messages
                     if m.log_level_str() in ("WARNING", "ERROR", "CRITICAL", "ALERT", "EMERGENCY")][:10],
    }


def parse_bin(path):
    from pymavlink import mavutil
    mlog = mavutil.mavlink_connection(path)
    counts, modes, errors, texts = Counter(), [], [], []
    first = last = None
    max_alt = None
    while True:
        m = mlog.recv_msg()
        if m is None:
            break
        t = m.get_type()
        counts[t] += 1
        ts = getattr(m, "_timestamp", None)
        if ts and ts > 1.6e9:
            first = first or ts
            last = ts
        if t == "MODE":
            modes.append([round(ts - first, 1) if first else None, mlog.flightmode])
        elif t == "ERR":
            errors.append(f"Subsys {m.Subsys} ECode {m.ECode}")
        elif t == "MSG":
            texts.append(m.Message)
        elif t == "CTUN":
            max_alt = m.Alt if max_alt is None else max(max_alt, m.Alt)
    dur = (last - first) if first else 0
    return {
        "kind": "ardupilot-bin", "software": next((x for x in texts if x.startswith("ArduCopter")), None),
        "start_unix": first, "duration_s": round(dur, 1), "message_types": len(counts),
        "rate_hz": {k: round(counts[k] / dur, 1) for k in ("IMU", "ATT", "GPS") if dur},
        "max_alt_rel_m": round(max_alt, 1) if max_alt is not None else None,
        "modes": modes, "errors": errors[:10], "messages": texts[:10],
    }


def parse(path):
    with open(path, "rb") as f:
        head = f.read(7)
    if head.startswith(b"ULog"):
        return parse_ulog(path)
    if head[:2] == b"\xa3\x95":          # DataFlash binary message header
        return parse_bin(path)
    raise ValueError("not a ULog or DataFlash file")


if __name__ == "__main__":
    print(json.dumps(parse(sys.argv[1]), indent=1))

"""A test "drone camera" for measuring glass-to-glass latency.

Every frame carries the wall clock of the moment it was made (milliseconds, modulo 2^32) as 32 black/white
blocks in the top-left corner, between a white and a black reference block. The viewer page reads the blocks
back from the decoded video and subtracts them from its own clock.

  python3 camera.py <rtsp-tcp|rtsp-udp|rtmp|srt> [bitrate_kbps] [host] [path]
"""
import subprocess
import sys
import time

import numpy as np

W, H, FPS = 640, 360, 15
BLOCK, X0, Y0 = 16, 16, 16

proto = sys.argv[1]
kbps = int(sys.argv[2]) if len(sys.argv) > 2 else 600
host = sys.argv[3] if len(sys.argv) > 3 else "lab-dops-mtx"
path = sys.argv[4] if len(sys.argv) > 4 else "cam"

OUT = {
    "rtsp-tcp": ["-f", "rtsp", "-rtsp_transport", "tcp", f"rtsp://{host}:8554/{path}"],
    "rtsp-udp": ["-f", "rtsp", "-rtsp_transport", "udp", f"rtsp://{host}:8554/{path}"],
    "rtmp": ["-f", "flv", f"rtmp://{host}:1935/{path}"],
    "srt": ["-f", "mpegts", f"srt://{host}:8890?streamid=publish:{path}&pkt_size=1316&latency=200000"],
}[proto]

ffmpeg = subprocess.Popen(
    ["ffmpeg", "-hide_banner", "-loglevel", "error", "-f", "rawvideo", "-pix_fmt", "rgb24", "-s", f"{W}x{H}",
     "-r", str(FPS), "-i", "-", "-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency",
     "-pix_fmt", "yuv420p", "-g", str(FPS), "-b:v", f"{kbps}k", "-maxrate", f"{kbps}k",
     "-bufsize", f"{kbps // 2}k", *OUT],
    stdin=subprocess.PIPE)

# a moving scene so the encoder has real work: 30 prepared frames of shifted stripes and noise, played in a loop
rng = np.random.default_rng(7)
noise = rng.integers(0, 50, size=(H, W), dtype=np.uint8)
stripes = ((np.arange(W + H)[None, :] // 24) % 2 * 120).astype(np.uint8)
SCENE = []
for k in range(FPS):
    g = stripes[:, k * 8 % 48:][:, :W].repeat(H, axis=0) + noise
    SCENE.append(np.stack([g, g // 2 + 60, 200 - g // 2], axis=2))


def frame(n, stamp):
    img = SCENE[n % FPS].copy()
    img[0:Y0 * 2 + BLOCK, 0:X0 * 2 + 34 * BLOCK] = 128           # grey band behind the code
    bits = [1] + [(stamp >> (31 - i)) & 1 for i in range(32)] + [0]  # white ref, 32 bits, black ref
    for i, b in enumerate(bits):
        x = X0 + i * BLOCK
        img[Y0:Y0 + BLOCK, x:x + BLOCK] = 255 if b else 0
    return img.tobytes()


start = time.time()
n = 0
late = 0
while True:
    due = start + n / FPS
    now = time.time()
    if now < due:
        time.sleep(due - now)
    elif now - due > 1 / FPS:
        late += 1                                       # we could not keep up (the pipe was blocked)
    stamp = int(time.time() * 1000) % (1 << 32)
    try:
        ffmpeg.stdin.write(frame(n, stamp))
    except BrokenPipeError:
        sys.exit("ffmpeg stopped")
    n += 1
    if n % (FPS * 10) == 0:
        print(f"{proto}: {n} frames sent, {late} frames late, {n / (time.time() - start):.1f} fps", flush=True)

<!doctype html>
<html lang="en">
<head><meta charset="utf-8"><title>Glass to glass</title>
<style>body { margin: 0; background: #111; color: #eee; font: 16px monospace; } video { width: 640px; } #out { padding: 8px; }</style>
</head>
<body>
<video id="v" muted autoplay playsinline></video>
<div id="out">starting</div>
<script>
// 1) clock offset between this browser and the lab containers, from the round trip with the least delay
// 2) WHEP playback from MediaMTX
// 3) for every decoded frame: read the 32 blocks the camera drew, latency = our clock - camera clock
const W = 640, H = 360, BLOCK = 16, X0 = 16, Y0 = 16;
window.samples = []; window.bad = 0; window.frames = 0; window.offset = null;
const out = document.getElementById('out');

async function clockOffset() {
  let best = null;
  for (let i = 0; i < 20; i++) {
    const t0 = Date.now(); const r = await (await fetch('/api/time', { cache: 'no-store' })).json(); const t1 = Date.now();
    if (!best || t1 - t0 < best.rtt) best = { rtt: t1 - t0, off: r.t - (t0 + t1) / 2 };
  }
  return best;
}

async function play(url) {
  const pc = new RTCPeerConnection();
  pc.addTransceiver('video', { direction: 'recvonly' });
  const v = document.getElementById('v');
  pc.ontrack = (e) => { v.srcObject = e.streams[0]; };
  await pc.setLocalDescription(await pc.createOffer());
  await new Promise(r => { pc.onicegatheringstatechange = () => pc.iceGatheringState === 'complete' && r(); setTimeout(r, 2000); });
  const res = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/sdp' }, body: pc.localDescription.sdp });
  if (!res.ok) throw new Error('WHEP ' + res.status);
  await pc.setRemoteDescription({ type: 'answer', sdp: await res.text() });
  return v;
}

function decode(ctx) {
  const px = ctx.getImageData(0, 0, X0 + 34 * BLOCK + 8, Y0 + BLOCK).data;
  const lum = (i) => { const x = X0 + i * BLOCK + BLOCK / 2, y = Y0 + BLOCK / 2, k = (y * (X0 + 34 * BLOCK + 8) + x) * 4; return px[k]; };
  const white = lum(0), black = lum(33), mid = (white + black) / 2;
  if (white - black < 100) return null;          // the code is not readable (corrupt frame)
  let v = 0;
  for (let i = 1; i <= 32; i++) v = (v * 2) + (lum(i) > mid ? 1 : 0);
  return v;
}

(async () => {
  const p = new URLSearchParams(location.search);
  const off = await clockOffset(); window.offset = off;
  let v;
  for (let i = 0; ; i++) {                       // the camera may not be publishing yet: retry for a minute
    try { v = await play(p.get('whep')); break; } catch (e) { if (i > 30) throw e; out.textContent = 'waiting: ' + e; await new Promise(r => setTimeout(r, 2000)); }
  }
  const c = document.createElement('canvas'); c.width = W; c.height = H;
  const ctx = c.getContext('2d', { willReadFrequently: true });
  const onFrame = () => {
    window.frames++;
    ctx.drawImage(v, 0, 0, W, H);
    const code = decode(ctx);
    const now = (Date.now() + off.off) % 2 ** 32;    // our clock in the containers' time base
    if (code === null) window.bad++;
    else {
      const lat = now - code;
      if (lat >= 0 && lat < 60000) window.samples.push(lat); else window.bad++;
    }
    v.requestVideoFrameCallback(onFrame);
  };
  v.requestVideoFrameCallback(onFrame);
  setInterval(() => { const s = window.samples.slice(-60).sort((a, b) => a - b);
    out.textContent = `offset ${off.off.toFixed(1)} ms (rtt ${off.rtt} ms) frames ${window.frames} bad ${window.bad} median(last 60) ${s[s.length >> 1] ?? '-'} ms`; }, 1000);
})().catch(e => out.textContent = 'error ' + e);
</script>
</body>
</html>

// Glass-to-glass latency per ingest protocol: the test camera publishes to MediaMTX, a headless Chrome on the
// host plays it over WebRTC (WHEP) and reads the time code from every frame.
//   node content/articles/drone-ops-lab/lab/measure-latency.mjs [condition]
// condition: clean | loss2 (2% of the camera's packets dropped before MediaMTX reads them)
import { execFileSync } from 'node:child_process';
import { fileURLToPath } from 'node:url';
import { dirname, resolve } from 'node:path';
import { launch } from '../../../../scripts/lab/browser.mjs';

const condition = process.argv[2] || 'clean';
const CAMERA = resolve(dirname(fileURLToPath(import.meta.url)), 'camera.py');   // the current camera.py, mounted
const PROTOS = (process.env.PROTOS || 'rtsp-tcp,rtsp-udp,rtmp,srt').split(',');
const sh = (...a) => execFileSync('docker', a, { encoding: 'utf8', env: { ...process.env, MSYS_NO_PATHCONV: '1' } }).trim();
const sleep = (ms) => new Promise(r => setTimeout(r, ms));
// tc netem is not available in this Docker Desktop kernel, so the weak link is made with iptables rules on the
// receiving side (MediaMTX's INPUT, packets from the camera only); a DROP on the sender's own OUTPUT chain makes
// sendto() fail with EPERM and ffmpeg gives up, which is not what a radio link does
const RULES = {
    clean: [],
    loss2: (ip) => [['-A', 'INPUT', '-s', ip, '-m', 'statistic', '--mode', 'random', '--probability', '0.02', '-j', 'DROP']],
};
const mtxIptables = (...a) => sh('run', '--rm', '--network', 'container:lab-dops-mtx', '--cap-add', 'NET_ADMIN',
    'lab-dops-py:1', 'iptables', ...a);

const pct = (a, p) => a.length ? a[Math.min(a.length - 1, Math.floor(a.length * p))] : null;
const b = await launch({ port: 9452, width: 1000, height: 640 });
try {
    for (const proto of PROTOS) {
        const cam = `lab-dops-cam-${proto}`;      // --init: ffmpeg children are reaped and the container stops cleanly
        try { sh('rm', '-f', cam); } catch {}
        sh('run', '-d', '--init', '--name', cam, '--network', 'lab-dops-cloud', '-v', `${CAMERA}:/cam/camera.py:ro`,
            '--entrypoint', 'sh', 'lab-dops-cam:1', '-c', 'sleep 3600');
        const ip = sh('inspect', '-f', '{{(index .NetworkSettings.Networks "lab-dops-cloud").IPAddress}}', cam);
        mtxIptables('-F', 'INPUT');
        for (const r of (condition === 'clean' ? [] : RULES[condition](ip))) mtxIptables(...r);
        const path = `${condition}-${proto}`;           // a fresh MediaMTX path per run, so an old publisher cannot block it
        const since = new Date().toISOString();
        sh('exec', '-d', cam, 'sh', '-c', `python3 -u camera.py ${proto} 600 lab-dops-mtx ${path} > /tmp/cam.log 2>&1`);
        for (let i = 0; i < 45 && !sh('logs', '--since', since, 'lab-dops-mtx').includes(`publishing to path '${path}'`); i++) await sleep(2000);
        await b.goto('http://127.0.0.1:38080/web/latency.html?whep=http://127.0.0.1:38889/' + path + '/whep', { wait: 1000 });
        await sleep(15000);                                   // let the WebRTC jitter buffer settle
        const start = await b.evaluate('({ n: window.samples.length, f: window.frames, bad: window.bad })');
        await sleep(30000);
        const r = await b.evaluate('({ s: window.samples, f: window.frames, bad: window.bad, off: window.offset })');
        const s = r.s.slice(start.n).sort((x, y) => x - y);
        console.log(JSON.stringify({ condition, proto, frames_30s: r.f - start.f, unreadable: r.bad - start.bad,
            samples: s.length, min: s[0], p50: pct(s, 0.5), p95: pct(s, 0.95), max: s[s.length - 1],
            clock_offset_ms: Math.round(r.off.off), rtt_ms: r.off.rtt }));
        await b.goto('about:blank', { wait: 500 });
        console.log(sh('exec', cam, 'sh', '-c', 'tail -2 /tmp/cam.log'));
        try { sh('rm', '-f', cam); } catch {}
        mtxIptables('-F', 'INPUT');
    }
} finally {
    await b.close();
}