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>

drb-edge-node/app/internal/telemetry_uplink.py

@@ -33,9 +33,12 @@ async def telemetry_uplink_loop():
     while True:
         await asyncio.sleep(UPLINK_INTERVAL_SECONDS)
         config = load_node_config()
-        if config.secondary_sdr_mode != "adsb":
+        if config.secondary_sdr_mode not in ("adsb", "ais"):
             continue
         snapshot = await secondary_sdr_client.data()
-        if not snapshot or not snapshot.get("aircraft"):
+        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"]})

secondary-sdr-container/app/internal/decoder_control.py

@@ -2,6 +2,7 @@ import json
 import os
 import signal
 import subprocess
+import threading
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
@@ -9,6 +10,16 @@ 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
@@ -62,6 +73,58 @@ def _adsb_command() -> List[str]:
     ]
 
 
+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()
@@ -69,12 +132,23 @@ def start(mode: str) -> bool:
     if mode == "adsb":
         cmd = _adsb_command()
     elif mode == "ais":
-        raise ValueError("AIS mode is not wired yet (node-26#9) — only 'adsb' runs today.")
+        cmd = _ais_command()
+        with _ais_lock:
+            _ais_vessels.clear()
     else:
         raise ValueError(f"Unknown secondary SDR mode: {mode!r}")
 
     try:
-        proc = subprocess.Popen(cmd, preexec_fn=os.setsid)
+        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
@@ -144,4 +218,8 @@ def data() -> Dict[str, Any]:
     mode = _read_mode()
     if mode == "adsb":
         return {"mode": mode, "aircraft": _read_adsb_snapshot()}
-    return {"mode": mode, "aircraft": []}
+    if mode == "ais":
+        with _ais_lock:
+            vessels = list(_ais_vessels.values())
+        return {"mode": mode, "vessels": vessels}
+    return {"mode": mode, "aircraft": [], "vessels": []}