pytak

Python Team Awareness Kit (PyTAK) Module.

pytak.Worker(queue[, config])

Meta class for all other Worker Classes.

pytak.TXWorker(queue, config, writer)

Works data queue and hands off to Protocol Workers.

pytak.RXWorker(queue, config, reader)

Async receive (input) queue worker.

pytak.QueueWorker(queue, config)

Read non-CoT Messages from an async network client.

pytak.CLITool(config[, tx_queue, rx_queue])

Wrapper Object for CLITools.

pytak.create_udp_client(url[, iface, local_addr])

Create an AsyncIO UDP network client for Unicast, Broadcast & Multicast.

pytak.protocol_factory(config)

Create a network connection class instance.

pytak.txworker_factory(queue, config)

Create a PyTAK TXWorker based on URL parameters.

pytak.rxworker_factory(queue, config)

pytak.cli(app_name)

Abstract implementation of a Command Line Interface (CLI).

pytak.read_pref_package(pref_package)

Read a pref package / data package of preferences.

:py:obj:`pytak.TXWorker <pytak.TXWorker>`\ \(queue\, config\, writer\)

:py:obj:`pytak.RXWorker <pytak.RXWorker>`\ \(queue\, config\, reader\)

:py:obj:`pytak.QueueWorker <pytak.QueueWorker>`\ \(queue\, config\)

:py:obj:`pytak.CLITool <pytak.CLITool>`\ \(config\[\, tx\_queue\, rx\_queue\]\)

:py:obj:`pytak.create_udp_client <pytak.create_udp_client>`\ \(url\[\, iface\, local\_addr\]\)

:py:obj:`pytak.protocol_factory <pytak.protocol_factory>`\ \(config\)

:py:obj:`pytak.txworker_factory <pytak.txworker_factory>`\ \(queue\, config\)

:py:obj:`pytak.rxworker_factory <pytak.rxworker_factory>`\ \(queue\, config\)

:py:obj:`pytak.read_pref_package <pytak.read_pref_package>`\ \(pref\_package\)

`aiscot <https://github.com/snstac/aiscot>`_
Automatic Identification System (AIS) to COT Gateway. Transforms marine AIS position messages to COT PLI Events.

`adsbcot <https://github.com/snstac/adsbcot>`_
Automatic Dependent Surveillance-Broadcast (ADS-B) to COT Gateway. Transforms aircraft ADS-B position messages to COT PLI Events.

`adsbxcot <https://github.com/snstac/adsbxcot>`_
ADS-B Exchange to COT Gateway. Transforms aircraft ADS-B position messages to COT PLI Events.

`stratuxcot <https://github.com/snstac/stratuxcot>`_
Stratux ADS-B to COT Gateway. Transforms aircraft ADS-B position messages to COT PLI Events.

`aprscot <https://github.com/snstac/aprscot>`_
Automatic Packet Reporting System (APRS) to COT Gateway. Transforms APRS position messages to COT PLI Events.

`spotcot <https://github.com/snstac/spotcot>`_
Globalstar SPOT to COT Gateway. Transforms Spot satellite position messages to COT PLI Events.

`inrcot <https://github.com/snstac/inrcot>`_
Garmin inReach to COT Gateway. Transforms inReach satellite position messages to COT PLI Events.

`zellocot <https://github.com/snstac/zellocot>`_
ZelloWork to COT Gateway. Transforms ZelloWork user locations to COT PLI Events.

PyTAK is used in mission criticial environments, every day, across all official
`TAK Products <https://tak.gov>`_:

PyTAK has been tested and is compatible with many situational awareness (SA) & common
operating picture (COP) systems:

`Free TAK Server (FTS/FreeTAKServer) <https://github.com/FreeTAKTeam/FreeTakServer>`_

TLS Unicast: ``tls://host:port`` (see `TLS Support <https://github.com/snstac/pytak#tls-support>`_ section below)

PyTAK requires Python 3.6 or above and WILL NOT work on Python versions below 3.6. It
should run on almost any platform that supports Python 3.6+, including Linux, Windows,
Raspberry Pi, Android, et al.

