Mini Airspace Deconfliction Manager

The first requirement is a busy sky. I run three ArduPilot SITL instances on a homelab node, each with its own instance offset (-I, which shifts the base ports by 10 per instance) and its own MAVLink system ID (--sysid), and each streaming telemetry out to a dedicated UDP port where the manager listens. Every instance flies an autonomous waypoint mission loaded through MAVProxy, so the virtual field has real crossing traffic, not three drones hovering politely apart.

BASH, LAUNCHING THREE SITL INSTANCES

# terminal 1 - drone 1 (default ports, telemetry out on 14550)
sim_vehicle.py -v ArduCopter -I0 --sysid 1 --out=udp:127.0.0.1:14550

# terminal 2 - drone 2 (base ports shifted by 10, telemetry on 14551)
sim_vehicle.py -v ArduCopter -I1 --sysid 2 --out=udp:127.0.0.1:14551

# terminal 3 - drone 3 (telemetry on 14552)
sim_vehicle.py -v ArduCopter -I2 --sysid 3 --out=udp:127.0.0.1:14552

# load a mission on drone 1 from its MAVProxy console:
#   wp load ../missions/cross_a.txt
#   mode auto

What can go wrong: forgetting the -I offset makes the second instance collide with the first on ports 5760/5763, and the failure mode is a silently empty telemetry stream rather than an error. I verified each instance by watching for its heartbeat and distinct sysid on its own output port before moving on.

The manager opens one pymavlink connection per SITL instance, waits for the heartbeat, then explicitly requests a 10 Hz GLOBAL_POSITION_INT stream, the default stream rates are far too low for conflict prediction. Every message is normalised into one track model (id, lat, lon, alt, vx, vy, vz, ts, source) so the rest of the system never cares whether a track came from a simulator or a beacon.

PYTHON, TRACK INGEST (ingest.py, trimmed)

from pymavlink import mavutil

SOURCES = {
    1: "udp:127.0.0.1:14550",
    2: "udp:127.0.0.1:14551",
    3: "udp:127.0.0.1:14552",
}

conns = {sid: mavutil.mavlink_connection(url) for sid, url in SOURCES.items()}
for sid, c in conns.items():
    c.wait_heartbeat()
    # ask the autopilot for GLOBAL_POSITION_INT at 10 Hz (interval in microseconds)
    c.mav.command_long_send(
        c.target_system, c.target_component,
        mavutil.mavlink.MAV_CMD_SET_MESSAGE_INTERVAL, 0,
        mavutil.mavlink.MAVLINK_MSG_ID_GLOBAL_POSITION_INT,
        100000, 0, 0, 0, 0, 0)

msg = conns[1].recv_match(type="GLOBAL_POSITION_INT", blocking=True, timeout=2)
track = {
    "id": msg.get_srcSystem(), "lat": msg.lat / 1e7, "lon": msg.lon / 1e7,
    "alt": msg.relative_alt / 1000.0,
    "vx": msg.vx / 100.0, "vy": msg.vy / 100.0,
    # GLOBAL_POSITION_INT vz is cm/s positive DOWN; the track model is up-positive
    "vz": -msg.vz / 100.0,
    "source": "sitl",
}

How I verified: I printed the inter-arrival time of GLOBAL_POSITION_INT per instance; before the SET_MESSAGE_INTERVAL request it sat near 1 Hz and the predicted tracks jumped, after it the stream settled at 10 Hz and predictions smoothed out.

Three cheap ESP32 boards act as tracker beacons, stand-ins for live aircraft. Each runs a small Arduino sketch (Arduino-ESP32 core) that joins the bench Wi-Fi and broadcasts a JSON position packet over UDP to the manager at 5 Hz, in exactly the track model from step 02 so the ingest path cannot tell simulated from physical. The beacons matter less for their data than for their reality: they force the pipeline to survive dropped packets, jitter, and reboots, nuisances SITL never has.

ARDUINO, BEACON FIRMWARE (beacon.ino, trimmed)

#include <WiFi.h>
#include <WiFiUdp.h>

