Author SHA1 Message Date
Logan CusanoandClaude Sonnet 5 52f31bbcc0 Wire AIS end to end: AIS-catcher support in secondary-sdr container
CI / lint (push) Successful in 5s
CI / lint (pull_request) Successful in 5s
CI / test (push) Successful in 38s
CI / test (pull_request) Successful in 38s
node-26#9. decoder_control.py's AIS mode now actually launches AIS-catcher
instead of 400ing: a background thread reads its stdout JSON stream
(one message per line) and keeps a live in-memory snapshot keyed by mmsi,
since AIS-catcher streams rather than writing a periodic file the way
dump1090's --write-json does. Messages are merged onto the existing entry
per mmsi rather than replacing it, since AIS-catcher emits static data
(name) and position reports (lat/lon) as separate message types — a naive
overwrite would blank the name back to null on every position-only update
(caught by a standalone unit check against a fake stdout stream before
this fix, not by pytest — this container has no test suite yet).

edge-node's telemetry_uplink_loop now posts non-empty vessel snapshots to
the new /telemetry/ais the same way it already does for aircraft.

UNVERIFIED against real hardware/binary in this session, same caveat as
the ADS-B commit — AIS-catcher's JSON field names are believed correct
from its docs, not confirmed against a real capture.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-20 19:10:30 -04:00
Logan CusanoandClaude Sonnet 5 b2e3804dc3 Wire ADS-B end to end: secondary-sdr container running dump1090
node-26#9. New secondary-sdr-container claims the node's SECOND physical
SDR (RTL-SDR index 1 — op25 always claims index 0; no serial-based binding
yet, same gap op25 itself has). Its control API (start/stop/status/data,
mirroring op25_controller.py) launches dump1090 in adsb mode and exposes
the decoded aircraft.json snapshot; AIS mode 400s until it's wired next.

edge-node: on_config_push starts/stops it when secondary_sdr_mode changes,
lifespan resumes it after a restart if already configured, and a new
telemetry_uplink_loop polls its /secondary/data every 10s and POSTs
non-empty snapshots to C2's new /telemetry/adsb (same bearer-key pattern
call_recorder.py already uses for audio upload).

