Allarme prossimita' ADBS

Uno degli incubi mentre volo con il drone e' di imbattermi in Pegaso 1, l'elisoccorso della Toscana

La quota di volo di Pegaso minima (tranne quando decolla ed atterra) sembra essere di 1000 piedi (circa 300 m) quindi ben al di sopra dei 120 m di quota massima dei droni 

l'idea e' comunque di avere una ground station con allarme di prossimita' usando un dongle RTL-SDR impostando la propria posizione ed un raggio di 2 miglia nautiche

su Debian 13 si usa (scaricabile tramite apt)  

dump1090-fa    --net   --net-sbs-port 30003   --net-ro-port 30002   --net-http-port 8080   --lat 43.7696 --lon 11.2558  

il comando soprastante 

usando il programma Python (grazie Claude) si leggono i dati in uscita dump1009-fa e si selezionano solo gli aeromobili in 3 miglia nautiche di distanza e separazione verticale inferiore a 1500 piedi

 python3 adsb_proximity_alarm.py sbs --host 127.0.0.1 --port 30003     --lat 43.7696 --lon 11.2558 --elev-ft 150     --radius-nm 3 --vsep-ft 1500 -v 

 

 

#!/usr/bin/env python3

"""

adsb_proximity_alarm.py

Proximity alarm for ADS-B traffic received via an RTL-SDR + dump1090

(dump1090-fa, dump1090-mutability, readsb, or the dump1090 fork bundled

with Stratux).

It watches live aircraft positions and raises an alert when any aircraft

comes within a configurable horizontal distance AND vertical separation

of a reference point (e.g. your ground station / launch site).

Two input modes are supported, pick whichever your setup exposes:

1. SBS / BaseStation feed (raw TCP text feed, classic dump1090 port 30003)

Most dump1090 builds expose this. Stratux's internal dump1090 usually

does too, though it may not be reachable outside localhost unless you

tunnel it (e.g. `ssh -L 30003:localhost:30003 pi@stratux.local`).

2. aircraft.json polling (HTTP), as served by dump1090-fa/readsb/tar1090,

typically at http://<host>/dump1090-fa/data/aircraft.json or similar.

Check your install; the path varies by distro/build.

Usage examples

--------------

SBS mode, ground station at 43.7696N 11.2558E (Florence area), alarm inside

3 NM / 1500 ft:

python3 adsb_proximity_alarm.py sbs \\

--host 127.0.0.1 --port 30003 \\

--lat 43.7696 --lon 11.2558 --elev-ft 150 \\

--radius-nm 3 --vsep-ft 1500

JSON polling mode:

python3 adsb_proximity_alarm.py json \\

--url http://127.0.0.1/dump1090-fa/data/aircraft.json \\

--lat 43.7696 --lon 11.2558 --radius-nm 3 --vsep-ft 1500

No extra dependencies beyond the standard library and `requests`

(only needed for JSON mode: pip install requests).

"""

import argparse

import math

import socket

import sys

import time

import urllib.request

import json as jsonlib

from dataclasses import dataclass, field

from datetime import datetime, timezone

# --------------------------------------------------------------------------

# Geometry helpers

# --------------------------------------------------------------------------

EARTH_RADIUS_NM = 3440.065 # mean earth radius in nautical miles

def haversine_nm(lat1, lon1, lat2, lon2):

"""Great-circle distance in nautical miles."""

phi1, phi2 = math.radians(lat1), math.radians(lat2)

dphi = math.radians(lat2 - lat1)

dlambda = math.radians(lon2 - lon1)

a = (math.sin(dphi / 2) ** 2

+ math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2)

return 2 * EARTH_RADIUS_NM * math.asin(math.sqrt(a))

def bearing_deg(lat1, lon1, lat2, lon2):

"""Initial bearing from point 1 to point 2, in degrees (0=N, 90=E)."""

phi1, phi2 = math.radians(lat1), math.radians(lat2)

dlambda = math.radians(lon2 - lon1)

y = math.sin(dlambda) * math.cos(phi2)

x = (math.cos(phi1) * math.sin(phi2)

- math.sin(phi1) * math.cos(phi2) * math.cos(dlambda))

return (math.degrees(math.atan2(y, x)) + 360) % 360

def compass(deg):

dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]

