### Basic CLITool Setup and Run Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Demonstrates how to initialize CLITool with a configuration, set up workers, add custom tasks, and run the event loop. This is a common pattern for starting a PyTAK application.
```python
import asyncio
from configparser import ConfigParser
import pytak
async def main():
config = ConfigParser()
config["mytool"] = {"COT_URL": "tcp://takserver.example.com:8087"}
clitool = pytak.CLITool(config["mytool"])
await clitool.setup() # Create TX/RX workers
# Add custom tasks
clitool.add_tasks(set([MyWorker(clitool.tx_queue, config["mytool"])]))
# Run event loop until exception
await clitool.run()
asyncio.run(main())
```
--------------------------------
### CLITool Setup Method Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Shows how to use the setup() method to create and configure TX/RX workers. Workers are automatically dispatched based on the COT_URL scheme.
```python
clitool = pytak.CLITool(config)
await clitool.setup()
# Workers are now added to clitool.tasks
```
--------------------------------
### CLI Example: Enroll via tak:// URL
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Usage of the pytak CLI to enroll with a TAK server using a tak:// deep-link URL, which automatically handles connection setup.
```bash
# Enroll via tak:// and auto-connect
pytak "tak://com.atakmap.app/enroll?host=takserver.example.com&username=user&token=secret"
```
--------------------------------
### Install pytak with All Features
Source: https://github.com/snstac/pytak/blob/main/_autodocs/INDEX.md
Installs pytak with all available features, including TLS, WebSocket, and Protobuf support.
```bash
# Everything
pip install pytak[with-crypto,with-aiohttp,with-takproto]
```
--------------------------------
### Install PyTAK
Source: https://github.com/snstac/pytak/blob/main/README.md
Install the PyTAK library using pip.
```sh
python3 -m pip install pytak
```
--------------------------------
### Quick Sender Example
Source: https://github.com/snstac/pytak/blob/main/README.md
A quick example demonstrating how to create a custom sender using PyTAK to send CoT data. This example sends a simple 'event' CoT every 20 seconds.
```python
import asyncio, xml.etree.ElementTree as ET
from configparser import ConfigParser
import pytak
class MySender(pytak.QueueWorker):
async def handle_data(self, data):
await self.put_queue(data)
async def run(self):
while True:
root = ET.Element("event", version="2.0", type="t-x-d-d",
uid="myMarker", how="m-g",
time=pytak.cot_time(), start=pytak.cot_time(),
stale=pytak.cot_time(3600))
await self.handle_data(ET.tostring(root))
await asyncio.sleep(20)
async def main():
config = ConfigParser()
config["mytool"] = {"COT_URL": "tcp://takserver.example.com:8087"}
config = config["mytool"]
clitool = pytak.CLITool(config)
await clitool.setup()
clitool.add_tasks(set([MySender(clitool.tx_queue, config)]))
await clitool.run()
asyncio.run(main())
```
--------------------------------
### Install pytak Minimal
Source: https://github.com/snstac/pytak/blob/main/_autodocs/INDEX.md
Installs the pytak package with only the essential features, excluding optional components.
```bash
# Minimal (no optional features)
pip install pytak
```
--------------------------------
### Install PyTAK with TLS Certificate Enrollment
Source: https://github.com/snstac/pytak/blob/main/_autodocs/API-OVERVIEW.md
Installs PyTAK with support for TLS certificate enrollment and aiohttp.
```bash
python3 -m pip install pytak[with-crypto,with-aiohttp]
```
--------------------------------
### Complete Data Package Creation Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-events.md
A comprehensive example demonstrating the creation of a TAK Data Package, including adding certificate files and connection preferences, and generating a .dpk file with a manifest.
```python
import pytak
import tempfile
import os
# Create a data package for ATAK
pkg = pytak.TAKDataPackage(
name="My Device Config",
uid="device-12345",
on_receive_delete=False
)
# Add certificate files
pkg.add_file("/home/user/.pytak/certs/client.p12", zip_entry_name="certs/client.p12")
pkg.add_file("/home/user/.pytak/certs/ca.pem", zip_entry_name="certs/ca.pem")
# Add connection preferences
pkg.add_file("/tmp/prefs.xml", zip_entry_name="prefs/connection.xml")
# Generate the package
output_dir = tempfile.gettempdir()
output_path = os.path.join(output_dir, "my-device-config.dpk")
pkg.create_package(output_path, use_dpk_extension=True, include_manifest=True)
print(f"Package created: {output_path}")
```
--------------------------------
### Install pytak with necessary dependencies
Source: https://github.com/snstac/pytak/blob/main/docs/onboarding.md
Install the pytak package with aiohttp and crypto support using pip.
```bash
python3 -m pip install pytak[with-aiohttp,with-crypto]
```
--------------------------------
### CLITool Hello Event Method Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Example of sending a "hello world" CoT event to the transmit queue. This can be disabled via configuration.
```python
await clitool.hello_event()
```
--------------------------------
### setup()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Asynchronously creates and configures TX/RX workers based on the COT_URL scheme. It automatically dispatches to the appropriate worker type.
```APIDOC
## setup()
### Description
Asynchronously creates and configures TX/RX workers based on the COT_URL scheme. It automatically dispatches to the appropriate worker type.
### Method
async def setup() -> None
### Parameters
None
### Returns
None
### Example
```python
clitool = pytak.CLITool(config)
await clitool.setup()
# Workers are now added to clitool.tasks
```
```
--------------------------------
### Install pytak with Protobuf Support
Source: https://github.com/snstac/pytak/blob/main/_autodocs/INDEX.md
Installs pytak with support for Protobuf encoding (TAK Protocol v1), requiring the `with-takproto` extra.
```bash
# With Protobuf (TAK Protocol v1)
pip install pytak[with-takproto]
```
--------------------------------
### Install TAK Protocol Payload (Debian)
Source: https://github.com/snstac/pytak/blob/main/docs/compatibility.md
Install the takproto Debian package for TAK Protocol v1 (Protobuf) support. This involves downloading the .deb file and installing it using apt.
```sh
wget https://github.com/snstac/takproto/releases/latest/download/takproto_latest_all.deb
sudo apt install -f ./takproto_latest_all.deb
```
--------------------------------
### Install PyTAK with All Optional Extras
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs PyTAK with all optional extras enabled, including crypto, takproto, and aiohttp support.
```sh
python3 -m pip install pytak[with_crypto,with_takproto,with_aiohttp]
```
--------------------------------
### Build and Preview PyTAK Docs Locally
Source: https://github.com/snstac/pytak/blob/main/CLAUDE.md
Install dependencies and serve the documentation locally for live preview. Alternatively, build static output.
```bash
pip install -r docs/requirements.txt
mkdocs serve # live-reload at http://127.0.0.1:8000
mkdocs build # static output to site/
```
--------------------------------
### Install PyTAK on Debian/Ubuntu
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs the PyTAK Debian package and its dependencies. Ensure your system is up-to-date before installation.
```sh
sudo apt update -qq
wget https://github.com/snstac/pytak/releases/latest/download/pytak_latest_all.deb
sudo apt install -f ./pytak_latest_all.deb
```
--------------------------------
### Send CoT via Marti REST API
Source: https://github.com/snstac/pytak/blob/main/docs/examples.md
This example demonstrates sending CoT messages to a TAK Server using the Marti HTTP API. It requires the `pytak[with_aiohttp]` installation and is useful when direct streaming ports are blocked by firewalls.
```python
import asyncio
import xml.etree.ElementTree as ET
from configparser import ConfigParser
import pytak
class MySender(pytak.QueueWorker):
async def handle_data(self, data):
await self.put_queue(data)
async def run(self):
while True:
root = ET.Element("event", version="2.0", type="t-x-d-d",
uid="martiMarker", how="m-g",
time=pytak.cot_time(), start=pytak.cot_time(),
stale=pytak.cot_time(3600))
await self.handle_data(ET.tostring(root))
await asyncio.sleep(10)
async def main():
config = ConfigParser()
config["mytool"] = {
"COT_URL": "marti://takserver.example.com:8443",
# Optional: provide client cert for mTLS
# "PYTAK_TLS_CLIENT_CERT": "/etc/pytak/client.pem",
}
config = config["mytool"]
clitool = pytak.CLITool(config)
await clitool.setup()
clitool.add_tasks(set([MySender(clitool.tx_queue, config)]))
await clitool.run()
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------
### Install PyTAK from Source
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Clones the PyTAK repository and installs it in editable mode using pip. This is useful for development or when needing the latest unreleased code.
```sh
git clone https://github.com/snstac/pytak.git
cd pytak/
python3 -m pip install -e .
```
--------------------------------
### Install Specific PyTAK Dependencies
Source: https://github.com/snstac/pytak/blob/main/_autodocs/errors.md
Install individual optional dependencies like cryptography, aiohttp, or takproto if needed.
```bash
pip install cryptography aiohttp takproto
```
--------------------------------
### Install pytak with TLS and WebSocket Support
Source: https://github.com/snstac/pytak/blob/main/_autodocs/INDEX.md
Installs pytak with support for TLS enrollment and WebSocket transport, requiring the `with-crypto` and `with-aiohttp` extras.
```bash
# With TLS enrollment support
pip install pytak[with-crypto,with-aiohttp]
```
--------------------------------
### Install PyTAK on Windows using pip
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs the PyTAK Python package from PyPI on Windows. Requires Python 3.7+ installed from python.org.
```powershell
python -m pip install pytak
```
--------------------------------
### Example INI Configuration
Source: https://github.com/snstac/pytak/blob/main/docs/configuration.md
This is an example of how to configure PyTAK using an INI file. It sets the CoT URL, TLS client certificate path, and debug mode.
```ini
[mytool]
COT_URL = tls://takserver.example.com:8089
PYTAK_TLS_CLIENT_CERT = /etc/pytak/client.pem
DEBUG = 0
```
--------------------------------
### Install Marti REST API Support
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs the 'python3-aiohttp' package, required for Marti REST API and certificate enrollment.
```sh
sudo apt install -y python3-aiohttp
```
--------------------------------
### TAK Enrollment URL Command-Line Usage
Source: https://github.com/snstac/pytak/blob/main/docs/examples.md
Demonstrates command-line usage for enrolling a client certificate using a tak:// onboarding URL. PyTAK handles enrollment and TLS setup automatically. The URL can include host and port for specific WebSocket/Marti connections.
```sh
pytak "tak://com.atakmap.app/enroll?host=takserver.example.com&username=myuser&token=mytoken"
```
```sh
pytak "tak://com.atakmap.app/enroll?host=takserver.example.com:8443&username=myuser&token=mytoken"
```
--------------------------------
### Install PyTAK Test Dependencies
Source: https://github.com/snstac/pytak/blob/main/CLAUDE.md
Installs the necessary dependencies required for running tests within the PyTAK project. This command reads from the specified requirements file.
```bash
python3 -m pip install -r requirements_test.txt
```
--------------------------------
### CLITool Run Method Example with Exception Handling
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Illustrates the main run loop, including sending a hello event, spawning tasks, and handling potential exceptions from worker failures.
```python
await clitool.setup()
clitool.add_tasks({...})
try:
await clitool.run()
except Exception as e:
print(f"Worker failed: {e}")
```
--------------------------------
### XML Element Type Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/types.md
Demonstrates how to generate an XML Element using `gen_cot_xml` and verify its type. Includes examples for serializing the XML Element to both string and bytes.
```python
import xml.etree.ElementTree as ET
event_elem = pytak.gen_cot_xml(lat=0.0, lon=0.0, uid="test")
assert isinstance(event_elem, ET.Element)
# Serialization
xml_str = ET.tostring(event_elem, encoding="unicode")
xml_bytes = ET.tostring(event_elem, encoding="utf-8")
```
--------------------------------
### Verify PyTAK Installation
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Checks if PyTAK is installed correctly by importing the library and printing its version number.
```python
python3 -c "import pytak; print(pytak.__version__)"
```
--------------------------------
### Install PyTAK with Optional Extras
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs PyTAK with specific optional features enabled using pip extras. Choose extras based on required functionality.
```sh
pip install pytak[with_crypto]
pip install pytak[with_takproto]
pip install pytak[with_aiohttp]
```
--------------------------------
### CLITool Run Tasks Method Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Demonstrates spawning and running all pending worker tasks. The task list is cleared after spawning.
```python
clitool.add_tasks(set([MySender(...), MyReceiver(...)]))
clitool.run_tasks() # Start all workers
```
--------------------------------
### End-to-End CoT Flow Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/INDEX.md
Illustrates the complete workflow for generating, queuing, transmitting, and receiving CoT events, including optional Protobuf conversion.
```python
1. Application generates CoT:
event = pytak.gen_cot(lat=0, lon=0, uid="test")
2. Put on TX queue:
await clitool.tx_queue.put(event)
3. TXWorker dequeues:
data = await tx_queue.get()
4. Convert to Protobuf if TAK_PROTO=1:
if use_protobuf:
data = takproto.xml2proto(data)
5. Send via protocol:
writer.write(data)
await writer.drain()
6. TAK Server receives
7. Server sends response CoT back
8. RXWorker receives:
data = await reader.readuntil(b"")
9. Convert from Protobuf if needed:
if use_protobuf:
data = takproto.parse_proto(data)
10. Put on RX queue:
await rx_queue.put(data)
11. Application receives:
event = await clitool.rx_queue.get()
```
--------------------------------
### CLI Example: TLS with Client Certificate
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Usage of the pytak CLI to connect to a TAK server over TLS, providing client certificate and key.
```bash
# TLS with client certificate
pytak --tls-cert client.pem --tls-key client-key.pem \
tls://takserver.example.com:8089
```
--------------------------------
### CLI Example: Bidirectional
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Basic usage of the pytak CLI for bidirectional communication, sending and receiving CoT events with a TAK server.
```bash
# Bidirectional (send and receive)
pytak tcp://takserver.example.com:8087
```
--------------------------------
### Install TAK Data Package Support
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs the 'python3-cryptography' package, required for importing TAK Data Packages (.zip pref packages).
```sh
sudo apt install -y python3-cryptography
```
--------------------------------
### Multi-Connection Setup
Source: https://github.com/snstac/pytak/blob/main/_autodocs/configuration.md
Set up multiple connections to different TAK servers by defining separate configuration sections. The IMPORT_OTHER_CONFIGS=1 directive allows importing settings from other configurations.
```ini
[DEFAULT]
IMPORT_OTHER_CONFIGS=1
[server1]
COT_URL=tcp://server1.example.com:8087
COT_HOST_ID=myapp-server1
[server2]
COT_URL=tcp://server2.example.com:8087
COT_HOST_ID=myapp-server2
```
--------------------------------
### Install PyTAK with TAK Protocol Payload support
Source: https://github.com/snstac/pytak/blob/main/docs/compatibility.md
Install the pytak package with the optional takproto module for TAK Protocol v1 (Protobuf) support. This is done using pip.
```sh
pip install pytak[with_takproto]
```
--------------------------------
### Install aiohttp Extra for Marti/TAK Transport
Source: https://github.com/snstac/pytak/blob/main/docs/troubleshooting.md
Install the required 'aiohttp' extra for using the `marti://` URL scheme and `tak://` certificate enrollment.
```sh
pip install pytak[with_aiohttp]
```
```sh
sudo apt install -y python3-aiohttp
```
--------------------------------
### Install libffi-devel on CentOS/RHEL
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs the 'libffi-devel' package, a system prerequisite for building certain Python dependencies with pip on CentOS/RHEL systems.
```sh
sudo yum install libffi-devel
```
--------------------------------
### Custom QueueWorker Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/worker-classes.md
Demonstrates how to create a custom QueueWorker that generates and sends CoT events periodically. This example shows a producer pattern where events are generated and put onto the transmit queue.
```python
import asyncio
import pytak
class MySender(pytak.QueueWorker):
async def handle_data(self, data):
# Not used in producer pattern
pass
async def run(self):
while True:
event = pytak.gen_cot(
uid="test-marker",
lat=0.0,
lon=0.0,
cot_type="t-x-d-d"
)
await self.put_queue(event)
await asyncio.sleep(10)
# Usage:
clitool = pytak.CLITool(config)
await clitool.setup()
clitool.add_tasks(set([MySender(clitool.tx_queue, config)]))
await clitool.run()
```
--------------------------------
### Install libffi-dev on Debian/Ubuntu
Source: https://github.com/snstac/pytak/blob/main/docs/installation.md
Installs the 'libffi-dev' package, a system prerequisite for building certain Python dependencies with pip on Debian-based systems.
```sh
sudo apt update -qq
sudo apt install -y libffi-dev
```
--------------------------------
### CLI Example: Receive-only
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Basic usage of the pytak CLI to listen for CoT events from a TAK server in receive-only mode.
```bash
# Receive-only (listen for CoT)
pytak --rx-only tcp://takserver.example.com:8087
```
--------------------------------
### Equivalent Environment Variables
Source: https://github.com/snstac/pytak/blob/main/docs/configuration.md
These environment variables provide the same configuration as the example INI file. They must be set before running the PyTAK script.
```sh
export COT_URL=tls://takserver.example.com:8089
export PYTAK_TLS_CLIENT_CERT=/etc/pytak/client.pem
export DEBUG=0
```
--------------------------------
### CLITool Add Tasks Method Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Shows how to add a set of worker instances to the CLITool's pending task list.
```python
clitool.add_tasks(set([MySender(...), MyReceiver(...)]))
```
--------------------------------
### CLITool Create Workers Method Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Demonstrates creating TX/RX workers for a specific configuration section, which is useful for managing multiple connections simultaneously.
```python
clitool = pytak.CLITool(config)
await clitool.create_workers(config["section1"])
await clitool.create_workers(config["section2"]) # Multi-connection
```
--------------------------------
### Install PyTAK in Editable Mode
Source: https://github.com/snstac/pytak/blob/main/CLAUDE.md
Installs the PyTAK library in editable mode, allowing for direct code changes to be reflected without reinstallation. This is the default target for the make command.
```bash
python3 -m pip install -e .
```
--------------------------------
### COTEvent Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-events.md
Demonstrates the creation and conversion of a COTEvent object. Use this to create CoT events with detailed positional accuracy information.
```python
import pytak
import xml.etree.ElementTree as ET
event = pytak.COTEvent(
lat=40.7128,
lon=-74.0060,
uid="event-001",
stale=120,
cot_type="a-f-G-E-S",
ce=100.0,
hae=10.0,
le=50.0
)
xml = event.to_xml()
xml_bytes = event.to_bytes()
```
--------------------------------
### StdinWorker Example: Stream Multiple Events
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Demonstrates streaming multiple CoT XML events from a file to a TAK server using the pytak CLI with the --tx-only flag.
```bash
# Stream multiple events from a file
cat events.xml | pytak tcp://takserver.example.com:8087 --tx-only
```
--------------------------------
### Enroll TAK and Get Certificate
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-functions.md
Perform TAK enrollment asynchronously using `enroll_tak` to obtain a certificate path and passphrase. Requires `cryptography` and `aiohttp` packages. The passphrase can be auto-generated if not provided.
```python
import asyncio
import pytak
async def main():
cert_path, passphrase = await pytak.enroll_tak(
host="takserver.example.com",
username="myuser",
password="mytoken"
)
print(f"Certificate saved to: {cert_path}")
print(f"Passphrase: {passphrase}")
asyncio.run(main())
```
--------------------------------
### CLI Example: Transmit from File
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Usage of the pytak CLI to send CoT events from a specified file to a TAK server in transmit-only mode.
```bash
# Send from a file
pytak --tx-file my-events.xml --tx-only tcp://takserver.example.com:8087
```
--------------------------------
### TAKDataPackage add_file() Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-events.md
Adds a file to the TAKDataPackage. Use this to include necessary files like certificates or configuration files within the data package. Specify a custom zip entry name if needed.
```python
import pytak
pkg = pytak.TAKDataPackage("My Package")
pkg.add_file("/path/to/client.p12", zip_entry_name="certs/client.p12")
pkg.add_file("/path/to/cert.pem", zip_entry_name="certs/ca.pem")
```
--------------------------------
### StdoutWorker Example: Pipe through xmllint
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Receives CoT events from a TAK server and pipes them through xmllint for pretty-printing using the pytak CLI.
```bash
# Pipe received CoT through xmllint
pytak tcp://takserver.example.com:8087 --rx-only | xmllint --format -
```
--------------------------------
### Run the PyTAK sender script
Source: https://github.com/snstac/pytak/blob/main/docs/quickstart.md
Execute the Python script to start sending CoT events. Remember to update the COT_URL in the script to your TAK Server's address.
```sh
python3 send.py
```
--------------------------------
### StdinWorker Example: Send Single Event
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Demonstrates sending a single CoT XML event from standard input to a TAK server using the pytak CLI.
```bash
# Send a single event from stdin
echo '' \
| pytak tcp://takserver.example.com:8087
```
--------------------------------
### StdoutWorker Example: Receive and Print
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Receives CoT events from a TAK server and prints them to the terminal using the pytak CLI with the --rx-only flag.
```bash
# Receive CoT from server and print to terminal
pytak tcp://takserver.example.com:8087 --rx-only
```
--------------------------------
### StdoutWorker Example: Pipe through multiple servers
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Demonstrates piping CoT events received from one TAK server to another using two instances of the pytak CLI.
```bash
# Pipe CoT through multiple servers
pytak --rx-only tcp://server1:8087 | pytak --tx-only tcp://server2:8087
```
--------------------------------
### FileWorker Example: Send Events with TLS
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Sends CoT events from a file to a TAK server over TLS, specifying client certificate and key, using the pytak CLI.
```bash
# Send events with suppressed output (no RX)
pytak --tx-file /path/to/CoT.xml tls://takserver.example.com:8089 --tls-cert client.pem
```
--------------------------------
### Use PyTAK Standard CLI Entry Point
Source: https://github.com/snstac/pytak/blob/main/_autodocs/API-OVERVIEW.md
Illustrates using the standard `pytak.cli()` entry point for applications, allowing configuration via environment variables or config files. Requires a `create_tasks` factory function.
```python
# my_tool.py
import pytak
def create_tasks(config, clitool):
"""Factory function called by pytak.cli()."""
return set([MySender(clitool.tx_queue, config)])
DEFAULT_COT_STALE = "120"
if __name__ == "__main__":
pytak.cli("my_tool")
# Usage:
# COT_URL=tcp://takserver.example.com:8087 python my_tool.py
# python my_tool.py -c config.ini
# TAK_URL="tak://..." python my_tool.py
```
--------------------------------
### FileWorker Example: Send Events from File
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Sends CoT events from a local file to a TAK server using the pytak CLI with --tx-file and --tx-only flags.
```bash
# Send events from a file
pytak --tx-file /tmp/events.xml tcp://takserver.example.com:8087 --tx-only
```
--------------------------------
### Import Additional Configuration Sections
Source: https://github.com/snstac/pytak/blob/main/_autodocs/configuration.md
Enable the import of additional configuration sections for multi-connection setups. When enabled, CLITool creates workers for all defined config sections.
```ini
[DEFAULT]
IMPORT_OTHER_CONFIGS=1
[server1]
COT_URL=tcp://server1.example.com:8087
[server2]
COT_URL=tcp://server2.example.com:8087
```
--------------------------------
### Connect with TAK Onboarding URL (CLI)
Source: https://github.com/snstac/pytak/blob/main/README.md
Connect to a TAK server directly from the command line using a TAK onboarding URL.
```sh
pytak "tak://com.atakmap.app/enroll?host=takserver.example.com&username=myuser&token=mytoken"
```
--------------------------------
### Run PyTAK CLI Help
Source: https://github.com/snstac/pytak/blob/main/docs/docker.md
Runs the PyTAK container and displays the CLI help message. Replace `` with the desired release tag.
```sh
docker run --rm ghcr.io/snstac/pytak-deb: --help
```
--------------------------------
### Enroll TAK user and generate data packages
Source: https://github.com/snstac/pytak/blob/main/docs/onboarding.md
Use the `pytak dp` command with a TAK enrollment deep-link to generate connection data packages. The output paths are printed to stdout.
```bash
pytak dp 'tak://com.atakmap.app/enroll?host=takserver.example.com&username=myuser&token=SECRET'
```
--------------------------------
### Basic pytak CLI Execution
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
This snippet shows the basic entry point for running a pytak tool from the command line. It imports the pytak library and calls the CLI function with a tool name.
```python
if __name__ == "__main__":
import pytak
pytak.cli("my_tool")
```
--------------------------------
### Run pytak with TAK Server Enrollment
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
Execute a pytak tool, enrolling with a TAK server and establishing a connection. The TAK_URL environment variable must be configured with enrollment details.
```bash
# Enroll via TAK server and connect
TAK_URL="tak://com.atakmap.app/enroll?host=takserver.example.com&username=user&token=secret" python my_tool.py
```
--------------------------------
### Generate Onboarding Data Packages (CLI)
Source: https://github.com/snstac/pytak/blob/main/README.md
Generate onboarding data packages (PKCS#12, PEM, ATAK + iTAK ZIPs) without connecting to a TAK server.
```sh
pytak dp 'tak://com.atakmap.app/enroll?host=takserver.example.com&username=myuser&token=mytoken'
```
--------------------------------
### cli
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
The standard entry point for PyTAK-based applications, handling argument parsing, configuration merging, TAK URL enrollment, worker creation, and event loop management.
```APIDOC
## cli()
### Description
Standard entry point for PyTAK-based applications.
### Method
def cli(app_name: str) -> None
### Parameters
#### Path Parameters
None
#### Query Parameters
None
#### Request Body
None
### Parameters
- **app_name** (str) - Required - Application module name (e.g. "my_tool")
### Handles:
1. Argument parsing (`-c CONFIG_FILE`, `-p PREF_PACKAGE`)
2. Configuration merging (environment variables → config.ini → pref packages)
3. TAK URL enrollment (if COT_URL starts with `tak://`)
4. Worker creation via the app's `create_tasks(config, clitool)` function
5. Event loop management
### Expected Application Structure:
```python
# my_tool/__init__.py or my_tool.py
def create_tasks(config, clitool):
"""Return set of tasks for the CLITool."""
return set([MyWorker(clitool.tx_queue, config)])
# Optional: custom defaults
DEFAULT_COT_STALE = "120"
```
```
--------------------------------
### Enroll with tak:// URL using Docker
Source: https://github.com/snstac/pytak/blob/main/docs/docker.md
Runs PyTAK in a container to enroll with a TAK server using a tak:// URL. Mounts a local directory for certificates to persist them between runs. Replace `` with the desired release tag.
```sh
docker run --rm \
-v "$PWD/pytak-certs:/home/pytak/.pytak/certs" \
ghcr.io/snstac/pytak-deb: \
"tak://com.atakmap.app/enroll?host=takserver.example.com&username=user&token=token"
```
--------------------------------
### SimpleCOTEvent to_xml() Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-events.md
Returns the SimpleCOTEvent as an xml.etree.ElementTree.Element. This is useful for further XML manipulation or parsing.
```python
import pytak
import xml.etree.ElementTree as ET
event = pytak.SimpleCOTEvent(lat=40.0, lon=-74.0, uid="marker1")
elem = event.to_xml()
ET.dump(elem)
```
--------------------------------
### Enable Protobuf for TAK Protocol v1
Source: https://github.com/snstac/pytak/blob/main/_autodocs/API-OVERVIEW.md
Configuration setting to enable the Protobuf wire format for TAK Protocol v1. Requires the 'takproto' package.
```ini
TAK_PROTO=1 # Enable Protobuf
```
--------------------------------
### run_tasks()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Spawns all pending worker tasks and starts their `run()` coroutines. The pending task list is cleared after spawning.
```APIDOC
## run_tasks()
### Description
Spawns all pending worker tasks and starts their `run()` coroutines. The pending task list is cleared after spawning.
### Method
def run_tasks(self, tasks: Set[Worker] = None) -> None
### Parameters
#### Path Parameters
None
#### Query Parameters
None
#### Request Body
None
### Parameters
- **tasks** (Set[Worker]) - Optional - Tasks to run (default: all pending)
### Example
```python
clitool.add_tasks(set([MySender(...), MyReceiver(...)]))
clitool.run_tasks() # Start all workers
```
```
--------------------------------
### PyTAK CLI Entry Point
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
The `cli` function serves as the standard entry point for PyTAK applications, handling argument parsing, configuration merging, TAK URL enrollment, worker creation, and event loop management.
```python
def cli(app_name: str) -> None:
# ... implementation details ...
pass
```
```python
# my_tool/__init__.py or my_tool.py
def create_tasks(config, clitool):
"""Return set of tasks for the CLITool."""
return set([MyWorker(clitool.tx_queue, config)])
# Optional: custom defaults
DEFAULT_COT_STALE = "120"
```
--------------------------------
### PyTAK Command-Line Tool: Enroll and Auto-connect
Source: https://github.com/snstac/pytak/blob/main/_autodocs/API-OVERVIEW.md
Use the `pytak` command-line tool to enroll with a TAK server using a deep-link URL and automatically connect.
```bash
pytak "tak://com.atakmap.app/enroll?host=...&username=...&token=..."
```
--------------------------------
### Create UDP Client
Source: https://github.com/snstac/pytak/blob/main/_autodocs/INDEX.md
Initializes a UDP client connection to a specified URL.
```python
create_udp_client(url)
```
--------------------------------
### SimpleCOTEvent to_bytes() Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-events.md
Returns the SimpleCOTEvent as UTF-8 encoded CoT XML bytes. This is necessary for transmitting the event over a network or queue.
```python
import pytak
event = pytak.SimpleCOTEvent(lat=40.0, lon=-74.0, uid="marker1")
xml_bytes = event.to_bytes()
# await clitool.tx_queue.put(xml_bytes) # Example of sending bytes
```
--------------------------------
### Programmatic enrollment and data package generation
Source: https://github.com/snstac/pytak/blob/main/docs/onboarding.md
Use the `enroll_onboarding_package` function from `pytak.onboarding_packages` to generate data packages within your Python application. The path to the generated ZIP file is available in the result.
```python
import asyncio
from pytak.onboarding_packages import enroll_onboarding_package
result = asyncio.run(
enroll_onboarding_package(
"tak://com.atak.app/enroll?host=takserver.example.com&username=u&token=t",
"data-packages",
)
)
print(result["data_package_zip"])
```
--------------------------------
### Run pytak with ATAK Preferences
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
Execute a pytak tool, loading connection settings from an ATAK preferences package zip file using the '-p' flag.
```bash
# Load connection settings from ATAK pref package
python my_tool.py -p ~/device-config.zip
```
--------------------------------
### Enable Debug Mode
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Run pytak in debug mode to get verbose logging output. This is helpful for troubleshooting connection issues.
```bash
pytak --debug tcp://takserver.example.com:8087
```
--------------------------------
### enroll_tak()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-functions.md
Performs TAK enrollment with a TAK Server, returning the path to the generated certificate and its passphrase. Requires 'cryptography' and 'aiohttp' packages.
```APIDOC
## enroll_tak()
### Description
Perform TAK enrollment and return the certificate path and passphrase.
### Signature
`async def enroll_tak(host: str, username: str, password: str, passphrase: Optional[str] = None) -> Tuple[str, str]`
### Parameters
#### Path Parameters
- **host** (str) - Required - TAK Server hostname
- **username** (str) - Required - Enrollment username
- **password** (str) - Required - Enrollment password/token
- **passphrase** (Optional[str]) - Optional - PKCS#12 certificate passphrase
### Returns
Tuple of (cert_path, passphrase)
### Raises
Exception on enrollment failure
### Requires
`cryptography` and `aiohttp` packages
### Example
```python
import asyncio
import pytak
async def main():
cert_path, passphrase = await pytak.enroll_tak(
host="takserver.example.com",
username="myuser",
password="mytoken"
)
print(f"Certificate saved to: {cert_path}")
print(f"Passphrase: {passphrase}")
asyncio.run(main())
```
```
--------------------------------
### CLITool Add Task Method Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Illustrates adding a single worker task to the CLITool's pending task list before execution.
```python
clitool.add_task(MySender(clitool.tx_queue, config))
```
--------------------------------
### Handle ImportError for Missing Optional Packages
Source: https://github.com/snstac/pytak/blob/main/_autodocs/errors.md
Catch ImportError when optional packages required for specific PyTAK features (like aiohttp or cryptography) are not installed.
```python
try:
await pytak.marti_txworker_factory(queue, config)
except ImportError as e:
print(f"Missing package: {e}")
```
--------------------------------
### Single TAK Server Connection with Producer
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Connects to a single TAK server and demonstrates how to create a producer to send CoT events. Ensure the `COT_URL` and `COT_HOST_ID` are correctly configured.
```python
import asyncio
from configparser import ConfigParser
import pytak
class MyProducer(pytak.QueueWorker):
async def handle_data(self, data):
pass
async def run(self):
while True:
event = pytak.gen_cot(uid="marker1", lat=0.0, lon=0.0)
await self.put_queue(event)
await asyncio.sleep(5)
async def main():
config = ConfigParser()
config["app"] = {
"COT_URL": "tcp://takserver.example.com:8087",
"COT_HOST_ID": "myapp@localhost",
}
clitool = pytak.CLITool(config["app"])
await clitool.setup()
clitool.add_tasks({MyProducer(clitool.tx_queue, config["app"]}})
await clitool.run()
asyncio.run(main())
```
--------------------------------
### Run pytak with Custom Config File
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
Execute a pytak tool, specifying a custom configuration file path using the '-c' flag.
```bash
# Read config from a specific file
python my_tool.py -c /etc/my_tool/config.ini
```
--------------------------------
### run()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
The main asynchronous run loop for the CLITool. It sends an initial hello event (if enabled), spawns all added tasks, and monitors for exceptions. It cleans up resources upon task failure.
```APIDOC
## run()
### Description
The main asynchronous run loop for the CLITool. It sends an initial hello event (if enabled), spawns all added tasks, and monitors for exceptions. It cleans up resources upon task failure.
### Method
async def run() -> None
### Parameters
None
### Returns
None (raises exception on worker failure)
### Example
```python
await clitool.setup()
clitool.add_tasks({...})
try:
await clitool.run()
except Exception as e:
print(f"Worker failed: {e}")
```
```
--------------------------------
### SimpleCOTEvent __str__() Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-events.md
Returns a formatted string representation of the SimpleCOTEvent dataclass as CoT XML. This is useful for displaying the event in a human-readable XML format.
```python
import pytak
event = pytak.SimpleCOTEvent(lat=40.0, lon=-74.0, uid="marker1", cot_type="t-x-d-d")
print(event)
```
--------------------------------
### CLI Example: Transmit-only from Stdin
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Basic usage of the pytak CLI to send CoT events from standard input to a TAK server in transmit-only mode.
```bash
# Transmit-only (send CoT from stdin)
echo '' \
| pytak --tx-only tcp://takserver.example.com:8087
```
--------------------------------
### hello_event()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/clitool.md
Asynchronously sends a "hello world" style CoT event to the transmit queue. This is disabled if `PYTAK_NO_HELLO=1` is set in the configuration.
```APIDOC
## hello_event()
### Description
Asynchronously sends a "hello world" style CoT event to the transmit queue. This is disabled if `PYTAK_NO_HELLO=1` is set in the configuration.
### Method
async def hello_event() -> None
### Parameters
None
### Returns
None
### Example
```python
await clitool.hello_event()
```
```
--------------------------------
### TAK Server TLS Input Stanza
Source: https://github.com/snstac/pytak/blob/main/docs/configuration.md
Example XML configuration for a TAK Server to accept TLS connections on port 8089 using x509 authentication.
```xml
```
--------------------------------
### Configure PyTAK with Environment Variables
Source: https://github.com/snstac/pytak/blob/main/docs/docker.md
Runs the PyTAK RPM container, passing environment variables for configuration like DEBUG mode and COT_URL. Replace `` with the desired release tag.
```sh
docker run --rm \
-e DEBUG=1 \
-e COT_URL=tcp://takserver.example.com:8087 \
ghcr.io/snstac/pytak-rpm:
```
--------------------------------
### Configure TAK Onboarding URL
Source: https://github.com/snstac/pytak/blob/main/docs/configuration.md
Set the COT_URL to a tak:// enrollment deep-link URL. This is used for TAK Server enrollment, similar to ATAK's QR-code onboarding.
```ini
COT_URL = tak://com.atakmap.app/enroll?host=takserver.example.com&username=myuser&token=mytoken
```
--------------------------------
### StdoutWorker Example: Filter events by type
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cli-workers.md
Receives CoT events from a TAK server and filters them by type using grep, demonstrating compatibility with Unix pipe chains.
```bash
# Filter events by type
pytak tcp://takserver.example.com:8087 --rx-only | grep 'type="a-f'
```
--------------------------------
### Run pytak with UDP Multicast
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
Execute a pytak tool, reading configuration from 'config.ini' and sending data to a UDP multicast address. Ensure the COT_URL environment variable is set.
```bash
# Read config from config.ini, send to UDP multicast
COT_URL=udp+wo://239.2.3.1:6969 python my_tool.py
```
--------------------------------
### Disable Hello Event on Startup
Source: https://github.com/snstac/pytak/blob/main/_autodocs/configuration.md
Set PYTAK_NO_HELLO to 1 to prevent PyTAK from sending a hello event when CLITool starts. This is useful for environments where hello events are not desired.
```ini
PYTAK_NO_HELLO=1
```
--------------------------------
### cot_time()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-functions.md
Get the current UTC datetime as a W3C XML Schema dateTime primitive. This function can also return a future time by adding a specified number of seconds.
```APIDOC
## cot_time()
### Description
Get the current UTC datetime as a W3C XML Schema dateTime primitive. This function can also return a future time by adding a specified number of seconds.
### Signature
`def cot_time(cot_stale: Union[int, None] = None) -> str`
### Parameters
#### Query Parameters
- **cot_stale** (Union[int, None]) - Optional - None - Seconds to add to current time (for stale timeouts)
### Returns
ISO 8601 UTC datetime string (W3C XML Schema format: `YYYY-MM-DDTHH:MM:SS.fffZ`)
### Example
```python
import pytak
# Current time
now = pytak.cot_time() # "2026-06-03T12:34:56.123Z"
# Time 120 seconds in the future (for stale timeout)
stale = pytak.cot_time(120) # "2026-06-03T12:36:56.123Z"
```
```
--------------------------------
### MartiTXWorker Payload Example
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/worker-classes.md
Illustrates the JSON payload structure for sending CoT events via the Marti REST API. This is used when injecting CoT data to a TAK Server.
```json
{"uid": "", "toInject": ""}
```
--------------------------------
### hello_event()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-functions.md
Generate a Hello CoT Event (takPing). This function can optionally take a custom unique identifier.
```APIDOC
## hello_event()
### Description
Generate a Hello CoT Event ("takPing"). This function can optionally take a custom unique identifier.
### Signature
`def hello_event(uid: Optional[bytes] = None) -> bytes`
### Parameters
#### Query Parameters
- **uid** (Optional[bytes]) - Optional - "takPing" - Unique identifier for the hello event
### Returns
CoT XML bytes
### Example
```python
import pytak
hello = pytak.hello_event() # Default uid="takPing"
hello_custom = pytak.hello_event(b"myapp-ping")
```
```
--------------------------------
### Standard CLI Harness
Source: https://github.com/snstac/pytak/blob/main/_autodocs/INDEX.md
Provides a standard command-line interface harness for applications.
```python
cli(app_name)
```
--------------------------------
### Create Network Client with protocol_factory
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/factory-functions.md
Use protocol_factory to create network reader and writer clients based on the COT_URL scheme. It dispatches based on the URL's scheme.
```python
config_dict = {"COT_URL": "tcp://takserver.example.com:8087"}
reader, writer = await pytak.protocol_factory(config)
```
--------------------------------
### Generate Current UTC Time for CoT
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-functions.md
Get the current UTC datetime in W3C XML Schema format. Use the `cot_stale` parameter to specify seconds to add for stale timeouts.
```python
import pytak
# Current time
now = pytak.cot_time() # "2026-06-03T12:34:56.123Z"
# Time 120 seconds in the future (for stale timeout)
stale = pytak.cot_time(120) # "2026-06-03T12:36:56.123Z"
```
--------------------------------
### Generate a Hello CoT Event
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-functions.md
Create a 'takPing' CoT event. An optional `uid` can be provided for a custom identifier.
```python
import pytak
hello = pytak.hello_event() # Default uid="takPing"
hello_custom = pytak.hello_event(b"myapp-ping")
```
--------------------------------
### Programmatically Create TAK Data Packages
Source: https://github.com/snstac/pytak/blob/main/_autodocs/API-OVERVIEW.md
Create TAK data packages programmatically using the `TAKDataPackage` class. This allows for adding files and creating the package with a specified extension.
```python
pkg = pytak.TAKDataPackage("My Device", uid="device-001")
pkg.add_file("/path/to/client.p12", zip_entry_name="certs/client.p12")
pkg.create_package("output.dpk", use_dpk_extension=True)
```
--------------------------------
### create_package()
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/cot-events.md
Creates the TAK Data Package zip file. This method can optionally use the .dpk extension and include a manifest.xml file.
```APIDOC
## create_package()
### Description
Create the TAK Data Package zip file.
### Method Signature
```python
def create_package(self, output_path: str, use_dpk_extension: bool = False, include_manifest: bool = True)
```
### Parameters
#### Path Parameters
- **output_path** (str) - Required - Path where to save the package
- **use_dpk_extension** (bool) - Optional - Use `.dpk` extension instead of `.zip`. Defaults to False.
- **include_manifest** (bool) - Optional - Whether to include manifest.xml. Defaults to True.
### Raises
- `ValueError` if no files have been added
### Example
```python
pkg = pytak.TAKDataPackage("Device Config", uid="my-device-001")
pkg.add_file("/path/to/client.p12")
pkg.add_file("/path/to/prefs.xml")
pkg.create_package("/tmp/device-config.zip")
# Output: TAK Data Package created: /tmp/device-config.zip
```
```
--------------------------------
### Initialize RXWorker
Source: https://github.com/snstac/pytak/blob/main/_autodocs/api-reference/worker-classes.md
Instantiate RXWorker for receiving CoT events from the network. Requires a queue, configuration, and a network reader object.
```python
rx = pytak.RXWorker(queue, config, reader)
```
--------------------------------
### Send CoT from File using Docker
Source: https://github.com/snstac/pytak/blob/main/docs/docker.md
Runs PyTAK in a container to send CoT events from a local XML file to a TAK server. Mounts the local events file into the container. Replace `` with the desired release tag.
```sh
docker run --rm \
-v "$PWD/events.xml:/data/events.xml:ro" \
ghcr.io/snstac/pytak-deb: \
--tx-file /data/events.xml tcp://takserver.example.com:8087
```