const char* SSID = "benchlab";
const char* PASS = "....";
const char* MANAGER_IP = "192.168.1.50";
const uint16_t MANAGER_PORT = 5005;
const int BEACON_ID = 101;      // unique per board

WiFiUDP udp;

void setup() {
  WiFi.mode(WIFI_STA);
  WiFi.setSleep(false);          // modem sleep eats UDP packets in bursts
  WiFi.begin(SSID, PASS);
  while (WiFi.status() != WL_CONNECTED) delay(250);
}

void loop() {
  // stand-in track: a slow circle, like a drone loitering on station
  float t = millis() / 1000.0f;
  float lat = 43.7615f + 0.0005f * cosf(t / 20.0f);
  float lon = -79.4111f + 0.0005f * sinf(t / 20.0f);
  char pkt[160];
  snprintf(pkt, sizeof(pkt),
    "{\"src\":\"beacon\",\"id\":%d,\"lat\":%.6f,\"lon\":%.6f,\"alt\":%.1f,\"vx\":%.2f,\"vy\":%.2f}",
    BEACON_ID, lat, lon, 120.0f, 0.0f, 0.0f);
  udp.beginPacket(MANAGER_IP, MANAGER_PORT);
  udp.write((const uint8_t*)pkt, strlen(pkt));
  udp.endPacket();
  delay(200);                    // 5 Hz
}

What can go wrong: ESP32 modem sleep (WIFI_PS_MIN_MODEM is the default) buffers and bursts UDP packets, which looks exactly like a manoeuvring aircraft to the conflict engine. WiFi.setSleep(false) fixed it; I confirmed with a packet-gap histogram on the manager side.

The manager runs a small UDP listener for the beacons. The important design decision: timestamp at ingest, never trust the device clock, the ESP32s have no real-time clock, so their idea of "now" is useless for a system whose entire job is time-based prediction. The listener also watches per-beacon health and marks a track STALE when packets stop, so a dead beacon drops out of conflict computations instead of freezing in place as a ghost hazard.

PYTHON, BEACON UDP LISTENER

import json, socket, time

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(("0.0.0.0", 5005))
sock.settimeout(1.0)

last_seen = {}
STALE_AFTER_S = 2.0

while True:
    try:
        data, addr = sock.recvfrom(2048)
    except socket.timeout:
        data = None
    if data:
        msg = json.loads(data)
        msg["ts"] = time.time()          # stamp here - beacon has no RTC
        tracks.update(msg)
        last_seen[msg["id"]] = msg["ts"]
    now = time.time()
    for bid, ts in last_seen.items():
        tracks.mark_stale(bid, now - ts > STALE_AFTER_S)

How I verified: I pulled the USB power on one beacon mid-run and watched the dashboard: the track must flip to STALE within the timeout and disappear from the conflict engine's pair list. Before the stale logic, a powered-off beacon lingered forever as a phantom conflict.

The core of the system. For every live track I propagate a constant-velocity prediction forward and compute the time to closest point of approach (TCPA) between every pair, in horizontal and vertical components separately, two aircraft 10 m apart horizontally but 80 m apart vertically are not a conflict. If TCPA falls inside the lookahead window and both miss distances fall inside the separation bubble, the engine raises an alert before separation is actually lost. The pairwise check is O(n²), trivial at this scale; the real work is in the tunables (bubble radius, vertical separation, lookahead) and in alert stability.

PYTHON, TCPA BETWEEN TWO TRACKS

import math

def tcpa(a, b):
    """Time and miss distance of closest approach, two constant-velocity tracks.
    Closest approach is taken on the horizontal plane; vertical separation is
    read at that time, so a flyby that is far apart vertically never reads
    as a conflict."""
    rx, ry, rz = b.x - a.x, b.y - a.y, b.alt - a.alt
    vx, vy, vz = b.vx - a.vx, b.vy - a.vy, b.vz - a.vz
    vv = vx*vx + vy*vy + vz*vz
    if vv < 1e-6:
        # parallel tracks: closest approach is right now
        return 0.0, math.hypot(rx, ry), abs(rz)
    hv = vx*vx + vy*vy
    if hv < 1e-6:
        t = 0.0   # no horizontal motion: horizontal miss is current separation
    else:
        t = max(0.0, -(rx*vx + ry*vy) / hv)
    horiz = math.hypot(rx + vx*t, ry + vy*t)   # horizontal miss (m)
    vert = abs(rz + vz*t)                       # vertical miss at that time (m)
    return t, horiz, vert

