Repository navigation
Conversation
Closes smittix#179. Publishes decoded ADS-B, AIS, and APRS position events as CoT XML to a configured UDP/TCP destination (defaults to the ATAK SA multicast group 239.2.3.1:6969, or point it at a TAK Server), following the same optional/env-gated export pattern already used by the MQTT export (utils/mqtt.py): disabled by default, no external dependency, hooked into the shared event pipeline so all three modes get it for free. New env vars: INTERCEPT_COT_HOST (unset = disabled), INTERCEPT_COT_PORT, INTERCEPT_COT_PROTO, INTERCEPT_COT_STALE_SECONDS.
Meshtastic was the only mode not wired into the shared event pipeline (no alerts, recording, MQTT, or now CoT). This hooks POSITION_APP packet handling into utils.event_pipeline.process_event(), the same call already made for ADS-B/AIS/APRS, and adds a "meshtastic" entry to cot.py's mode map (CoT type a-f-G-U-C, uid MESH-<node id>). Position updates from mesh nodes now flow through the same alerts/recording/MQTT/CoT infrastructure as every other mode, so a Meshtastic node with GPS (e.g. the Heltec T114) shows up on ATAK/WinTAK alongside tracked aircraft/vessels/APRS stations when CoT export is enabled.
|
Thanks — this is well-built. Opt-in behind Two things I'd like to resolve before merging. 1. TCP opens a new connection per event. with socket.create_connection((cfg.COT_HOST, cfg.COT_PORT), timeout=5) as sock:
sock.sendall(payload)
UDP (the default) doesn't have this problem, so this only bites TCP users, but they're likely the ones pointing at a real TAK Server. A persistent connection reused across events, reconnecting on failure, would fix it; a bounded queue with a sender thread would be even better, at more complexity. Happy with either, or with a documented caveat if you'd rather keep it simple for now. 2. f'<point lat="{lat}" lon="{lon}" hae="{hae_val}" ce="9999999" le="9999999"/>'
Neither is a blocker on the default UDP path; I'd just rather fix them while the code is fresh in your head than discover them from a bug report. |
|
Thanks for this — reviewed and it looks solid. |
Closes #179.
Adds optional CoT (Cursor-on-Target) export for ADS-B, AIS, and APRS position events, so they can be piped into ATAK, WinTAK, or a TAK Server.
What this does
utils/cot.py: converts a decoded position event into CoT XML and sends it via UDP (default: the ATAK SA multicast group239.2.3.1:6969) or TCP to a configured host/port.utils/event_pipeline.py, alongside the existing alerts/recording/MQTT dispatch, so ADS-B, AIS, and APRS events all get exported for free without touching each route file individually.a-f-A(uidICAO-<icao>), AIS vessels →a-f-S(uidMMSI-<mmsi>), APRS stations →a-f-G-U-C(uidAPRS-<callsign>).utils/mqtt.py): disabled by default, zero behavior change unless configured, no new external dependency (pure stdlibsocket/XML — nopytak, to keep it consistent with this codebase's synchronous/gevent model rather than pytak's asyncio design).Configuration (all optional, unset = disabled)
INTERCEPT_COT_HOST— destination host or multicast group (required to enable)INTERCEPT_COT_PORT— destination port (default6969)INTERCEPT_COT_PROTO—udportcp(defaultudp)INTERCEPT_COT_STALE_SECONDS— how long a client should treat the event as valid (default60)Scope
Deliberately limited to outbound CoT broadcast/TAK-server send for ADS-B + AIS + APRS (the three modes with lat/lon position data today). A bidirectional pyTAK router was floated in #179 but is a separate, larger feature — happy to follow up if there's interest.
Testing
New
tests/test_cot.py— unit tests for XML generation (per-mode type/uid mapping, missing-position handling, unknown-mode handling) and the UDP/TCP send paths (mocked socket, verifies no exception on send failure). All passing locally viapytest tests/test_cot.py.I've been running this on my own deployment; happy to adjust the type mapping or config surface if you'd prefer something different.