Streaming

Streaming

CotStream wraps a TCP connection and turns raw CoT traffic into CotEvent objects in both directions.

Usage: await CotStream.connect(host, port, ssl=False)

Parameter	Type	Description
host	str	Server hostname or IP. Required.
port	int	Streaming port (8087 plaintext, 8089 TLS). Required.
ssl	bool | SSLContext	False (default), True, or a prepared context
callsign, team, role	str	Optional identity; sends the initial SA on connect

Sending and receiving

Receiving is an async iterator — the pythonic way, backpressure included:

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

Sending takes any CotEvent; missing timestamps are filled automatically. When an identity is configured, non-self SA messages and markers get a parent-producer link pointing back at us automatically — other clients use it to attribute markers that are not about the sender:

await stream.send(CotEvent(cot_type="a-f-G-U-C", lat=1.0, lon=2.0))
# detail now contains: <link uid=our-uid type=a-f-G-U-C parent_callsign=our-callsign relation="p-p"/>

Control messages (t-x-*) and our own events are never decorated; chats do not need it either.

Closing is graceful by default — a t-x-d-d delete task referencing our own UID (with our actual type) tells the server to drop our contact entry and lets remote clients expire our marker, then the connection closes:

await stream.close()

Position updates

The server only knows where we are through SA reports. Update our position and an out-of-interval SA goes out immediately — no waiting for the next periodic report:

await stream.set_position(48.208, 16.373)

Pass it as position=(lat, lon) at connect time to include it in the initial identity SA. Changing what we are works the same way: stream.set_cot_type("a-f-A-C-F") re-announces immediately.

Initial identification

Give the identity kwargs and the stream announces itself right after connecting — before you send or receive anything else. Verified live against TAK Server 5.7 + ATAK-CIV 5.8: without a fully valid initial SA about ourselves, the server does not route everything to us; GeoChat messages only arrive for identified clients.

stream = await CotStream.connect(
    "tak.example.com", 8089, ssl=ctx,
    callsign="MYCALL", team="Cyan", role="Team Member",
)

The initial SA carries contact with the endpoint="*:-1:stcp" marker (this is what makes other clients treat us as geochat-capable), __group with a valid team color and role, a status block and takv with platform/version. Re-send it anytime with await stream.announce_identity().

The protobuf offer is still ignored; the session stays CoT XML forever.

Behaviour notes

• Incoming bytes are buffered and sliced into complete <event>...</event> frames. Real servers deliver several concatenated events per TCP segment; this is handled correctly, as are self-closing events and > characters inside quoted attribute values.
• Parsing uses defusedxml — never trust the wire.
• A background pump drains inbound traffic continuously, so keepalive timers stay honest even when your application is not currently reading.
• Keepalive follows the TAK client implementation (see cotxml.md in the TAK workspace): t-x-c-t ping with uid <deviceuid>-ping after 15 s without inbound traffic, repeated at most every 4.5 s; the connection is declared dead after 25 s of silence and closed.
• With an identity configured we also refresh our own SA every 30 s (SA_INTERVAL_SECONDS) so the server never expires our marker; position changes go out immediately via set_position() instead of waiting for the interval.
• send(event, fill_defaults=True) stamps time/start/stale onto the event object itself, then writes it to the wire.

Note

There is no reconnect logic and no protobuf negotiation yet. The server's t-x-takp-v offer arrives as a regular event (type t-x-takp-v) and is safely ignored if you do nothing with it — the session stays CoT XML forever. Reconnect is planned; see the README roadmap in the repository.

Live testing against a real server

The repository ships live integration tests, kept separate from unit tests (live_tests/, pytest marker live). They read their configuration from an untracked .env in the repository root:

TAK_LIVE_HOST=tak.example.com
TAK_LIVE_PORT=8089
TAK_LIVE_TLS=true
TAK_LIVE_CERT=path/to/client.pem
TAK_LIVE_KEY=path/to/client.key

Run them locally only:

poetry run pytest live_tests/ -m live

Without a configured .env every live test skips cleanly. Real hostnames and certificates must never be committed — pre-commit hooks scan staged files for cert material and private addresses, and they will bite you.