def check_pair(a, b, lookahead_s=60, h_sep=100, v_sep=30):
    t, horiz, vert = tcpa(a, b)
    return t <= lookahead_s and horiz < h_sep and vert < v_sep

What can go wrong: with a single-sample threshold, alerts flap on and off as telemetry jitters near the bubble boundary. I require three consecutive violating predictions to raise an alert and a clear margin to release it (hysteresis), verified by walking a simulated track tangentially along the bubble edge and watching the queue stay quiet.

The geofence service holds 3D keep-in and keep-out volumes, cylinders and boxes with floor and ceiling altitudes, defined as plain JSON so the dashboard can create, move, and resize them live. Every incoming track update is tested against all active volumes and a violation alerts immediately. This mirrors the real use case: an exercise safety officer declaring a temporary restricted volume over a live-fire range, or shrinking the keep-in box as weather moves in.

JSON, A KEEP-OUT VOLUME

{
  "id": "RZ-ALPHA",
  "type": "keep-out",
  "shape": "cylinder",
  "center": {"lat": 43.7615, "lon": -79.4111},
  "radius_m": 250,
  "floor_m": 0,
  "ceil_m": 150
}

PYTHON, POINT-IN-VOLUME CHECK

def inside_cylinder(track, vol):
    d = haversine_m(track.lat, track.lon,
                    vol["center"]["lat"], vol["center"]["lon"])
    in_footprint = d <= vol["radius_m"]
    in_altband = vol["floor_m"] <= track.alt <= vol["ceil_m"]
    return in_footprint and in_altband

What can go wrong: naive degree arithmetic (treating lat/lon as flat) distorts distances badly at Toronto's latitude, a 250 m radius becomes an ellipse. I convert through haversine / a local ENU frame everywhere, and verified by checking that a circle drawn on the dashboard measures the same radius north-south as east-west.

The UI is a small FastAPI backend on the homelab pushing state over a WebSocket to a Leaflet page: tracks as moving markers, geofence volumes as shaded 2D footprints with floor and ceiling labels (Leaflet renders in 2D, so true 3D extrusion would need a separate rendering layer), and the alert queue down the side with approve/reject buttons. It runs at a 2 Hz push of current state only, the browser is a display, not a database.

PYTHON, DASHBOARD BACKEND (trimmed)

import asyncio
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def feed(ws: WebSocket):
    await ws.accept()
    while True:
        await ws.send_json({
            "tracks": tracks.snapshot(),      # current state only
            "volumes": geofences.snapshot(),
            "alerts": alert_queue.pending(),  # advisory queue, needs a human
        })
        await asyncio.sleep(0.5)              # 2 Hz UI refresh

What can go wrong: my first version shipped full track history every tick and the browser tab crawled after a few minutes. Sending only the latest state (and letting the front end keep short trails) fixed it, verified with the browser's performance panel over a long soak.

When the conflict engine fires, the manager proposes manoeuvres, an altitude change for one track, a hold, or a reroute around the predicted conflict point, but the queue is advisory, never autonomous: nothing reaches an aircraft until a human clicks approve. Approved resolutions go back to the SITL instance as MAVLink commands, and the dashboard then tracks whether the approved manoeuvre actually cleared the conflict; rejected proposals are logged with their outcomes for the test-case review.

PYTHON, SENDING AN APPROVED ALTITUDE RESOLUTION TO SITL

