Claude (homelab agent)andClaude Opus 5.5
56eb5af0ff
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
96 lines
4.0 KiB
Python
96 lines
4.0 KiB
Python
"""ADS-B context for airband recordings (snapshot of aircraft near the receiver) and the airband frequency test."""
|
|
import json
|
|
import time
|
|
|
|
from radio.config import ADSB_FILE
|
|
from radio.util import write_json_compact
|
|
|
|
# ---- ADS-B context for airband recordings -------------------------------------------------
|
|
# Snapshot of aircraft near the receiver when an airband recording is classified (seconds after it
|
|
# ended). Source order: local readsb (future dedicated 1090 MHz RTL-SDR) -> adsb.lol public API.
|
|
# At most one remote request per minute (cache), short timeouts; failures only leave a note.
|
|
# NOTE 2026-10-20: adsb.lol / adsb.fi / OpenSky have NO coverage over St. Petersburg (0 aircraft
|
|
# while London returns dozens), so until a local receiver exists the list will usually be empty.
|
|
RX_LAT, RX_LON = 0.0, 0.0 # receiver (home area) -- adjust if the site is elsewhere
|
|
ULLI_LAT, ULLI_LON = 59.8003, 30.2625 # Pulkovo
|
|
ADSB_RADIUS_NM = 60
|
|
LOCAL_ADSB = ["/run/readsb/aircraft.json", "http://127.0.0.1:8080/data/aircraft.json"]
|
|
REMOTE_ADSB = f"https://api.adsb.lol/v2/point/{RX_LAT}/{RX_LON}/{ADSB_RADIUS_NM}"
|
|
_adsb_cache = {"t": 0.0, "data": None}
|
|
|
|
|
|
def _nm(lat1, lon1, lat2, lon2):
|
|
import math
|
|
p1, p2 = math.radians(lat1), math.radians(lat2)
|
|
dp, dl = p2 - p1, math.radians(lon2 - lon1)
|
|
a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
|
|
d = 2 * 3440.065 * math.asin(min(1.0, math.sqrt(a)))
|
|
y = math.sin(dl) * math.cos(p2)
|
|
x = math.cos(p1) * math.sin(p2) - math.sin(p1) * math.cos(p2) * math.cos(dl)
|
|
return d, (math.degrees(math.atan2(y, x)) + 360) % 360
|
|
|
|
|
|
def _fetch_json(url, timeout=4):
|
|
import urllib.request
|
|
if url.startswith("/"):
|
|
with open(url, encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
req = urllib.request.Request(url, headers={"User-Agent": "home-sdr-scanner/1"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
return json.loads(r.read(2_000_000).decode("utf-8"))
|
|
|
|
|
|
def adsb_snapshot():
|
|
now = time.time()
|
|
if _adsb_cache["data"] is not None and now - _adsb_cache["t"] < 60:
|
|
return _adsb_cache["data"]
|
|
src, raw, err = None, None, None
|
|
for u in LOCAL_ADSB + [REMOTE_ADSB]:
|
|
try:
|
|
raw = _fetch_json(u)
|
|
src = "readsb" if u in LOCAL_ADSB else "adsb.lol"
|
|
break
|
|
except Exception as e: # noqa: BLE001
|
|
err = f"{type(e).__name__}"
|
|
continue
|
|
out = {"t": int(now), "src": src, "list": []}
|
|
if raw is None:
|
|
out["err"] = err or "unavailable"
|
|
else:
|
|
for a in raw.get("ac", raw.get("aircraft", [])) or []:
|
|
lat, lon = a.get("lat"), a.get("lon")
|
|
if lat is None or lon is None:
|
|
continue
|
|
d, b = _nm(RX_LAT, RX_LON, lat, lon)
|
|
if d > ADSB_RADIUS_NM:
|
|
continue
|
|
alt = a.get("alt_baro")
|
|
alt = 0 if alt == "ground" else alt
|
|
dp, _ = _nm(ULLI_LAT, ULLI_LON, lat, lon)
|
|
sq = str(a.get("squawk") or "")
|
|
out["list"].append({
|
|
"cs": (a.get("flight") or "").strip() or a.get("r") or a.get("hex"),
|
|
"hex": a.get("hex"), "sq": sq or None, "alt": alt, "gs": a.get("gs"),
|
|
"trk": a.get("track"), "d": round(d, 1), "b": int(b), "cat": a.get("category"),
|
|
"t": a.get("t"),
|
|
"emg": sq in ("7500", "7600", "7700") or (a.get("emergency") not in (None, "none")),
|
|
"ulli": bool(alt is not None and isinstance(alt, (int, float)) and alt < 10000 and dp < 25),
|
|
})
|
|
out["list"].sort(key=lambda x: x["d"])
|
|
out["n"] = len(out["list"])
|
|
out["list"] = out["list"][:20]
|
|
_adsb_cache.update(t=now, data=out)
|
|
try:
|
|
write_json_compact(ADSB_FILE, out)
|
|
except OSError:
|
|
pass
|
|
return out
|
|
|
|
|
|
def is_avia(name):
|
|
try:
|
|
f = int(name.rsplit("_", 1)[1][:-4])
|
|
except (ValueError, IndexError):
|
|
return False
|
|
return 108_000_000 <= f < 137_000_000
|