UNVERIFIED: this container has not been built or run against real hardware
in this session (sandboxed authoring machine, no docker) — dump1090's
--write-json field names are believed correct from its docs but not
confirmed against a real capture. Build + hardware smoke test before this
reaches a real node.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-20 19:10:24 -04:00
Logan CusanoandClaude Sonnet 5 8f5fca8757 Add second-SDR plumbing: secondary_sdr_mode config + hardware reporting
Adds a per-node secondary_sdr_mode config field (none|adsb|ais|op25_2),
applied the same way as hardware_preset/ppm_override so it survives system
reassignment. op25-container gets a GET /devices endpoint that counts
connected SDRs via lsusb; the edge-node checkin now reports sdr_count and
secondary_sdr_mode up to the server, the first node-initiated hardware
report (everything else was C2 pushing config down). Tracked as node-26#9.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-20 19:10:17 -04:00
19 changed files with 571 additions and 1 deletions
+7
View File
@@ -111,6 +111,13 @@ OP25_TERMINAL_URL=http://localhost:8081
# for local development off a real node; leave false everywhere else. # for local development off a real node; leave false everywhere else.
OP25_DEBUG_EXPOSE=false OP25_DEBUG_EXPOSE=false
# Secondary SDR container (node-26#9) — only matters if a second physical SDR
# is present and secondary_sdr_mode is set to adsb|ais via the edge dashboard
# or C2. Usually no need to change.
SECONDARY_SDR_API_URL=http://localhost:8002
# Same caveat as OP25_DEBUG_EXPOSE — debugging aid only, leave false.
SECONDARY_SDR_DEBUG_EXPOSE=false
# --- Local dashboard / API login --------------------------------------------- # --- Local dashboard / API login ---------------------------------------------
# Protects the node's local dashboard (port 80) and JSON API. The node is # Protects the node's local dashboard (port 80) and JSON API. The node is
# reachable by anyone on whatever site's LAN it's deployed to, so this MUST be # reachable by anyone on whatever site's LAN it's deployed to, so this MUST be
+15
View File
@@ -32,6 +32,21 @@ services:
depends_on: depends_on:
- icecast - icecast
# Claims the node's SECOND physical SDR (op25 always claims the first).
# Only useful if secondary_sdr_mode is set to adsb|ais via the edge-node
# config; otherwise it just sits idle answering /secondary/status. See
# node-26#9. Same network/device access as op25 for the same reason: it
# needs the raw USB device, not a virtualized one.
secondary-sdr:
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/secondary-sdr:latest
build: ./secondary-sdr-container
restart: unless-stopped
privileged: true
network_mode: host
env_file: .env
volumes:
- /dev:/dev
edge-node: edge-node:
image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/edge-node:latest image: ${IMAGE_REGISTRY:-git.vpn.cusano.net}/${DOCKER_ORG:-logan}/${DOCKER_REPO:-node-26}/edge-node:latest
build: ./drb-edge-node build: ./drb-edge-node
+3
View File
@@ -136,6 +136,9 @@ class Settings(BaseSettings):
op25_api_url: str = "http://localhost:8001" op25_api_url: str = "http://localhost:8001"
op25_terminal_url: str = "http://localhost:8081" op25_terminal_url: str = "http://localhost:8081"
# Secondary SDR container (node-26#9) — ADS-B / AIS on a second SDR
secondary_sdr_api_url: str = "http://localhost:8002"
# Paths (volume mounts) # Paths (volume mounts)
config_path: str = "/configs" config_path: str = "/configs"
recordings_path: str = "/recordings" recordings_path: str = "/recordings"
@@ -213,7 +213,9 @@ class MQTTManager:
async def _publish_checkin(self): async def _publish_checkin(self):
from app.internal.discord_radio import radio_bot from app.internal.discord_radio import radio_bot
from app.internal.config_manager import load_node_config from app.internal.config_manager import load_node_config
from app.internal.op25_client import op25_client
config = load_node_config() config = load_node_config()
devices = await op25_client.devices()
payload = { payload = {
"node_id": settings.node_id, "node_id": settings.node_id,
"name": settings.node_name, "name": settings.node_name,
@@ -225,7 +227,12 @@ class MQTTManager:
"is_overridden": config.override_system_id is not None and config.node_type != "portable", "is_overridden": config.override_system_id is not None and config.node_type != "portable",
"override_system_id": config.override_system_id, "override_system_id": config.override_system_id,
"enforce_override_timeout": config.enforce_override_timeout, "enforce_override_timeout": config.enforce_override_timeout,
"secondary_sdr_mode": config.secondary_sdr_mode,
} }
# Best-effort — the first node-initiated hardware-report field. Omit
# rather than guess if op25's control API is unreachable.
if devices is not None:
payload["sdr_count"] = devices.get("count")
self._publish(self._t_checkin, payload, qos=1) self._publish(self._t_checkin, payload, qos=1)
def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False): def _publish(self, topic: str, payload: dict, qos: int = 0, retain: bool = False):
+10
View File
@@ -65,6 +65,16 @@ class OP25Client:
logger.error(f"OP25 status failed: {e}") logger.error(f"OP25 status failed: {e}")
return None return None
async def devices(self) -> Optional[Dict[str, Any]]:
try:
async with httpx.AsyncClient(timeout=5) as client:
r = await client.get(f"{self.api_url}/op25/devices")
r.raise_for_status()
return r.json()
except Exception as e:
logger.error(f"OP25 device enumeration failed: {e}")
return None
async def generate_config(self, config: Dict[str, Any]) -> bool: async def generate_config(self, config: Dict[str, Any]) -> bool:
try: try:
async with httpx.AsyncClient(timeout=10) as client: async with httpx.AsyncClient(timeout=10) as client:
@@ -0,0 +1,59 @@
import httpx
from typing import Any, Dict, Optional
from app.config import settings
from app.internal.logger import logger
class SecondarySdrClient:
"""Talks to the secondary-sdr-container (node-26#9) over its control API.
Mirrors op25_client.py's shape on purpose — same failure handling (log
and return None/False rather than raise), since this container is
optional and its absence must never break the primary op25 radio path.
"""
def __init__(self):
self.api_url = settings.secondary_sdr_api_url
async def start(self, mode: str) -> bool:
try:
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{self.api_url}/secondary/start", json={"mode": mode})
r.raise_for_status()
return True
except Exception as e:
logger.error(f"Secondary SDR start (mode={mode!r}) failed: {e}")
return False
async def stop(self) -> bool:
try:
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{self.api_url}/secondary/stop")
r.raise_for_status()
return True
except Exception as e:
logger.error(f"Secondary SDR stop failed: {e}")
return False
async def status(self) -> Optional[Dict[str, Any]]:
try:
async with httpx.AsyncClient(timeout=5) as client:
r = await client.get(f"{self.api_url}/secondary/status")
r.raise_for_status()
return r.json()
except Exception as e:
logger.error(f"Secondary SDR status failed: {e}")
return None
async def data(self) -> Optional[Dict[str, Any]]:
try:
async with httpx.AsyncClient(timeout=5) as client:
r = await client.get(f"{self.api_url}/secondary/data")
r.raise_for_status()
return r.json()
except Exception as e:
logger.error(f"Secondary SDR data fetch failed: {e}")
return None
secondary_sdr_client = SecondarySdrClient()
@@ -0,0 +1,44 @@
import asyncio
import httpx
from app.config import settings
from app.internal import credentials
from app.internal.config_manager import load_node_config
from app.internal.logger import logger
from app.internal.secondary_sdr_client import secondary_sdr_client
# How often the second-SDR decoder's current snapshot is forwarded to C2
# (node-26#9). This is a live-map overlay, not a flight/vessel history, so
# there is no backlog/retry on a missed tick — the next one supersedes it.
UPLINK_INTERVAL_SECONDS = 10
async def _post_snapshot(path: str, body: dict) -> None:
if not settings.c2_url:
return
api_key = credentials.get_api_key()
if not api_key:
return
headers = {"Authorization": f"Bearer {api_key}"}
try:
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{settings.c2_url}{path}", json=body, headers=headers)
r.raise_for_status()
except Exception as e:
logger.debug(f"Telemetry uplink to {path} failed: {e}")
async def telemetry_uplink_loop():
while True:
await asyncio.sleep(UPLINK_INTERVAL_SECONDS)
config = load_node_config()
if config.secondary_sdr_mode not in ("adsb", "ais"):
continue
snapshot = await secondary_sdr_client.data()
if not snapshot:
continue
if config.secondary_sdr_mode == "adsb" and snapshot.get("aircraft"):
await _post_snapshot("/telemetry/adsb", {"aircraft": snapshot["aircraft"]})
elif config.secondary_sdr_mode == "ais" and snapshot.get("vessels"):
await _post_snapshot("/telemetry/ais", {"vessels": snapshot["vessels"]})
+18
View File
@@ -224,6 +224,7 @@ async def on_config_push(payload: dict):
hardware_preset = payload.pop("hardware_preset", None) hardware_preset = payload.pop("hardware_preset", None)
ppm_override = payload.pop("ppm_override", None) ppm_override = payload.pop("ppm_override", None)
node_type = payload.pop("node_type", None) node_type = payload.pop("node_type", None)
secondary_sdr_mode = payload.pop("secondary_sdr_mode", None)
enforce_override_timeout = payload.pop("enforce_override_timeout", None) enforce_override_timeout = payload.pop("enforce_override_timeout", None)
try: try:
config = SystemConfig(**payload) config = SystemConfig(**payload)
@@ -243,6 +244,8 @@ async def on_config_push(payload: dict):
node_cfg.ppm_override = float(ppm_override) node_cfg.ppm_override = float(ppm_override)
if node_type is not None: if node_type is not None:
node_cfg.node_type = node_type node_cfg.node_type = node_type
if secondary_sdr_mode is not None:
node_cfg.secondary_sdr_mode = secondary_sdr_mode
if enforce_override_timeout is not None: if enforce_override_timeout is not None:
node_cfg.enforce_override_timeout = bool(enforce_override_timeout) node_cfg.enforce_override_timeout = bool(enforce_override_timeout)
save_node_config(node_cfg) save_node_config(node_cfg)
@@ -257,6 +260,13 @@ async def on_config_push(payload: dict):
await op25_client.start() await op25_client.start()
logger.info(f"Config push applied: {config.name}") logger.info(f"Config push applied: {config.name}")
if secondary_sdr_mode is not None:
from app.internal.secondary_sdr_client import secondary_sdr_client
if secondary_sdr_mode in ("adsb", "ais"):
await secondary_sdr_client.start(secondary_sdr_mode)
else:
await secondary_sdr_client.stop()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# App lifecycle # App lifecycle
@@ -312,12 +322,20 @@ async def lifespan(app: FastAPI):
logger.warning(f"OP25 not ready yet (attempt {attempt + 1}/10), retrying in 3s…") logger.warning(f"OP25 not ready yet (attempt {attempt + 1}/10), retrying in 3s…")
await asyncio.sleep(3) await asyncio.sleep(3)
if node_cfg.secondary_sdr_mode in ("adsb", "ais"):
from app.internal.secondary_sdr_client import secondary_sdr_client
logger.info(f"Resuming secondary SDR (mode={node_cfg.secondary_sdr_mode!r}) after restart.")
await secondary_sdr_client.start(node_cfg.secondary_sdr_mode)
heartbeat_task = asyncio.create_task(mqtt_manager.heartbeat_loop()) heartbeat_task = asyncio.create_task(mqtt_manager.heartbeat_loop())
from app.internal.telemetry_uplink import telemetry_uplink_loop
telemetry_task = asyncio.create_task(telemetry_uplink_loop())
yield # --- app running --- yield # --- app running ---
logger.info("Edge node shutting down.") logger.info("Edge node shutting down.")
heartbeat_task.cancel() heartbeat_task.cancel()
telemetry_task.cancel()
await metadata_watcher.stop() await metadata_watcher.stop()
await call_recorder.stop() await call_recorder.stop()
await radio_bot.stop() await radio_bot.stop()
+1
View File
@@ -34,6 +34,7 @@ class NodeConfig(BaseModel):
hardware_preset: str = "rtl-sdr-v3" hardware_preset: str = "rtl-sdr-v3"
ppm_override: Optional[float] = None ppm_override: Optional[float] = None
node_type: str = "fixed" # fixed or portable node_type: str = "fixed" # fixed or portable
secondary_sdr_mode: str = "none" # none | adsb | ais | op25_2 — requires a second physical SDR
enforce_override_timeout: bool = True enforce_override_timeout: bool = True
override_system_id: Optional[str] = None override_system_id: Optional[str] = None
override_config: Optional[SystemConfig] = None override_config: Optional[SystemConfig] = None
+1 -1
View File
@@ -12,7 +12,7 @@ ENV DEBIAN_FRONTEND=noninteractive
# Install system dependencies # Install system dependencies
RUN apt-get update && \ RUN apt-get update && \
apt-get upgrade -y && \ apt-get upgrade -y && \
apt-get install git pulseaudio pulseaudio-utils liquidsoap -y apt-get install git pulseaudio pulseaudio-utils liquidsoap usbutils -y
# Install custom PulseAudio system config (enables anonymous access for edge-node) # Install custom PulseAudio system config (enables anonymous access for edge-node)
COPY system.pa /etc/pulse/system.pa COPY system.pa /etc/pulse/system.pa
@@ -1,6 +1,7 @@
from fastapi import HTTPException, APIRouter from fastapi import HTTPException, APIRouter
import subprocess import subprocess
import os import os
import re
import signal import signal
import json import json
from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS from models import ConfigGenerator, DecodeMode, ChannelConfig, DeviceConfig, TrunkingConfig, TrunkingChannelConfig, TerminalConfig, MetadataConfig, MetadataStreamConfig, HARDWARE_PRESETS
@@ -69,6 +70,24 @@ def create_op25_router():
async def get_status(): async def get_status():
return {"status": "running" if _is_running() else "stopped"} return {"status": "running" if _is_running() else "stopped"}
@router.get("/devices")
async def list_sdr_devices():
"""Enumerate connected SDR-looking USB devices via lsusb.
Same match heuristic as install.sh's one-shot host check — good enough
to answer "is a second SDR plugged in", not a serial-level device
binding (op25's DeviceConfig.args has no serial concept yet either).
"""
devices = []
try:
out = subprocess.run(["lsusb"], capture_output=True, text=True, timeout=5).stdout
for line in out.splitlines():
if re.search(r"rtl2838|realtek.*283[28]|sdr", line, re.IGNORECASE):
devices.append(line.strip())
except Exception as e:
LOGGER.warning(f"SDR device enumeration failed: {e}")
return {"count": len(devices), "devices": devices}
@router.post("/generate-config") @router.post("/generate-config")
async def generate_config(generator: ConfigGenerator): async def generate_config(generator: ConfigGenerator):
try: try:
+52
View File
@@ -0,0 +1,52 @@
# Secondary-SDR Container — node-26#9
#
# Claims the node's SECOND physical SDR (the first is always op25's). Mode is
# chosen at runtime via the control API, not baked in: dump1090 for ADS-B,
# AIS-catcher for AIS (op25_2 mode is not handled here yet — see node-26#9).
#
# Device claiming is by RTL-SDR index, not serial (op25's DeviceConfig.args
# has no serial concept either — see op25-container/app/models.py). Index 0
# is reserved for op25; this container always addresses index 1. That's a
# real limitation once serial-stable device binding matters (hot-unplug /
# replug can swap indices) — tracked in node-26#9, not fixed here.
#
# UNVERIFIED: this image has not been built or run against real hardware in
# this session (sandboxed authoring machine, no docker). dump1090 and
# AIS-catcher's exact CLI flags below are believed correct from their
# published docs but not confirmed against a real capture — the CTO/QA
# review before this ships to a real node should build and smoke-test it.
FROM python:3.14-slim
ENV DEBIAN_FRONTEND=noninteractive
RUN apt-get update && \
apt-get upgrade -y && \
apt-get install -y --no-install-recommends \
git build-essential cmake pkg-config \
librtlsdr-dev libusb-1.0-0-dev libssl-dev zlib1g-dev usbutils
# dump1090 (antirez/classic) — ADS-B decoder. --write-json support is a
# long-standing, well-documented feature of this fork.
RUN git clone https://github.com/antirez/dump1090 /opt/dump1090 && \
cd /opt/dump1090 && make
# AIS-catcher — AIS decoder.
RUN git clone https://github.com/jvde-github/AIS-catcher /opt/AIS-catcher && \
cd /opt/AIS-catcher && mkdir build && cd build && cmake .. && make
EXPOSE 8002
VOLUME ["/configs"]
WORKDIR /app
COPY ./app /app
COPY docker-entrypoint.sh /usr/local/bin/
RUN sed -i 's/\r$//' /usr/local/bin/docker-entrypoint.sh && \
chmod +x /usr/local/bin/docker-entrypoint.sh
COPY requirements.txt /tmp/requirements.txt
RUN pip3 install --no-cache-dir -r /tmp/requirements.txt
ENTRYPOINT ["/usr/local/bin/docker-entrypoint.sh"]
CMD ["python", "main.py"]
+20
View File
@@ -0,0 +1,20 @@
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# Same rationale as op25-container's OP25_DEBUG_EXPOSE: both containers
# share the host network namespace (network_mode: host), so edge-node
# reaches this control API over localhost regardless of this flag. False
# (default) binds 127.0.0.1; true exposes unauthenticated start/stop to
# the node's LAN and should only ever be set for local development.
secondary_sdr_debug_expose: bool = False
class Config:
env_file = ".env"
settings = Settings()
def bind_host() -> str:
return "0.0.0.0" if settings.secondary_sdr_debug_expose else "127.0.0.1"
@@ -0,0 +1,225 @@
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()
# The node's SECOND SDR, addressed by RTL-SDR index — not serial. op25 always
# claims index 0 (its DeviceConfig.args has no serial concept either, see
# op25-container/app/models.py). No hot-plug re-detection: if the two
# dongles' USB enumeration order changes, this claims the wrong one. Tracked
# as a real gap in node-26#9, not fixed here.
SECONDARY_SDR_DEVICE_INDEX = 1
_PGID_FILE = "/tmp/secondary_sdr.pgid"
_MODE_FILE = "/tmp/secondary_sdr.mode"
ADSB_JSON_DIR = Path("/tmp/adsb")
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:
pgid = _read_pgid()
if pgid is None:
return False
try:
os.killpg(pgid, 0)
return True
except OSError:
return False
def _adsb_command() -> List[str]:
ADSB_JSON_DIR.mkdir(parents=True, exist_ok=True)
return [
"/opt/dump1090/dump1090",
"--net",
"--device-index", str(SECONDARY_SDR_DEVICE_INDEX),
"--write-json", str(ADSB_JSON_DIR),
"--write-json-every", "1",
]
def _ais_command() -> List[str]:
return [
"/opt/AIS-catcher/build/AIS-catcher",
"-d", str(SECONDARY_SDR_DEVICE_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:
if is_running():
stop()
if mode == "adsb":
cmd = _adsb_command()
elif mode == "ais":
cmd = _ais_command()
with _ais_lock:
_ais_vessels.clear()
else:
raise ValueError(f"Unknown secondary SDR mode: {mode!r}")
try:
needs_stdout = mode == "ais"
proc = subprocess.Popen(
cmd,
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,
)
if needs_stdout:
threading.Thread(target=_ais_reader, args=(proc,), daemon=True).start()
_save_state(proc.pid, mode)
LOGGER.info(f"Started secondary SDR decoder mode={mode!r} pid={proc.pid}")
return True
except Exception as e:
LOGGER.error(f"Failed to start secondary SDR decoder mode={mode!r}: {e}")
return False
def stop() -> bool:
pgid = _read_pgid()
if pgid is None:
return True
try:
os.killpg(pgid, signal.SIGTERM)
except OSError:
pass
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 _read_adsb_snapshot() -> List[Dict[str, Any]]:
"""
Map dump1090's aircraft.json (--write-json output) to the server's
telemetry schema. Field names (hex/flight/lat/lon/altitude/speed/track)
match dump1090's long-documented JSON format — UNVERIFIED against a real
capture in this session, see the Dockerfile's caveat.
"""
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": a.get("altitude"),
"ground_speed_kt": 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": []}
@@ -0,0 +1,31 @@
import logging
from logging.handlers import RotatingFileHandler
def create_logger(name, level=logging.DEBUG, max_bytes=10485760, backup_count=2):
debug_log_file = "./secondary-sdr.debug.log"
info_log_file = "./secondary-sdr.log"
logger = logging.getLogger(name)
logger.setLevel(level)
if not logger.hasHandlers():
console_handler = logging.StreamHandler()
console_handler.setLevel(level)
debug_file_handler = RotatingFileHandler(debug_log_file, maxBytes=max_bytes, backupCount=backup_count)
debug_file_handler.setLevel(logging.DEBUG)
info_file_handler = RotatingFileHandler(info_log_file, maxBytes=max_bytes, backupCount=backup_count)
info_file_handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)
debug_file_handler.setFormatter(formatter)
info_file_handler.setFormatter(formatter)
logger.addHandler(console_handler)
logger.addHandler(debug_file_handler)
logger.addHandler(info_file_handler)
return logger
+13
View File
@@ -0,0 +1,13 @@
from fastapi import FastAPI
import routers.secondary_controller as secondary_controller
from config import bind_host
app = FastAPI()
app.include_router(secondary_controller.create_secondary_router(), prefix="/secondary")
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host=bind_host(), port=8002, reload=True)
@@ -0,0 +1,40 @@
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from internal import decoder_control
from internal.logger import create_logger
LOGGER = create_logger(__name__)
class StartBody(BaseModel):
mode: str # adsb | ais
def create_secondary_router():
router = APIRouter()
@router.post("/start")
async def start(body: StartBody):
try:
ok = decoder_control.start(body.mode)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
if not ok:
raise HTTPException(status_code=500, detail="Failed to start secondary SDR decoder")
return {"status": f"secondary SDR started ({body.mode})"}
@router.post("/stop")
async def stop():
decoder_control.stop()
return {"status": "secondary SDR stopped"}
@router.get("/status")
async def get_status():
return decoder_control.status()
@router.get("/data")
async def get_data():
return decoder_control.data()
return router
@@ -0,0 +1,3 @@
#!/bin/bash
mkdir -p /tmp/adsb
exec "$@"
+3
View File
@@ -0,0 +1,3 @@
uvicorn
fastapi
pydantic-settings