def send_guided_alt(conn, lat_deg, lon_deg, alt_m):
    # switch the vehicle to GUIDED with an explicit mode command, then verify;
    # MAV_CMD_NAV_GUIDED_ENABLE is for Guided-within-Auto, not a plain mode switch
    guided = conn.mode_mapping()["GUIDED"]
    conn.mav.command_long_send(
        conn.target_system, conn.target_component,
        mavutil.mavlink.MAV_CMD_DO_SET_MODE, 0,
        mavutil.mavlink.MAV_MODE_FLAG_CUSTOM_MODE_ENABLED,
        guided, 0, 0, 0, 0, 0)
    conn.mav.set_position_target_global_int_send(
        0, conn.target_system, conn.target_component,
        mavutil.mavlink.MAV_FRAME_GLOBAL_RELATIVE_ALT_INT,
        0b110111111000,      # type_mask: use lat/lon/alt, ignore vel/accel/yaw
        int(lat_deg * 1e7), int(lon_deg * 1e7), alt_m,
        0, 0, 0,             # velocity ignored
        0, 0, 0,             # acceleration ignored
        0, 0)                # yaw, yaw_rate ignored

How I verified: I watch for the COMMAND_ACK and then re-run the TCPA computation on the post-command track, an approved climb is only counted as a resolution when the predicted closest approach moves outside the bubble. Commands sent before the mode switch are silently ignored, which was my first "why didn't it climb?" bug, and I confirm the heartbeat's custom_mode reports GUIDED before the position target goes out.

Deconfliction logic is only worth trusting if it survives repeatable scenarios, so I scripted a set of conflict cases, crossing paths at a common altitude, converging altitudes on parallel tracks, and a deliberate geofence breach, each defined as mission files plus any volumes, runnable repeatedly with identical initial conditions. Every run produces a result sheet (the paper copies on the bench in the photo) that gets signed off with what the system detected, when it alerted, and how the resolution played out.

PYTHON, SCENARIO HARNESS (trimmed)

SCENARIOS = {
    "crossing-paths":    {"missions": {1: "missions/cross_a.txt",
                                       2: "missions/cross_b.txt"}},
    "altitude-converge": {"missions": {1: "missions/level_a.txt",
                                       2: "missions/climb_b.txt"}},
    "geofence-breach":   {"missions": {1: "missions/straight_a.txt"},
                          "volumes": ["volumes/rz_alpha.json"]},
}

def run(name, repeat=10):
    for i in range(repeat):
        reset_sitl()                        # wipe state between runs
        load_scenario(SCENARIOS[name])      # wp load per drone, push volumes
        t0 = time.time()
        result = watch_until_resolved(timeout=300)
        sheet.log(name, i,
                  time_to_alert_s=result.alert_t - t0,
                  margin_s=result.loss_t - result.alert_t,
                  cleared=result.cleared)

What can go wrong: SITL leaks state between runs (leftover waypoints, changed parameters), which makes "repeatable" runs anything but. The harness restarts instances and wipes (sim_vehicle.py -w) before every run; I confirmed repeatability by diffing the mission state at t0 across runs.

The manager does not live alone: its tracks and alerts publish into the multi-domain common operating picture over the same MQTT bus the other defence nodes use, signed the same way, so air tracks become one more layer alongside the passive radar, acoustic, and ground sensors, and the COP's human-in-the-loop queue is the same one this manager writes to.

PYTHON, PUBLISHING ALERTS TO THE COP BUS

import json
import paho.mqtt.client as mqtt

bus = mqtt.Client(client_id="airspace-manager")
# paho queues outbound publishes in memory while disconnected; this is NOT a
# disk-backed store-and-forward queue and does not survive a process restart
bus.max_queued_messages_set(100)
bus.reconnect_delay_set(min_delay=1, max_delay=30)
bus.connect("mqtt.homelab.local", 1883)

def on_alert(alert):
    payload = json.dumps(alert, separators=(",", ":"))
    bus.publish(
        "cop/air/alerts",
        json.dumps({"msg": payload, "sig": sign_ed25519(payload)}),
        qos=1)          # at-least-once; survives Wi-Fi blips while connected

What can go wrong: at QoS 0 an alert can vanish on a Wi-Fi hiccup, the worst kind of failure for an advisory system, because the operator never knows it existed. QoS 1 plus paho's in-memory queued messages while the broker is unreachable covers brief outages, with a caveat: that queue is in memory only, so a broker outage that outlasts the process loses alerts. A disk-backed outbox like the COP build's is the hardened path; I verify by killing the broker mid-scenario and confirming the alert arrives on restart.