FTS (Free TAK Server) has built-in anti-Denial-of-Service (DoS) support, which
restricts the number of CoT Events a client can send to a listening TCP Port.
Currently this FTS feature cannot be disabled or changed, so clients must meter
their input speed.

To use a PyTAK-based client with FTS, set the ``FTS_COMPAT`` configuration parameter
to ``True``. This will cause the PyTAK client to sleep a random number of seconds
between transmitting CoT to a FTS server::

. This will cause the PyTAK client to sleep a random number of seconds
between transmitting CoT to a FTS server:

Alternatively you can specify a static sleep period by setting ``PYTAK_SLEEP`` to an
integer number of seconds::

Version 1 of the TAK Protocol Payload is a Google Protocol Buffer based
payload.  Each Payload consists of one (and only one)
atakmap::commoncommo::v1::TakMessage message which is serialized using
Google protocol buffers version 3.

Source: https://github.com/deptofdefense/AndroidTacticalAssaultKit-CIV/blob/master/commoncommo/core/impl/protobuf/protocol.txt

PyTAK natively sends and receives "TAK Protocol Payload - Version 0", aka plain XML. If
you'd like to receive & decode "Version 1" protobuf with PyTAK, install the optional
`takproto <https://github.com/snstac/takproto>`_ Python module::

PyTAK natively sends and receives “TAK Protocol Payload - Version 0”, aka plain XML. If
you’d like to receive & decode “Version 1” protobuf with PyTAK, install the optional

Here is an example of receiving & decoding "Version 1" using ``takproto``.

N.B. The data type returned from this implementation differs from that of the
"Version 0" implementation (``bytes`` vs ``object``)::

