Add CoT (Cursor-on-Target) export for ATAK/WinTAK/TAK Server

config.py

@@ -566,6 +566,12 @@ def _get_env_bool(key: str, default: bool) -> bool:
 MQTT_TOPIC_PREFIX = _get_env("MQTT_TOPIC_PREFIX", "intercept")
 MQTT_RETAIN = _get_env_bool("MQTT_RETAIN", False)
 
+# CoT (Cursor-on-Target) export for ATAK/WinTAK/TAK Server (optional — disabled when COT_HOST is empty)
+COT_HOST = _get_env("COT_HOST", "")
+COT_PORT = _get_env_int("COT_PORT", 6969)
+COT_PROTO = _get_env("COT_PROTO", "udp")  # udp or tcp
+COT_STALE_SECONDS = _get_env_int("COT_STALE_SECONDS", 60)
+
 # Admin credentials
 ADMIN_USERNAME = _get_env("ADMIN_USERNAME", "admin")
 ADMIN_PASSWORD = _get_env("ADMIN_PASSWORD", "admin")

tests/test_cot.py

@@ -0,0 +1,140 @@
+"""Tests for utils/cot.py — CoT (Cursor-on-Target) export."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import utils.cot as cot
+
+
+def _reset_module_state():
+    cot._enabled = None
+    cot._udp_socket = None
+
+
+class FakeConfig:
+    COT_HOST = "239.2.3.1"
+    COT_PORT = 6969
+    COT_PROTO = "udp"
+    COT_STALE_SECONDS = 60
+
+
+class DisabledConfig(FakeConfig):
+    COT_HOST = ""
+
+
+def test_disabled_when_host_empty():
+    _reset_module_state()
+    with patch.object(cot, "_get_config", return_value=DisabledConfig):
+        cot.publish("adsb", {"icao": "ABC123", "lat": -33.9, "lon": 18.4})
+        assert cot._is_enabled() is False
+
+
+def test_event_to_cot_xml_adsb():
+    _reset_module_state()
+    with patch.object(cot, "_get_config", return_value=FakeConfig):
+        event = {"icao": "ABC123", "callsign": "SAA123", "lat": -33.9, "lon": 18.4, "altitude": 35000}
+        xml = cot.event_to_cot_xml("adsb", event)
+
+    assert xml is not None
+    assert 'type="a-f-A"' in xml
+    assert 'uid="ICAO-ABC123"' in xml
+    assert 'lat="-33.9"' in xml
+    assert 'lon="18.4"' in xml
+    assert 'hae="35000"' in xml
+    assert 'callsign="SAA123"' in xml
+
+
+def test_event_to_cot_xml_ais():
+    _reset_module_state()
+    with patch.object(cot, "_get_config", return_value=FakeConfig):
+        event = {"mmsi": "601234000", "callsign": "ZS1234", "lat": -33.9, "lon": 18.4}
+        xml = cot.event_to_cot_xml("ais", event)
+
+    assert xml is not None
+    assert 'type="a-f-S"' in xml
+    assert 'uid="MMSI-601234000"' in xml
+
+
+def test_event_to_cot_xml_aprs():
+    _reset_module_state()
+    with patch.object(cot, "_get_config", return_value=FakeConfig):
+        event = {"callsign": "ZS1ABC-9", "lat": -33.9, "lon": 18.4}
+        xml = cot.event_to_cot_xml("aprs", event)
+
+    assert xml is not None
+    assert 'type="a-f-G-U-C"' in xml
+    assert 'uid="APRS-ZS1ABC-9"' in xml
+
+
+def test_event_to_cot_xml_meshtastic():
+    _reset_module_state()
+    with patch.object(cot, "_get_config", return_value=FakeConfig):
+        event = {"id": "!55890aeb", "callsign": "Whskybadger47", "lat": -33.9, "lon": 18.4, "altitude": 15}
+        xml = cot.event_to_cot_xml("meshtastic", event)
+
+    assert xml is not None
+    assert 'type="a-f-G-U-C"' in xml
+    assert 'uid="MESH-!55890aeb"' in xml
+    assert 'callsign="Whskybadger47"' in xml
+
+
+def test_event_to_cot_xml_missing_position_returns_none():
+    _reset_module_state()
+    with patch.object(cot, "_get_config", return_value=FakeConfig):
+        assert cot.event_to_cot_xml("adsb", {"icao": "ABC123"}) is None
+
+
+def test_event_to_cot_xml_unknown_mode_returns_none():
+    _reset_module_state()
+    with patch.object(cot, "_get_config", return_value=FakeConfig):
+        assert cot.event_to_cot_xml("pager", {"lat": -33.9, "lon": 18.4}) is None
+
+
+def test_publish_sends_udp_datagram():
+    _reset_module_state()
+    fake_socket = MagicMock()
+    with (
+        patch.object(cot, "_get_config", return_value=FakeConfig),
+        patch.object(cot, "_get_udp_socket", return_value=fake_socket),
+    ):
+        cot.publish("adsb", {"icao": "ABC123", "lat": -33.9, "lon": 18.4})
+
+    fake_socket.sendto.assert_called_once()
+    args, _ = fake_socket.sendto.call_args
+    payload, addr = args
+    assert addr == ("239.2.3.1", 6969)
+    assert b"<event" in payload
+    assert b'uid="ICAO-ABC123"' in payload
+
+
+def test_publish_sends_tcp_when_configured():
+    _reset_module_state()
+
+    class TcpConfig(FakeConfig):
+        COT_PROTO = "tcp"
+
+    fake_conn = MagicMock()
+    fake_conn.__enter__ = MagicMock(return_value=fake_conn)
+    fake_conn.__exit__ = MagicMock(return_value=False)
+
+    with (
+        patch.object(cot, "_get_config", return_value=TcpConfig),
+        patch("socket.create_connection", return_value=fake_conn) as mock_connect,
+    ):
+        cot.publish("ais", {"mmsi": "601234000", "lat": -33.9, "lon": 18.4})
+
+    mock_connect.assert_called_once_with(("239.2.3.1", 6969), timeout=5)
+    fake_conn.sendall.assert_called_once()
+
+
+def test_publish_swallows_send_errors():
+    _reset_module_state()
+    fake_socket = MagicMock()
+    fake_socket.sendto.side_effect = OSError("network unreachable")
+    with (
+        patch.object(cot, "_get_config", return_value=FakeConfig),
+        patch.object(cot, "_get_udp_socket", return_value=fake_socket),
+    ):
+        # Should not raise.
+        cot.publish("adsb", {"icao": "ABC123", "lat": -33.9, "lon": 18.4})

tests/test_meshtastic.py

@@ -450,4 +450,56 @@ def test_client_disconnect_idempotent(self):
         client.disconnect()
         client.disconnect()
 
+
+class TestMeshtasticPositionEventPipeline:
+    """Tests for POSITION_APP packets feeding the shared event pipeline (alerts/recording/MQTT/CoT)."""
+
+    def test_position_packet_publishes_to_event_pipeline(self):
+        """A POSITION_APP packet with valid lat/lon should call process_event with mode='meshtastic'."""
+        from utils.meshtastic import MeshtasticClient
+
+        client = MeshtasticClient()
+        packet = {"from": 0x55890AEB}
+        decoded = {"position": {"latitude": -33.9, "longitude": 18.4, "altitude": 15}}
+
+        with patch("utils.event_pipeline.process_event") as mock_process_event:
+            client._track_node_from_packet(packet, decoded, "POSITION_APP")
+
+        mock_process_event.assert_called_once()
+        mode, event, event_type = mock_process_event.call_args[0]
+        assert mode == "meshtastic"
+        assert event_type == "position"
+        assert event["lat"] == -33.9
+        assert event["lon"] == 18.4
+        assert event["altitude"] == 15
+        assert event["id"] == "!55890aeb"
+
+    def test_position_packet_without_coordinates_does_not_publish(self):
+        """A POSITION_APP packet lacking lat/lon should not touch the event pipeline."""
+        from utils.meshtastic import MeshtasticClient
+
+        client = MeshtasticClient()
+        packet = {"from": 0x55890AEB}
+        decoded = {"position": {}}
+
+        with patch("utils.event_pipeline.process_event") as mock_process_event:
+            client._track_node_from_packet(packet, decoded, "POSITION_APP")
+
+        mock_process_event.assert_not_called()
+
+    def test_position_pipeline_failure_does_not_break_tracking(self):
+        """If the event pipeline raises, node tracking should still succeed."""
+        from utils.meshtastic import MeshtasticClient
+
+        client = MeshtasticClient()
+        packet = {"from": 0x55890AEB}
+        decoded = {"position": {"latitude": -33.9, "longitude": 18.4}}
+
+        with patch("utils.event_pipeline.process_event", side_effect=RuntimeError("boom")):
+            client._track_node_from_packet(packet, decoded, "POSITION_APP")
+
+        node = client._nodes[0x55890AEB]
+        assert node.latitude == -33.9
+        assert node.longitude == 18.4
+
         assert client.is_running is False

utils/cot.py

@@ -0,0 +1,139 @@
+"""
+Optional CoT (Cursor-on-Target) export for ATAK / WinTAK / TAK Server.
+
+Converts decoded position events (ADS-B aircraft, AIS vessels, APRS stations)
+to CoT XML and sends them over UDP or TCP to a configured host:port — the
+default ATAK SA multicast group (239.2.3.1:6969) or a TAK Server. Disabled
+when INTERCEPT_COT_HOST is not set — no host, no socket, no error.
+
+Configure via environment variables:
+
+    INTERCEPT_COT_HOST           destination host / multicast group (required to enable)
+    INTERCEPT_COT_PORT           destination port (default 6969)
+    INTERCEPT_COT_PROTO          "udp" or "tcp" (default "udp")
+    INTERCEPT_COT_STALE_SECONDS  how long a client should treat the event as valid (default 60)
+"""
+
+from __future__ import annotations
+
+import logging
+import socket
+import threading
+import time
+from typing import Any
+from uuid import uuid4
+from xml.sax.saxutils import escape
+
+logger = logging.getLogger(__name__)
+
+_socket_lock = threading.Lock()
+_udp_socket: socket.socket | None = None
+_enabled: bool | None = None  # None = not yet initialised
+
+# mode -> (cot_type, uid_field, uid_prefix, callsign_field)
+_MODE_MAP = {
+    "adsb": ("a-f-A", "icao", "ICAO", "callsign"),
+    "ais": ("a-f-S", "mmsi", "MMSI", "callsign"),
+    "aprs": ("a-f-G-U-C", "callsign", "APRS", "callsign"),
+    "meshtastic": ("a-f-G-U-C", "id", "MESH", "callsign"),
+}
+
+
+def _get_config():
+    import config
+
+    return config
+
+
+def _is_enabled() -> bool:
+    global _enabled
+    if _enabled is None:
+        cfg = _get_config()
+        _enabled = bool(cfg.COT_HOST)
+    return _enabled
+
+
+def _get_udp_socket() -> socket.socket:
+    global _udp_socket
+    if _udp_socket is not None:
+        return _udp_socket
+
+    with _socket_lock:
+        if _udp_socket is None:
+            _udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    return _udp_socket
+
+
+def _build_event_xml(
+    cot_type: str,
+    uid: str,
+    lat: float,
+    lon: float,
+    hae: float | None,
+    callsign: str | None,
+    stale_seconds: int,
+) -> str:
+    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+    stale = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + stale_seconds))
+    hae_val = hae if hae is not None else 0
+    contact = f'<contact callsign="{escape(str(callsign))}"/>' if callsign else ""
+
+    return (
+        f'<?xml version="1.0" encoding="UTF-8"?>'
+        f'<event version="2.0" uid="{escape(uid)}" type="{cot_type}" '
+        f'time="{now}" start="{now}" stale="{stale}" how="m-g">'
+        f'<point lat="{lat}" lon="{lon}" hae="{hae_val}" ce="9999999" le="9999999"/>'
+        f"<detail>{contact}</detail>"
+        f"</event>"
+    )
+
+
+def event_to_cot_xml(mode: str, event: dict[str, Any]) -> str | None:
+    """Build CoT XML for a decoded event, or None if it lacks a position or a known mapping."""
+    mapping = _MODE_MAP.get(mode)
+    if mapping is None:
+        return None
+
+    cot_type, uid_field, uid_prefix, callsign_field = mapping
+
+    lat = event.get("lat")
+    lon = event.get("lon")
+    if lat is None or lon is None:
+        return None
+
+    uid_value = event.get(uid_field)
+    uid = f"{uid_prefix}-{uid_value}" if uid_value else f"{uid_prefix}-{uuid4()}"
+
+    hae = event.get("altitude")
+    callsign = event.get(callsign_field)
+    cfg = _get_config()
+
+    return _build_event_xml(cot_type, uid, lat, lon, hae, callsign, cfg.COT_STALE_SECONDS)
+
+
+def publish(mode: str, event: dict[str, Any], event_type: str | None = None) -> None:
+    """Send a decoded event to the configured CoT destination, if enabled.
+
+    Args:
+        mode:       Source module name (e.g. 'adsb', 'ais', 'aprs').
+        event:      The decoded event dict.
+        event_type: Optional sub-type string (unused for CoT — position-only export).
+    """
+    if not _is_enabled():
+        return
+
+    xml = event_to_cot_xml(mode, event)
+    if xml is None:
+        return
+
+    cfg = _get_config()
+    payload = xml.encode("utf-8")
+
+    try:
+        if cfg.COT_PROTO == "tcp":
+            with socket.create_connection((cfg.COT_HOST, cfg.COT_PORT), timeout=5) as sock:
+                sock.sendall(payload)
+        else:
+            _get_udp_socket().sendto(payload, (cfg.COT_HOST, cfg.COT_PORT))
+    except Exception as exc:
+        logger.debug("CoT send error to %s:%d: %s", cfg.COT_HOST, cfg.COT_PORT, exc)

utils/event_pipeline.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from utils.alerts import get_alert_manager
+from utils.cot import publish as cot_publish
 from utils.mqtt import publish as mqtt_publish
 from utils.recording import get_recording_manager
 from utils.temporal_patterns import get_pattern_detector
@@ -61,6 +62,12 @@ def process_event(mode: str, event: dict | Any, event_type: str | None = None) -
         # MQTT failures should never break streaming
         pass
 
+    try:
+        cot_publish(mode, event, event_type)
+    except Exception:
+        # CoT failures should never break streaming
+        pass
+
 
 def _extract_device_id(event: dict) -> str | None:
     for field in DEVICE_ID_FIELDS:

utils/meshtastic.py

@@ -648,6 +648,24 @@ def _track_node_from_packet(self, packet: dict, decoded: dict, portnum: str) ->
                     node.longitude = lon
                     node.altitude = position.get("altitude", node.altitude)
 
+                    try:
+                        from utils.event_pipeline import process_event
+
+                        process_event(
+                            "meshtastic",
+                            {
+                                "id": node.user_id,
+                                "callsign": node.long_name or node.short_name,
+                                "lat": node.latitude,
+                                "lon": node.longitude,
+                                "altitude": node.altitude,
+                            },
+                            "position",
+                        )
+                    except Exception:
+                        # Event pipeline failures should never break mesh packet handling
+                        pass
+
         # Parse TELEMETRY_APP for battery and other metrics
         elif portnum == "TELEMETRY_APP":
             telemetry = decoded.get("telemetry", {})