Examples
Send TAK Data¶
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).
To run this example as-is, save the following code-block out to a file named
send.py
and run the command python3 send.py
::
#!/usr/bin/env python3
import asyncio
import xml.etree.ElementTree as ET
from configparser import ConfigParser
import pytak
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):
"""Handle pre-CoT data, serialize to CoT Event, then puts on queue."""
event = data
await self.put_queue(event)
async def run(self):
"""Run the loop for processing or generating pre-CoT data."""
while True:
data = tak_pong()
await self.handle_data(data)
await asyncio.sleep(20)
def tak_pong():
"""Generate 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)
async def main():
"""Main definition of your program, sets config params and
adds your serializer to the asyncio task list.
"""
config = ConfigParser()
config["mycottool"] = {"COT_URL": "tcp://takserver.example.com:8087"}
config = config["mycottool"]
# Initializes worker queues and tasks.
clitool = pytak.CLITool(config)
await clitool.setup()
# Add your serializer to the asyncio task list.
clitool.add_tasks(set([MySerializer(clitool.tx_queue, config)]))
# Start all tasks.
await clitool.run()
if __name__ == "__main__":
asyncio.run(main())
Send & Receive TAK Data¶
TK TK TK
To run this example as-is, save the following code-block out to a file named
send_receive.py
and run the command python3 send_receive.py
::
#!/usr/bin/env python3
import asyncio
import xml.etree.ElementTree as ET
import pytak
from configparser import ConfigParser
def gen_cot():
"""Generate CoT Event."""
root = ET.Element("event")
root.set("version", "2.0")
root.set("type", "a-h-A-M-A") # insert your type of marker
root.set("uid", "name_your_marker")
root.set("how", "m-g")
root.set("time", pytak.cot_time())
root.set("start", pytak.cot_time())
root.set(
"stale", pytak.cot_time(60)
) # time difference in seconds from 'start' when stale initiates
pt_attr = {
"lat": "40.781789", # set your lat (this loc points to Central Park NY)
"lon": "-73.968698", # set your long (this loc points to Central Park NY)
"hae": "0",
"ce": "10",
"le": "10",
}
ET.SubElement(root, "point", attrib=pt_attr)
return ET.tostring(root)
class MySender(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):
"""Handle pre-CoT data, serialize to CoT Event, then puts on queue."""
event = data
await self.put_queue(event)
async def run(self):
"""Run the loop for processing or generating pre-CoT data."""
while True:
data = gen_cot()
self._logger.info("Sending:\n%s\n", data.decode())
await self.handle_data(data)
await asyncio.sleep(5)
class MyReceiver(pytak.QueueWorker):
"""Defines how you will handle events from RX Queue."""
async def handle_data(self, data):
"""Handle data from the receive queue."""
self._logger.info("Received:\n%s\n", data.decode())
async def run(self):
"""Read from the receive queue, put data onto handler."""
while True:
data = (
await self.queue.get()
) # this is how we get the received CoT from rx_queue
await self.handle_data(data)
async def main():
"""Main definition of your program, sets config params and
adds your serializer to the asyncio task list.
"""
config = ConfigParser()
config["mycottool"] = {"COT_URL": "tcp://takserver.example.com:8087"}
config = config["mycottool"]
# Initializes worker queues and tasks.
clitool = pytak.CLITool(config)
await clitool.setup()
# Add your serializer to the asyncio task list.
clitool.add_tasks(
set([MySender(clitool.tx_queue, config), MyReceiver(clitool.rx_queue, config)])
)
# Start all tasks.
await clitool.run()
if __name__ == "__main__":
asyncio.run(main())