“Version 0” implementation (

#!/usr/bin/env python3

import asyncio

from configparser import ConfigParser

import takproto

import pytak

class MyRXWorker(pytak.RXWorker):
    async def readcot(self):
        if hasattr(self.reader, 'readuntil'):
            cot = await self.reader.readuntil("</event>".encode("UTF-8"))
        elif hasattr(self.reader, 'recv'):
            cot, src = await self.reader.recv()
        tak_v1 = takproto.parse_proto(cot)
        if tak_v1 != -1:
            cot = tak_v1
        return cot

async def my_setup(clitool) -> None:
    reader, writer = await pytak.protocol_factory(clitool.config)
    write_worker = pytak.TXWorker(clitool.tx_queue, clitool.config, writer)
    read_worker = MyRXWorker(clitool.rx_queue, clitool.config, reader)
    clitool.add_task(write_worker)
    clitool.add_task(read_worker)

async def main():
    """
    The main definition of your program, sets config params and
    adds your serializer to the asyncio task list.
    """
    config = ConfigParser()
    config["mycottool"] = {"COT_URL": "udp://239.2.3.1:6969"}
    config = config["mycottool"]

    # Initializes worker queues and tasks.
    clitool = pytak.CLITool(config)
    await my_setup(clitool)

    # Start all tasks.
    await clitool.run()

if __name__ == "__main__":
    asyncio.run(main())

Destination for Cursor on Target messages. Defaults to ``udp://239.2.3.1:6969`` (ATAK Multicast UDP / Mesh SA Default)

TAK_PROTO
Sets TAK Protocol to use for CoT output, one of: 0 (XML), 2 (Mesh), 2 (Stream).

* Default: 0 (XML)

FTS_COMPAT
If set, implements random-seconds-sleep period to avoid FTS DoS protections.

* Default: False

PYTAK_SLEEP
If set, implements given sleep period of seconds between emitting CoT Events.

* Default: 0

PREF_PACKAGE
(If PyTAK is installed with optional with_crypto support.)

PyTAK supports importing TAK Data Packages containing TAK Server connection settings,
TLS certificates, etc.

To use a .zip file with PyTAK, set the ``PREF_PACKAGE`` config parameter to the
path to the .zip file.

For example, given a Pref Package named ``ADSB3_FIRE.zip``, you could either:

Using ``config.ini``: Add the line ``PREF_PACKAGE=ADSB3_FIRE.zip``

Using the commandline of a utility: Add the argument ``-p DSB3_FIRE.zip``

PyTAK can send & receive data over TLS by setting the following configuration
parameters (at a minimum)::

parameters (at a minimum):

Specify ``tls://`` in the CoT Destination URL, for example: ``tls://takserver.example.com:8089``

Client Certificates, Client Key, CA Certificate & Key must be specified in PEM format.

*N.B*: Encrypted private keys are not supported and must be saved in clear-text: ``openssl rsa -in my_cert.key.pem -out my_cert-nopass.key.pem``

: Encrypted private keys are not supported and must be saved in clear-text: 

PYTAK_TLS_CLIENT_CERT
Path to a file containing the Client Certificate for PyTAK. File must be
unencrypted plain-text PEM.

This file can contain both the Client Cert & Client Key, or the Client Cert alone. In
the later case (cert alone), ``PYTAK_TLS_CLIENT_KEY`` must be set to the Client Key.

For example, to connect to a TAK Server listening for TLS on port 8089::

    PYTAK_TLS_CLIENT_CERT=client_cert_and_key.pem
    COT_URL=tls://tak.example.com:8089

the later case (cert alone), 

For example, to connect to a TAK Server listening for TLS on port 8089:

PYTAK_TLS_CLIENT_KEY
Path to a file containing the Client Private Key for the associated
``PYTAK_TLS_CLIENT_CERT``. File must be unencrypted plain-text PEM.

PYTAK_TLS_DONT_VERIFY
Disable destination TLS Certificate Verification. Will print a WARNING if set.

PYTAK_TLS_DONT_CHECK_HOSTNAME
Disable destination TLS Certificate Common Name (CN) Verification. Will print a
WARNING if set.

PYTAK_TLS_CLIENT_CAFILE
Path to a file containing the CA Trust Store to use for remote certificate verification.

PYTAK_TLS_CLIENT_CIPHERS
Colon (":") seperated list of TLS Cipher Suites to allow.

For example: ``PYTAK_TLS_CLIENT_CIPHERS=ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384``

* Default: ``ALL``

The following Python 3.7+ code example creates a TAK Client that generates ``takPong``
CoT every 20 seconds, and sends them to a TAK Server at
``tcp://takserver.example.com:8087`` (plain / clear TCP).

The following Python 3.7+ code example creates a TAK Client that generates 

For secure TLS, see `TLS Support <https://github.com/snstac/pytak#tls-support>`_ below.

To run this example as-is, save the following code-block out to a file named
``example.py`` and run the command ``python3 example.py``::

import xml.etree.ElementTree as ET

class MySerializer(pytak.QueueWorker):
    """
    Defines how you process or generate your Cursor-On-Target Events.
    From there it adds the COT Events to a queue for TX to a COT_URL.
    """

    async def handle_data(self, data):
        """
        Handles pre-COT data and serializes to COT Events, then puts on queue.
        """
        event = data
        await self.put_queue(event)

    async def run(self, number_of_iterations=-1):
        """
        Runs the loop for processing or generating pre-COT data.
        """
        while 1:
            data = tak_pong()
            await self.handle_data(data)
            await asyncio.sleep(20)

def tak_pong():
    """
    Generates a simple takPong COT Event.
    """
    root = ET.Element("event")
    root.set("version", "2.0")
    root.set("type", "t-x-d-d")
    root.set("uid", "takPong")
    root.set("how", "m-g")
    root.set("time", pytak.cot_time())
    root.set("start", pytak.cot_time())
    root.set("stale", pytak.cot_time(3600))
    return ET.tostring(root)

    """
    """
    config["mycottool"] = {"COT_URL": "tcp://takserver.example.com:8087"}

    await clitool.setup()

    # Add your serializer to the asyncio task list.
    clitool.add_tasks(set([MySerializer(clitool.tx_queue, config)]))

CLITool(config: ~configparser.ConfigParser, tx_queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue | None = None, rx_queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue | None = None)

/Users/gba/src/SNS/pytak/pytak/classes.py:docstring of pytak.classes.CLITool

config: ~configparser.ConfigParser, tx_queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue | None = None, rx_queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue | None = None

CLITool.__init__(config: ~configparser.ConfigParser, tx_queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue | None = None, rx_queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue | None = None) -> None

__init__(config[, tx_queue, rx_queue])

Initialize CLITool instance.

add_task(task)

Add the given task to our coroutine task list.

add_tasks(tasks)

Add the given list or set of tasks to our couroutine task list.

create_workers(i_config)

Create and run queue workers with specified config parameters.

hello_event()

Send a 'hello world' style event to the Queue.

run()

Run this Thread and its associated coroutine tasks.

run_task(task)

Run the given coroutine task.

run_tasks([tasks])

Run the given list or set of couroutine tasks.

setup()

Set up CLITool.

:py:obj:`__init__ <pytak.CLITool.__init__>`\ \(config\[\, tx\_queue\, rx\_queue\]\)

:py:obj:`create_workers <pytak.CLITool.create_workers>`\ \(i\_config\)

QueueWorker(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict)

/Users/gba/src/SNS/pytak/pytak/classes.py:docstring of pytak.classes.QueueWorker

queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict

(`asyncio.Protocol` or similar async network client)
Serializes it as COT, and puts it onto an `asyncio.Queue`.

 or similar async network client)
