Skip to content

Instantly share code, notes, and snippets.

@ampledata
Last active April 4, 2023 19:22
Show Gist options
  • Save ampledata/825f9e9c323c8dfdc925551dafe7f932 to your computer and use it in GitHub Desktop.
Save ampledata/825f9e9c323c8dfdc925551dafe7f932 to your computer and use it in GitHub Desktop.
PyTAK TAK Server receiver
#!/usr/bin/env python3
"""Example PyTAK Receiver."""
import asyncio
import xml.etree.ElementTree as ET
from configparser import ConfigParser
import pytak
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 data: %s", data)
async def run(self): # pylint: disable=arguments-differ
"""Read from the receive queue, put data onto handler."""
while 1:
data = await self.queue.get()
await self.handle_data(data)
def tak_pong():
"""Generate a 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():
"""Definition of your program.
Sets config params and adds your receiver 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 clitool.setup()
# Add your receiver to the asyncio task list.
clitool.add_tasks(set([MyReceiver(clitool.rx_queue, config)]))
# Start all tasks.
await clitool.run()
if __name__ == "__main__":
asyncio.run(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment