import json import os import signal import subprocess import threading from pathlib import Path from typing import Any, Dict, List, Optional from internal.logger import create_logger LOGGER = create_logger(__name__) # AIS-catcher streams one JSON object per received message on stdout rather # than writing a periodic snapshot file (dump1090's approach above) — so the # current-vessel snapshot lives in memory, keyed by mmsi, kept warm by a # background reader thread for as long as the subprocess we started is # alive. Unlike the pgid-file state, this does NOT survive this API # process restarting independently of its subprocess — acceptable since # nothing here does that today. _ais_vessels: Dict[str, Dict[str, Any]] = {} _ais_lock = threading.Lock() # Which RTL-SDR index op25 holds is NOT fixed — on radio-box op25 had index 1 # and index 0 was free, so "op25 is always 0" was wrong. rtlsdr can't open a # dongle another process has claimed, and the decoders exit within ~50ms when # that happens, so start() tries each index and keeps the first that stays up. # op25 is never disturbed: a failed claim doesn't touch its dongle. MAX_SDR_INDEX = 4 _STARTUP_GRACE_S = 2.0 _PGID_FILE = "/tmp/secondary_sdr.pgid" _MODE_FILE = "/tmp/secondary_sdr.mode" ADSB_JSON_DIR = Path("/tmp/adsb") # The live decoder handle. poll() is the only reliable liveness check: a # decoder that dies on startup (e.g. SDR busy) stays an unreaped zombie, and # killpg(pgid, 0) still succeeds on a zombie — status reported "running". _proc: Optional[subprocess.Popen] = None def _save_state(pgid: int, mode: str) -> None: Path(_PGID_FILE).write_text(str(pgid)) Path(_MODE_FILE).write_text(mode) def _read_pgid() -> Optional[int]: try: return int(Path(_PGID_FILE).read_text().strip()) except Exception: return None def _read_mode() -> Optional[str]: try: return Path(_MODE_FILE).read_text().strip() except Exception: return None def is_running() -> bool: if _proc is not None: return _proc.poll() is None pgid = _read_pgid() if pgid is None: return False try: os.killpg(pgid, 0) return True except OSError: return False def _adsb_command(index: int) -> List[str]: ADSB_JSON_DIR.mkdir(parents=True, exist_ok=True) return [ "/opt/readsb/readsb", "--net", "--device-type", "rtlsdr", "--device", str(index), "--write-json", str(ADSB_JSON_DIR), "--write-json-every", "1", ] def _ais_command(index: int) -> List[str]: return [ "/opt/AIS-catcher/build/AIS-catcher", f"-d:{index}", # "-d " would select by serial, not index "-o", "JSON", ] def _ais_reader(proc: subprocess.Popen) -> None: """ Consume AIS-catcher's stdout, one JSON message per line, and keep the latest report per mmsi. Field names (mmsi/lat/lon/speed/course or heading/shipname or name) are believed correct from AIS-catcher's published JSON output docs but UNVERIFIED against a real capture in this session — same caveat as dump1090's aircraft.json mapping. Malformed/partial lines (e.g. static-data-only messages with no position) are skipped rather than raising, since dropping one line must never kill the reader thread. """ if not proc.stdout: return for line in proc.stdout: try: msg = json.loads(line) except Exception: continue mmsi = msg.get("mmsi") if not mmsi: continue # AIS-catcher emits separate message TYPES per mmsi — static data # (name, no position) and position reports (lat/lon, no name) arrive # as distinct lines. Merge onto the existing entry, only overwriting # a field the new message actually carries, so a position-only # report doesn't blank out a name learned from an earlier message. name = (msg.get("shipname") or msg.get("name") or "").strip() or None heading = msg.get("heading") if msg.get("heading") is not None else msg.get("course") updates = { "mmsi": str(mmsi), "name": name, "lat": msg.get("lat"), "lon": msg.get("lon"), "speed_kt": msg.get("speed"), "heading_deg": heading, } with _ais_lock: existing = _ais_vessels.get(str(mmsi), {}) for key, value in updates.items(): if value is not None: existing[key] = value _ais_vessels[str(mmsi)] = existing def start(mode: str) -> bool: global _proc if is_running(): stop() if mode == "adsb": build = _adsb_command elif mode == "ais": build = _ais_command with _ais_lock: _ais_vessels.clear() else: raise ValueError(f"Unknown secondary SDR mode: {mode!r}") needs_stdout = mode == "ais" for index in range(MAX_SDR_INDEX): try: proc = subprocess.Popen( build(index), preexec_fn=os.setsid, stdout=subprocess.PIPE if needs_stdout else None, text=True if needs_stdout else None, bufsize=1 if needs_stdout else -1, ) except Exception as e: LOGGER.error(f"Failed to start secondary SDR decoder mode={mode!r}: {e}") return False try: proc.wait(timeout=_STARTUP_GRACE_S) LOGGER.info(f"Secondary SDR decoder mode={mode!r} could not use SDR index {index}, trying next") continue except subprocess.TimeoutExpired: pass if needs_stdout: threading.Thread(target=_ais_reader, args=(proc,), daemon=True).start() _proc = proc _save_state(proc.pid, mode) LOGGER.info(f"Started secondary SDR decoder mode={mode!r} on SDR index {index} pid={proc.pid}") return True LOGGER.error(f"Secondary SDR decoder mode={mode!r}: no free SDR found (op25 holds one; is a second plugged in?)") return False def stop() -> bool: global _proc pgid = _read_pgid() if pgid is None: return True try: os.killpg(pgid, signal.SIGTERM) except OSError: pass if _proc is not None: try: _proc.wait(timeout=5) except subprocess.TimeoutExpired: pass _proc = None try: os.remove(_PGID_FILE) except OSError: pass try: os.remove(_MODE_FILE) except OSError: pass return True def status() -> Dict[str, Any]: running = is_running() return { "status": "running" if running else "stopped", "mode": _read_mode() if running else None, } def _altitude(a: Dict[str, Any]) -> Optional[int]: alt = a.get("alt_baro", a.get("altitude")) if alt == "ground": return 0 return alt if isinstance(alt, (int, float)) else None def _read_adsb_snapshot() -> List[Dict[str, Any]]: """ Map readsb's aircraft.json (--write-json output) to the server's telemetry schema. readsb uses the dump1090-fa field names: alt_baro (int, or the string "ground"), gs, track. Older dump1090 forks used altitude/speed, kept as a fallback. """ path = ADSB_JSON_DIR / "aircraft.json" try: raw = json.loads(path.read_text()) except Exception: return [] out = [] for a in raw.get("aircraft", []): icao = a.get("hex") if not icao: continue out.append({ "icao": icao.upper(), "callsign": (a.get("flight") or "").strip() or None, "lat": a.get("lat"), "lon": a.get("lon"), "altitude_ft": _altitude(a), "ground_speed_kt": a.get("gs", a.get("speed")), "track_deg": a.get("track"), }) return out def data() -> Dict[str, Any]: mode = _read_mode() if mode == "adsb": return {"mode": mode, "aircraft": _read_adsb_snapshot()} if mode == "ais": with _ais_lock: vessels = list(_ais_vessels.values()) return {"mode": mode, "vessels": vessels} return {"mode": mode, "aircraft": [], "vessels": []}