Serializes it as COT, and puts it onto an 

Implementations should handle serializing messages as COT Events, and
putting them onto the `event_queue`.

putting them onto the 

QueueWorker.__init__(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict) -> None

__init__(queue, config)

Initialize a Worker instance.

fts_compat()

Apply FreeTAKServer (FTS) compatibility.

handle_data(data)

Handle data (placeholder method, please override).

put_queue(data[, queue_arg])

Put Data onto the Queue.

run([number_of_iterations])

Run this Thread, reads Data from Queue & passes data to next Handler.

:py:obj:`__init__ <pytak.QueueWorker.__init__>`\ \(queue\, config\)

:py:obj:`put_queue <pytak.QueueWorker.put_queue>`\ \(data\[\, queue\_arg\]\)

:py:obj:`run <pytak.QueueWorker.run>`\ \(\[number\_of\_iterations\]\)

RXWorker(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict, reader: ~asyncio.protocols.Protocol)

/Users/gba/src/SNS/pytak/pytak/classes.py:docstring of pytak.classes.RXWorker

queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict, reader: ~asyncio.protocols.Protocol

Reads events from a `pytak.protocol_factory()` reader and adds them to
an `rx_queue`.

Most implementations use this to drain an RX buffer on a socket.

RXWorker.__init__(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict, reader: ~asyncio.protocols.Protocol) -> None

__init__(queue, config, reader)

Initialize a RXWorker instance.

readcot()

Read CoT from the wire until we hit an event boundary.

Run this worker.

:py:obj:`__init__ <pytak.RXWorker.__init__>`\ \(queue\, config\, reader\)

:py:obj:`run <pytak.RXWorker.run>`\ \(\[number\_of\_iterations\]\)

TXWorker(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict, writer: ~asyncio.protocols.Protocol)

/Users/gba/src/SNS/pytak/pytak/classes.py:docstring of pytak.classes.TXWorker

queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict, writer: ~asyncio.protocols.Protocol

You should create an TXWorker Instance using the `pytak.txworker_factory()`
Function.

Data is put onto the Queue using a `pytak.QueueWorker()` instance.

TXWorker.__init__(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict, writer: ~asyncio.protocols.Protocol) -> None

__init__(queue, config, writer)

Initialize a TXWorker instance.

Accept CoT event from CoT event queue and process for writing.

send_data(data)

Send Data using the appropriate Protocol method.

:py:obj:`__init__ <pytak.TXWorker.__init__>`\ \(queue\, config\, writer\)

:py:obj:`run <pytak.TXWorker.run>`\ \(\[number\_of\_iterations\]\)

Worker(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict = None)

/Users/gba/src/SNS/pytak/pytak/classes.py:docstring of pytak.classes.Worker

queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict = None

Worker.__init__(queue: ~asyncio.queues.Queue | ~multiprocessing.context.BaseContext.Queue, config: None | ~configparser.SectionProxy | dict = None) -> None

