Files
node-26/drb-edge-node/app/internal/telemetry_uplink.py
Logan CusanoandClaude Sonnet 5 52f31bbcc0
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
Wire AIS end to end: AIS-catcher support in secondary-sdr container
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

45 lines
1.6 KiB
Python

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"]})