python-tak-cot-streaming

Streaming-friendly TAK Cursor-on-Target library for Python. Work with typed Python objects, never with XML trees, sockets or protocol details.

Status: early proof of concept. The object model, the streaming client (TLS, identification, keepalive, graceful disconnect) and the helper layer work; automatic reconnect and protobuf negotiation are on the roadmap.

• CotEvent object model covering the CoT event envelope and point, with round-trip fidelity for unknown attributes and detail blocks
• from_xml() / to_xml() serialization with round-trip fidelity.
• Async streaming over TCP/TLS (CotStream) - receive events as an async iterator, send events with automatic time/stale defaults; initial identification SA, position updates, parent-link attribution, graceful t-x-d-d disconnect and spec-conformant keepalive are built in.
• Detail helpers: typed setters, event factories (marker, SA report, GeoChat message and receipts, emergency alert/cancel) and predicates.

• Python 3.11+
• Poetry for dependency management

poetry install

import asyncio

from takstream import CotEvent, CotStream


async def main() -> None:
    # TLS: just pass cert/key - the context is built internally.
    # Without cert/key you get cleartext.
    stream = await CotStream.connect(
        "tak.example.com",
        8089,
        cert="client.pem",
        key="client.key",
        callsign="MYCALL", team="Cyan", role="Team Member",
    )

    async for event in stream:
        print(event.uid, event.cot_type, event.point.lat, event.point.lon)


asyncio.run(main())

Position is read from the spec-style <point> child element when present, with legacy attribute-style (<event lat=...>) as fallback. Serialization always emits a <point> child per spec.

Fluent setters upsert typed blocks into detail without touching unknown siblings:

event.set_contact("MYCALL")
event.set_group("Cyan", "Team Member")
event.set_track(speed=5.0, course=90.0)
event.set_remarks("some text")
event.set_status(readiness=True, battery=88)
event.set_precision_location(geopointsrc="GPS", altsrc="GPS")
event.set_color(-65536)

# Map marker
marker = CotEvent.marker(48.208, 16.373, callsign="PIT", remarks="landmark")

# Situational-awareness position report
sa = CotEvent.sa_report(48.208, 16.373, callsign="MYCALL", team="Cyan", role="Team Member")

# GeoChat message
chat = CotEvent.chat_message("roger that", sender_callsign="ALPHA", sender_uid="ANDROID-123")

# Emergency alert / cancel
alert = CotEvent.emergency_alert(48.2, 16.3, callsign="MYCALL")
cancel = CotEvent.emergency_cancel(uid=alert.uid, callsign="MYCALL")

await stream.send(marker)

is_sa(), is_marker(), is_chat(), is_emergency(), is_stale().

Convenience kwargs on construction, readable/writable properties afterwards:

event = CotEvent(cot_type="a-f-G-U-C", callsign="MYCALL", team="Cyan", role="RTO", remarks="hi")
event.callsign = "RENAMED"

CotStream.connect(...) accepts callsign=, team=, role= to send an initial SA identification right after connecting - the server needs this before routing everything (e.g. GeoChat) to you:

stream = await CotStream.connect(
    "tak.example.com", 8089,
    cert="client.pem", key="client.key",
    callsign="MYCALL", team="Cyan", role="Team Member",
)

Helpers: event.callsign, event.age(), event.is_stale().

lat, lon, hae, ce, le - unknown values use the CoT sentinel 9999999.

• await CotStream.connect(host, port, ...) - open a TCP CoT stream. TLS is used automatically when cert=/key= are given, cleartext otherwise; pass ca= to verify the server. Optional identity kwargs (callsign, team, role, cot_type, uid, position) announce a fully valid initial SA right after connecting.
• async for event in stream - iterate incoming events.
• await stream.send(event) - serialize and send an event; non-self SA/markers automatically get a parent-producer link back to us.
• await stream.set_position(lat, lon) - update position and immediately send a fresh SA report.
• stream.set_cot_type(type) - change our own type and re-announce.
• await stream.close() - send a graceful t-x-d-d delete task and close.

Keepalive runs internally: spec-conformant t-x-c-t pings after 15 s idle, dead-connection detection after 25 s silence. The protobuf offer (t-x-takp-v) is ignored - sessions stay CoT XML.

poetry install
poetry run pre-commit install
poetry run pytest            # unit tests

Live tests run against a real TAK server and are skipped unless configured. Create an untracked .env file in the repository root:

TAK_LIVE_HOST=tak.example.com
TAK_LIVE_PORT=8087
TAK_LIVE_CERT=path/to/client.pem
TAK_LIVE_KEY=path/to/client.key

Then run:

poetry run pytest live_tests/ -m live

Certificates and real hostnames must never be committed - .gitignore excludes cert material and env files, and pre-commit hooks scan staged files.

GPL-3.0-or-later, see LICENSE.

Need the TAK Server REST API (missions, data feeds, user management, certificate enrollment, ...)? That is python-takserver-api — this library's sibling. It speaks HTTP to port 8443 while takstream speaks CoT to the streaming ports; together they cover both sides of a TAK Server.

It is also a convenient way to obtain client certificates in the first place: its cert-manager/enrollment wrappers can request and download a client certificate from the server, which you can then hand to CotStream.connect(cert=..., key=...).