__init__(queue[, config])

:py:obj:`__init__ <pytak.Worker.__init__>`\ \(queue\[\, config\]\)

:py:obj:`run <pytak.Worker.run>`\ \(\[number\_of\_iterations\]\)

create_udp_client(url: ~urllib.parse.ParseResult, iface: str | None = None, local_addr=None) -> ~typing.Tuple[~pytak.asyncio_dgram.aio.DatagramClient | None, ~pytak.asyncio_dgram.aio.DatagramClient]

[<desc_sig_keyword: <#text: 'async'>>, <desc_sig_space: <#text: ' '>>]

url: ~urllib.parse.ParseResult, iface: str | None = None, local_addr=None

~typing.Tuple[~pytak.asyncio_dgram.aio.DatagramClient | None, ~pytak.asyncio_dgram.aio.DatagramClient]

url : `ParseResult`
A parsed fully-qualified URL parsed with `urllib.parse.urlparse()`.
For example: udp://tak.example.com:4242

protocol_factory(config: ~configparser.SectionProxy) -> ~typing.Any

Class is for the protocol specified by the COT_URL parameter in the config object.

rxworker_factory(queue: ~asyncio.queues.Queue, config: ~configparser.SectionProxy) -> ~pytak.classes.RXWorker

queue: ~asyncio.queues.Queue, config: ~configparser.SectionProxy

txworker_factory(queue: ~asyncio.queues.Queue, config: ~configparser.SectionProxy) -> ~pytak.classes.TXWorker

.. image:: ../atak_screenshot_with_pytak_logo-x25.jpg
   :alt: ATAK Screenshot with PyTAK Logo.

Python Team Awareness Kit (PyTAK) is a Python Module for creating TAK clients,
servers & gateways and includes classes for handling Cursor on Target (CoT) & non-CoT
data, as well as functions for serializing CoT data, and sending and receiving CoT data
over a network.

Check out the :doc:`install` further information on installing this project. Configuration
parameters and their behavior can be found in the :doc:`config`.

 further information on installing this project. Configuration
parameters and their behavior can be found in the 

`pytak source code on Github <https://github.com/snstac/pytak>`_

PyTAK is distributed as a Debian package (``.deb``). PyTAK should be compatible with
most contemporary system-Python versions from Python 3.6 onward. The `Advanced Package
Tool (apt) <https://wiki.debian.org/PackageManagement>`_ is used to install this
and other related packages.

). PyTAK should be compatible with
most contemporary system-Python versions from Python 3.6 onward. The 

`Advanced Package
Tool (apt) <https://wiki.debian.org/PackageManagement>`_

To install PyTAK, download the pytak package and install using apt::

To install PyTAK, download the pytak package and install using apt:

$ sudo apt update -y
$ wget https://github.com/snstac/pytak/releases/latest/download/python3-pytak_latest_all.deb
$ sudo apt install -f ./python3-pytak_latest_all.deb

To install PyTAK with Deta Package support, you must also install the Python
`cryptography <https://cryptography.io/en/latest/installation/>`_ module using apt::

To install PyTAK with TAK Protocol Version 1 (protobuf) support, you must also install
the Python TAKProto module `takproto <https://github.com/snstac/takproto>`_.

the Python TAKProto module 

To install the TAKProto module, download the takproto package and install using apt::

To install the TAKProto module, download the takproto package and install using apt:

$ wget https://github.com/snstak/takproto/releases/latest/download/python3-takproto_latest_all.deb
$ sudo apt install -f ./python3-takproto_latest_all.deb

You can install from PyPI or from source. Both of these methods will require manual
installation of additional libraries.

1a. Debian, Ubuntu, Raspberry Pi: Install `LibFFI <https://sourceware.org/libffi/>`_::

1b. RedHat, CentOS: Install `LibFFI <https://sourceware.org/libffi/>`_::

$ python3 -m pip install pytak
$ python3 -m pip install pytak[with_crypto]
$ python3 -m pip install pytak[with_takproto]

$ git clone https://github.com/snstac/pytak.git
$ cd pytak/
$ python3 -m pip install .