return dirs[int((deg + 22.5) // 45) % 8]

# --------------------------------------------------------------------------

# Aircraft state

# --------------------------------------------------------------------------

@dataclass

class Aircraft:

icao: str

callsign: str = ""

lat: float = None

lon: float = None

alt_ft: float = None

last_seen: float = field(default_factory=time.monotonic)

last_alert: float = 0.0

def has_position(self):

return self.lat is not None and self.lon is not None

class AircraftTable:

"""Keeps track of recently seen aircraft and prunes stale ones."""

def __init__(self, stale_after_s=60):

self.aircraft = {}

self.stale_after_s = stale_after_s

def update(self, icao, callsign=None, lat=None, lon=None, alt_ft=None):

ac = self.aircraft.get(icao)

if ac is None:

ac = Aircraft(icao=icao)

self.aircraft[icao] = ac

if callsign:

ac.callsign = callsign.strip()

if lat is not None:

ac.lat = lat

if lon is not None:

ac.lon = lon

if alt_ft is not None:

ac.alt_ft = alt_ft

ac.last_seen = time.monotonic()

return ac

def prune(self):

now = time.monotonic()

dead = [icao for icao, ac in self.aircraft.items()

if now - ac.last_seen > self.stale_after_s]

for icao in dead:

del self.aircraft[icao]

# --------------------------------------------------------------------------

# Alerting

# --------------------------------------------------------------------------

RED = "\033[91m"

YELLOW = "\033[93m"

GREEN = "\033[92m"

RESET = "\033[0m"

def alert(ac: Aircraft, ref_lat, ref_lon, dist_nm, vsep_ft, bell=True):

brg = bearing_deg(ref_lat, ref_lon, ac.lat, ac.lon)

ts = datetime.now(timezone.utc).strftime("%H:%M:%S")

tag = ac.callsign if ac.callsign else ac.icao

msg = (f"{RED}[ALERT {ts}Z]{RESET} {tag} (ICAO {ac.icao}) "

f"{dist_nm:.1f} NM {compass(brg)} ({brg:5.1f}°), "

f"alt {ac.alt_ft:.0f} ft, vertical sep {vsep_ft:.0f} ft")

print(msg)

if bell:

sys.stdout.write("\a")

sys.stdout.flush()

def status_line(ac: Aircraft, dist_nm, vsep_ft):

tag = ac.callsign if ac.callsign else ac.icao

print(f"{YELLOW} near{RESET} {tag:8s} {dist_nm:5.1f} NM "

f"alt {ac.alt_ft:6.0f} ft vsep {vsep_ft:6.0f} ft")

# --------------------------------------------------------------------------

# Core proximity check

# --------------------------------------------------------------------------

def check_proximity(ac: Aircraft, args):

if not ac.has_position() or ac.alt_ft is None:

return

dist_nm = haversine_nm(args.lat, args.lon, ac.lat, ac.lon)

vsep_ft = abs(ac.alt_ft - args.elev_ft)

if dist_nm <= args.radius_nm and vsep_ft <= args.vsep_ft:

now = time.monotonic()

if now - ac.last_alert >= args.cooldown:

alert(ac, args.lat, args.lon, dist_nm, vsep_ft, bell=not args.no_bell)

ac.last_alert = now

elif args.verbose and dist_nm <= args.radius_nm * 2.0:

# Show near-miss traffic even if outside the hard alert ring,

# useful for situational awareness.

status_line(ac, dist_nm, vsep_ft)

# --------------------------------------------------------------------------

# SBS (BaseStation) feed parser

# --------------------------------------------------------------------------

#

# Format reference (dump1090 SBS output), comma separated fields:

# MSG,type,session,aircraft,hex,flight,date,time,date,time,

# callsign,altitude,gspeed,track,lat,lon,vrate,squawk,alert,emergency,spi,onground

#

# Relevant message types:

# MSG,1 -> callsign (field 10)

# MSG,3 -> airborne position: altitude(11), lat(14), lon(15)

# MSG,4 -> velocity

# MSG,5/6/7/8 -> various, mostly ignored here

def run_sbs(args):

table = AircraftTable()

print(f"Connecting to SBS feed at {args.host}:{args.port} ...")

while True:

try:

with socket.create_connection((args.host, args.port), timeout=10) as sock:

print(f"{GREEN}Connected.{RESET} Watching for traffic within "

f"{args.radius_nm} NM / {args.vsep_ft} ft of "

f"{args.lat:.4f},{args.lon:.4f} (elev {args.elev_ft} ft)...")

buf = b""

sock.settimeout(5)

last_prune = time.monotonic()

while True:

try:

chunk = sock.recv(4096)

except socket.timeout:

chunk = b""

if chunk == b"" and time.monotonic() - last_prune > 30:

# no data for a while but connection alive; keep looping

pass

buf += chunk

while b"\n" in buf:

line, buf = buf.split(b"\n", 1)

line = line.decode(errors="ignore").strip()

if line:

handle_sbs_line(line, table, args)

if time.monotonic() - last_prune > 15:

table.prune()

last_prune = time.monotonic()

except (ConnectionRefusedError, OSError) as e:

print(f"Connection error ({e}); retrying in 5s...")

time.sleep(5)

def handle_sbs_line(line, table, args):

fields = line.split(",")

if len(fields) < 2 or fields[0] != "MSG":

return

msg_type = fields[1]

if len(fields) < 5:

return

icao = fields[4].strip()

if not icao:

return

callsign = None

lat = lon = alt_ft = None

try:

if msg_type == "1" and len(fields) > 10:

callsign = fields[10]

elif msg_type == "3" and len(fields) > 15:

alt_ft = float(fields[11]) if fields[11] else None

lat = float(fields[14]) if fields[14] else None

lon = float(fields[15]) if fields[15] else None

elif msg_type in ("5", "6", "7", "8") and len(fields) > 11 and fields[11]:

alt_ft = float(fields[11])

except ValueError:

return

ac = table.update(icao, callsign=callsign, lat=lat, lon=lon, alt_ft=alt_ft)

check_proximity(ac, args)

# --------------------------------------------------------------------------

# aircraft.json polling

# --------------------------------------------------------------------------

def run_json(args):

table = AircraftTable()

print(f"Polling {args.url} every {args.poll_interval}s ...")

print(f"Watching for traffic within {args.radius_nm} NM / {args.vsep_ft} ft of "

f"{args.lat:.4f},{args.lon:.4f} (elev {args.elev_ft} ft)...")

last_prune = time.monotonic()

while True:

try:

with urllib.request.urlopen(args.url, timeout=5) as resp:

data = jsonlib.loads(resp.read())

except Exception as e:

print(f"Fetch error ({e}); retrying in {args.poll_interval}s...")

time.sleep(args.poll_interval)

continue

aircraft_list = data.get("aircraft", data if isinstance(data, list) else [])

for entry in aircraft_list:

icao = (entry.get("hex") or entry.get("icao") or "").strip()

if not icao:

continue

callsign = entry.get("flight") or entry.get("callsign")

lat = entry.get("lat")

lon = entry.get("lon")

alt_ft = entry.get("alt_baro", entry.get("altitude"))

try:

alt_ft = float(alt_ft) if alt_ft not in (None, "ground") else None

except (TypeError, ValueError):

alt_ft = None

ac = table.update(icao, callsign=callsign, lat=lat, lon=lon, alt_ft=alt_ft)

check_proximity(ac, args)

if time.monotonic() - last_prune > 15:

table.prune()

last_prune = time.monotonic()

time.sleep(args.poll_interval)

# --------------------------------------------------------------------------

# CLI

# --------------------------------------------------------------------------

def build_parser():

# Shared options live on a "parent" parser so they can be passed either

# before or after the sbs/json subcommand, e.g. both of these work:

# adsb_proximity_alarm.py --lat 43.7 --lon 11.2 sbs --port 30003

# adsb_proximity_alarm.py sbs --port 30003 --lat 43.7 --lon 11.2

common = argparse.ArgumentParser(add_help=False)

common.add_argument("--lat", type=float, required=True,

help="Reference latitude (your ground station / launch site)")

common.add_argument("--lon", type=float, required=True,

help="Reference longitude")

common.add_argument("--elev-ft", type=float, default=0.0,

help="Reference point elevation, feet MSL (default 0)")

common.add_argument("--radius-nm", type=float, default=3.0,

help="Horizontal alert radius in nautical miles (default 3)")

common.add_argument("--vsep-ft", type=float, default=1500.0,

help="Vertical separation threshold in feet (default 1500)")

common.add_argument("--cooldown", type=float, default=30.0,

help="Seconds before re-alerting on the same aircraft (default 30)")

common.add_argument("--no-bell", action="store_true",

help="Disable terminal bell on alert")

common.add_argument("-v", "--verbose", action="store_true",

help="Also print traffic within 2x the alert radius")

p = argparse.ArgumentParser(

description="ADS-B proximity alarm using RTL-SDR / dump1090 feeds")

sub = p.add_subparsers(dest="mode", required=True)

sbs = sub.add_parser("sbs", parents=[common],

help="Connect to raw SBS/BaseStation TCP feed")

sbs.add_argument("--host", default="127.0.0.1")

sbs.add_argument("--port", type=int, default=30003)

js = sub.add_parser("json", parents=[common],

help="Poll an aircraft.json HTTP endpoint")

js.add_argument("--url", required=True,

help="Full URL to aircraft.json")

js.add_argument("--poll-interval", type=float, default=1.0,

help="Seconds between polls (default 1.0)")

return p

def main():

args = build_parser().parse_args()

try:

if args.mode == "sbs":

run_sbs(args)

else:

run_json(args)

except KeyboardInterrupt:

print("\nStopped.")

if __name__ == "__main__":

main()