PyTAK TAK Server receiver

example-pytak_rx.py

#!/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())