Telemetry Data Push Reference

`xovis.datapush.tcp_server`

Xovis SDK - Data Plane TCP Server

This module implements a high-speed TCP ingestion engine for Xovis telemetry. It handles raw concatenated JSON datapush using a sliding buffer and json.JSONDecoder().raw_decode() for zero-copy extraction, adhering to the Data Plane's throughput requirements.

Classes

`XovisTCPServer`

High-speed TCP Ingestion Engine for Xovis Telemetry.

Manages persistent TCP connections from Xovis sensors. Implements a specialized sliding buffer to extract concatenated JSON frames without delimiters, dispatching them to attached sinks with minimal latency.

Source code in src/xovis/datapush/tcp_server.py

class XovisTCPServer:
    """
    High-speed TCP Ingestion Engine for Xovis Telemetry.

    Manages persistent TCP connections from Xovis sensors. Implements a
    specialized sliding buffer to extract concatenated JSON frames without
    delimiters, dispatching them to attached sinks with minimal latency.
    """

    def __init__(self):
        """
        Initializes the XovisTCPServer.
        """
        self.sinks: list[XovisSink] = []

    def attach_sink(self, sink: XovisSink) -> "XovisTCPServer":
        """
        Attaches a telemetry consumer to the server.

        Args:
            sink (XovisSink): An object implementing the XovisSink protocol.

        Returns:
            XovisTCPServer: The server instance for method chaining.
        """
        self.sinks.append(sink)
        return self

    async def stop(self):
        """
        Gracefully tears down the TCP server.
        """
        if hasattr(self, "_server") and self._server:
            self._server.close()
            await self._server.wait_closed()
            logger.info("Xovis TCP Server stopped")

    async def start(self, host: str = "0.0.0.0", port: int = 9000):  # nosec B104
        """
        Starts the TCP server and begins listening for sensor connections.

        Args:
            host (str, optional): The network interface to bind to. Defaults to "0.0.0.0".
            port (int, optional): The TCP port to listen on. Defaults to 9000.

        Raises:
            OSError: If the port is already in use or binding fails.
        """
        self._server = await asyncio.start_server(self._handle_client, host, port)
        logger.info(f"Xovis TCP Server listening on {host}:{port}")
        async with self._server:
            await self._server.serve_forever()

    async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        """
        Handles an individual sensor connection lifecycle.

        Uses a sliding string buffer combined with `raw_decode` to safely
        extract concatenated JSON objects from the raw byte stream.

        Args:
            reader (asyncio.StreamReader): The client stream reader.
            writer (asyncio.StreamWriter): The client stream writer.
        """
        peer = writer.get_extra_info("peername")
        logger.info(f"Sensor connected: {peer}")

        # Check if we should log to studio debug log
        # We look for any attached MetricSink with debug=True
        studio_debug = False
        if hasattr(self, "sinks"):
            for sink in self.sinks:
                if hasattr(sink, "debug") and sink.debug:
                    studio_debug = True
                    break

        if studio_debug:
            try:
                with open("xovis_studio_debug.log", "a", encoding="utf-8") as f:
                    f.write(f"--- Sensor connected: {peer} ---\n")
            except Exception:
                pass

        buffer = ""
        decoder = json.JSONDecoder()

        try:
            while True:
                try:
                    chunk = await asyncio.wait_for(reader.read(8192), timeout=60.0)
                except asyncio.TimeoutError:
                    logger.debug(f"Read timeout for {peer}")
                    continue

                if not chunk:
                    logger.info(f"Connection closed by peer: {peer}")
                    break

                logger.debug(f"Received chunk from {peer}: {len(chunk)} bytes")
                buffer += chunk.decode("utf-8", errors="ignore")

                while buffer:
                    buffer = buffer.lstrip()
                    if not buffer:
                        break

                    try:
                        frame, index = decoder.raw_decode(buffer)
                        buffer = buffer[index:]

                        # Check if the frame actually contains data we care about
                        # Some Xovis frames might be heartbeat/metadata only
                        logger.debug(f"Dispatching frame from {peer}: {list(frame.keys())}")
                        asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

                    except json.JSONDecodeError:
                        # If we can't decode, it might be partial or truly malformed.
                        # To handle interleaved malformed data, we could try to skip one character and retry,
                        # but standard Xovis stream shouldn't have malformed data.
                        # However, for robustness, if it's not the start of a possible JSON object/array, we skip.
                        if not buffer.startswith(("{", "[")):
                            buffer = buffer[1:]
                            continue
                        break

        except Exception as e:
            logger.error(f"Stream error from {peer}: {e}")
        finally:
            writer.close()
            await writer.wait_closed()
            logger.info(f"Sensor disconnected: {peer}")

Methods:

`init()`

Initializes the XovisTCPServer.

Source code in src/xovis/datapush/tcp_server.py

def __init__(self):
    """
    Initializes the XovisTCPServer.
    """
    self.sinks: list[XovisSink] = []

`_handle_client(reader, writer)` `async`

Handles an individual sensor connection lifecycle.

Uses a sliding string buffer combined with raw_decode to safely extract concatenated JSON objects from the raw byte stream.

Parameters:

Name	Type	Description	Default
`reader`	`StreamReader`	The client stream reader.	required
`writer`	`StreamWriter`	The client stream writer.	required

Source code in src/xovis/datapush/tcp_server.py

async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    """
    Handles an individual sensor connection lifecycle.

    Uses a sliding string buffer combined with `raw_decode` to safely
    extract concatenated JSON objects from the raw byte stream.

    Args:
        reader (asyncio.StreamReader): The client stream reader.
        writer (asyncio.StreamWriter): The client stream writer.
    """
    peer = writer.get_extra_info("peername")
    logger.info(f"Sensor connected: {peer}")

    # Check if we should log to studio debug log
    # We look for any attached MetricSink with debug=True
    studio_debug = False
    if hasattr(self, "sinks"):
        for sink in self.sinks:
            if hasattr(sink, "debug") and sink.debug:
                studio_debug = True
                break

    if studio_debug:
        try:
            with open("xovis_studio_debug.log", "a", encoding="utf-8") as f:
                f.write(f"--- Sensor connected: {peer} ---\n")
        except Exception:
            pass

    buffer = ""
    decoder = json.JSONDecoder()

    try:
        while True:
            try:
                chunk = await asyncio.wait_for(reader.read(8192), timeout=60.0)
            except asyncio.TimeoutError:
                logger.debug(f"Read timeout for {peer}")
                continue

            if not chunk:
                logger.info(f"Connection closed by peer: {peer}")
                break

            logger.debug(f"Received chunk from {peer}: {len(chunk)} bytes")
            buffer += chunk.decode("utf-8", errors="ignore")

            while buffer:
                buffer = buffer.lstrip()
                if not buffer:
                    break

                try:
                    frame, index = decoder.raw_decode(buffer)
                    buffer = buffer[index:]

                    # Check if the frame actually contains data we care about
                    # Some Xovis frames might be heartbeat/metadata only
                    logger.debug(f"Dispatching frame from {peer}: {list(frame.keys())}")
                    asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

                except json.JSONDecodeError:
                    # If we can't decode, it might be partial or truly malformed.
                    # To handle interleaved malformed data, we could try to skip one character and retry,
                    # but standard Xovis stream shouldn't have malformed data.
                    # However, for robustness, if it's not the start of a possible JSON object/array, we skip.
                    if not buffer.startswith(("{", "[")):
                        buffer = buffer[1:]
                        continue
                    break

    except Exception as e:
        logger.error(f"Stream error from {peer}: {e}")
    finally:
        writer.close()
        await writer.wait_closed()
        logger.info(f"Sensor disconnected: {peer}")

`attach_sink(sink)`

Attaches a telemetry consumer to the server.

Parameters:

Name	Type	Description	Default
`sink`	`XovisSink`	An object implementing the XovisSink protocol.	required

Returns:

Name	Type	Description
`XovisTCPServer`	`XovisTCPServer`	The server instance for method chaining.

Source code in src/xovis/datapush/tcp_server.py

def attach_sink(self, sink: XovisSink) -> "XovisTCPServer":
    """
    Attaches a telemetry consumer to the server.

    Args:
        sink (XovisSink): An object implementing the XovisSink protocol.

    Returns:
        XovisTCPServer: The server instance for method chaining.
    """
    self.sinks.append(sink)
    return self

`start(host='0.0.0.0', port=9000)` `async`

Starts the TCP server and begins listening for sensor connections.

Parameters:

Name	Type	Description	Default
`host`	`str`	The network interface to bind to. Defaults to "0.0.0.0".	`'0.0.0.0'`
`port`	`int`	The TCP port to listen on. Defaults to 9000.	`9000`

Raises:

Type	Description
`OSError`	If the port is already in use or binding fails.

Source code in src/xovis/datapush/tcp_server.py

async def start(self, host: str = "0.0.0.0", port: int = 9000):  # nosec B104
    """
    Starts the TCP server and begins listening for sensor connections.

    Args:
        host (str, optional): The network interface to bind to. Defaults to "0.0.0.0".
        port (int, optional): The TCP port to listen on. Defaults to 9000.

    Raises:
        OSError: If the port is already in use or binding fails.
    """
    self._server = await asyncio.start_server(self._handle_client, host, port)
    logger.info(f"Xovis TCP Server listening on {host}:{port}")
    async with self._server:
        await self._server.serve_forever()

`stop()` `async`

Gracefully tears down the TCP server.

Source code in src/xovis/datapush/tcp_server.py

async def stop(self):
    """
    Gracefully tears down the TCP server.
    """
    if hasattr(self, "_server") and self._server:
        self._server.close()
        await self._server.wait_closed()
        logger.info("Xovis TCP Server stopped")

`xovis.datapush.http_server`

Xovis SDK - Data Plane HTTP Server

This module implements a high-performance ASGI Webhook Server for Xovis telemetry ingestion. It resides strictly within the Data Plane, utilizing aiohttp and orjson for zero-copy, lock-free processing of incoming HTTP pushes without Pydantic overhead.

Classes

`XovisHTTPServer`

High-performance ASGI Webhook Server for Xovis Telemetry.

Operates strictly within the Data Plane to ingest 12.5Hz telemetry via HTTP webhooks. Designed for maximum throughput, it bypasses UTF-8 decoding and Pydantic validation, dispatching parsed frames to attached sinks via asynchronous fire-and-forget tasks.

Source code in src/xovis/datapush/http_server.py

class XovisHTTPServer:
    """
    High-performance ASGI Webhook Server for Xovis Telemetry.

    Operates strictly within the Data Plane to ingest 12.5Hz telemetry via HTTP
    webhooks. Designed for maximum throughput, it bypasses UTF-8 decoding and
    Pydantic validation, dispatching parsed frames to attached sinks via
    asynchronous fire-and-forget tasks.
    """

    def __init__(self, expected_token: str = None, expected_user: str = None, expected_password: str = None):
        self.sinks: list[XovisSink] = []
        self.expected_token = expected_token
        self.expected_user = expected_user
        self.expected_password = expected_password
        self._app = web.Application()

        # Intercept all paths to prevent 404s when the sensor drops the URI path
        self._app.router.add_route("*", "/{tail:.*}", self._handle_post)

        self._runner = None
        self._site = None

    def attach_sink(self, sink: XovisSink) -> "XovisHTTPServer":
        """
        Attaches a telemetry consumer to the server.

        Args:
            sink (XovisSink): An object implementing the XovisSink protocol
                to receive ingested telemetry frames.

        Returns:
            XovisHTTPServer: The server instance for method chaining.
        """
        self.sinks.append(sink)
        return self

    async def start(self, host: str = "0.0.0.0", port: int = 9001):  # nosec B104
        """
        Initializes the aiohttp runner and binds the TCP socket.

        Enters a non-blocking sleep loop to keep the server alive until cancelled.

        Args:
            host (str, optional): The network interface to bind to. Defaults to "0.0.0.0".
            port (int, optional): The TCP port to listen on. Defaults to 9001.

        Raises:
            OSError: If the port is already in use or binding fails.
        """
        self._runner = web.AppRunner(self._app, access_log=None)
        await self._runner.setup()
        self._site = web.TCPSite(self._runner, host, port)
        await self._site.start()
        logger.info(f"Xovis HTTP Server listening on http://{host}:{port}/webhook")

        try:
            while True:
                await asyncio.sleep(3600)
        except asyncio.CancelledError:
            pass

    async def stop(self):
        """
        Gracefully tears down the HTTP server and cleans up active connections.
        """
        if self._site:
            await self._site.stop()
        if self._runner:
            await self._runner.cleanup()

    async def _handle_post(self, request: web.Request) -> web.Response:
        """
        Hot path ingestion for incoming HTTP requests.

        Directly reads bytes from the socket buffer and performs instantaneous
        C-level deserialization via orjson. Broadcasts frames to sinks
        asynchronously to unblock the HTTP response immediately.

        Args:
            request (web.Request): The incoming aiohttp request.

        Returns:
            web.Response: A 200 OK response on success, or appropriate error codes.
        """
        if self.expected_token:
            auth_header = request.headers.get("Authorization", "")
            if not auth_header.lower().startswith("bearer ") or auth_header[7:].strip() != self.expected_token:
                return web.Response(status=401, text="Unauthorized: Invalid Bearer Token")

        if self.expected_user and self.expected_password:
            auth_header = request.headers.get("Authorization", "")
            expected_b64 = base64.b64encode(f"{self.expected_user}:{self.expected_password}".encode()).decode("utf-8")
            if not auth_header.lower().startswith("basic ") or auth_header[6:].strip() != expected_b64:
                return web.Response(status=401, text="Unauthorized: Invalid Basic Auth")

        try:
            body_bytes = await request.read()

            if not body_bytes:
                return web.Response(status=200, text="OK")

            frame_data = DataPlaneIngestor.parse_frame(body_bytes)

            # Support firmware batching (Logics payloads are sometimes arrays)
            frames = frame_data if isinstance(frame_data, list) else [frame_data]

            for frame in frames:
                asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

            return web.Response(status=200, text="OK")

        except Exception as e:
            logger.error(f"HTTP stream error: {e}")
            return web.Response(status=500, text="Internal Server Error")

Methods:

`_handle_post(request)` `async`

Hot path ingestion for incoming HTTP requests.

Directly reads bytes from the socket buffer and performs instantaneous C-level deserialization via orjson. Broadcasts frames to sinks asynchronously to unblock the HTTP response immediately.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The incoming aiohttp request.	required

Returns:

Type	Description
`Response`	web.Response: A 200 OK response on success, or appropriate error codes.

Source code in src/xovis/datapush/http_server.py

async def _handle_post(self, request: web.Request) -> web.Response:
    """
    Hot path ingestion for incoming HTTP requests.

    Directly reads bytes from the socket buffer and performs instantaneous
    C-level deserialization via orjson. Broadcasts frames to sinks
    asynchronously to unblock the HTTP response immediately.

    Args:
        request (web.Request): The incoming aiohttp request.

    Returns:
        web.Response: A 200 OK response on success, or appropriate error codes.
    """
    if self.expected_token:
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.lower().startswith("bearer ") or auth_header[7:].strip() != self.expected_token:
            return web.Response(status=401, text="Unauthorized: Invalid Bearer Token")

    if self.expected_user and self.expected_password:
        auth_header = request.headers.get("Authorization", "")
        expected_b64 = base64.b64encode(f"{self.expected_user}:{self.expected_password}".encode()).decode("utf-8")
        if not auth_header.lower().startswith("basic ") or auth_header[6:].strip() != expected_b64:
            return web.Response(status=401, text="Unauthorized: Invalid Basic Auth")

    try:
        body_bytes = await request.read()

        if not body_bytes:
            return web.Response(status=200, text="OK")

        frame_data = DataPlaneIngestor.parse_frame(body_bytes)

        # Support firmware batching (Logics payloads are sometimes arrays)
        frames = frame_data if isinstance(frame_data, list) else [frame_data]

        for frame in frames:
            asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

        return web.Response(status=200, text="OK")

    except Exception as e:
        logger.error(f"HTTP stream error: {e}")
        return web.Response(status=500, text="Internal Server Error")

`attach_sink(sink)`

Attaches a telemetry consumer to the server.

Parameters:

Name	Type	Description	Default
`sink`	`XovisSink`	An object implementing the XovisSink protocol to receive ingested telemetry frames.	required

Returns:

Name	Type	Description
`XovisHTTPServer`	`XovisHTTPServer`	The server instance for method chaining.

Source code in src/xovis/datapush/http_server.py

def attach_sink(self, sink: XovisSink) -> "XovisHTTPServer":
    """
    Attaches a telemetry consumer to the server.

    Args:
        sink (XovisSink): An object implementing the XovisSink protocol
            to receive ingested telemetry frames.

    Returns:
        XovisHTTPServer: The server instance for method chaining.
    """
    self.sinks.append(sink)
    return self

`start(host='0.0.0.0', port=9001)` `async`

Initializes the aiohttp runner and binds the TCP socket.

Enters a non-blocking sleep loop to keep the server alive until cancelled.

Parameters:

Name	Type	Description	Default
`host`	`str`	The network interface to bind to. Defaults to "0.0.0.0".	`'0.0.0.0'`
`port`	`int`	The TCP port to listen on. Defaults to 9001.	`9001`

Raises:

Type	Description
`OSError`	If the port is already in use or binding fails.

Source code in src/xovis/datapush/http_server.py

async def start(self, host: str = "0.0.0.0", port: int = 9001):  # nosec B104
    """
    Initializes the aiohttp runner and binds the TCP socket.

    Enters a non-blocking sleep loop to keep the server alive until cancelled.

    Args:
        host (str, optional): The network interface to bind to. Defaults to "0.0.0.0".
        port (int, optional): The TCP port to listen on. Defaults to 9001.

    Raises:
        OSError: If the port is already in use or binding fails.
    """
    self._runner = web.AppRunner(self._app, access_log=None)
    await self._runner.setup()
    self._site = web.TCPSite(self._runner, host, port)
    await self._site.start()
    logger.info(f"Xovis HTTP Server listening on http://{host}:{port}/webhook")

    try:
        while True:
            await asyncio.sleep(3600)
    except asyncio.CancelledError:
        pass

`stop()` `async`

Gracefully tears down the HTTP server and cleans up active connections.

Source code in src/xovis/datapush/http_server.py

async def stop(self):
    """
    Gracefully tears down the HTTP server and cleans up active connections.
    """
    if self._site:
        await self._site.stop()
    if self._runner:
        await self._runner.cleanup()

`xovis.datapush.udp_server`

Xovis SDK - Data Plane UDP Server

This module implements a high-performance UDP ingestion engine for Xovis telemetry. It utilizes asyncio.DatagramProtocol for low-latency ingestion of discrete JSON packets, residing strictly within the Data Plane.

Classes

`XovisUDPProtocol`

Bases: DatagramProtocol

Protocol implementation for high-speed UDP ingestion.

Handles incoming datagrams from Xovis sensors. Each datagram is expected to be a complete JSON frame. This class operates in the hot path and dispatches telemetry to sinks with minimal overhead.

Source code in src/xovis/datapush/udp_server.py

class XovisUDPProtocol(asyncio.DatagramProtocol):
    """
    Protocol implementation for high-speed UDP ingestion.

    Handles incoming datagrams from Xovis sensors. Each datagram is expected
    to be a complete JSON frame. This class operates in the hot path and
    dispatches telemetry to sinks with minimal overhead.
    """

    def __init__(self, sinks: list[XovisSink]):
        """
        Initializes the UDP protocol.

        Args:
            sinks (List[XovisSink]): A list of attached telemetry sinks.
        """
        self.sinks = sinks
        self.transport: Optional[asyncio.DatagramTransport] = None

    def connection_made(self, transport: asyncio.DatagramTransport):
        """
        Called when the UDP socket is ready.

        Args:
            transport (asyncio.DatagramTransport): The transport for the socket.
        """
        self.transport = transport
        peer = transport.get_extra_info("peername")
        logger.debug(f"UDP socket ready: {peer}")

    def datagram_received(self, data: bytes, addr: tuple):
        """
        Hot path ingestion for UDP datagrams.

        Parses the incoming byte packet as a JSON object and routes it to
        the attached sinks.

        Args:
            data (bytes): The raw packet data.
            addr (tuple): The source address of the packet.
        """
        # Check if we should log to studio debug log
        studio_debug = False
        if hasattr(self, "sinks"):
            for sink in self.sinks:
                if hasattr(sink, "debug") and sink.debug:
                    studio_debug = True
                    break

        if studio_debug:
            try:
                with open("xovis_studio_debug.log", "a", encoding="utf-8") as f:
                    f.write(f"--- UDP datagram received from {addr}: {len(data)} bytes ---\n")
            except Exception:
                pass

        try:
            logger.debug(f"Received UDP datagram from {addr}: {len(data)} bytes")
            frame = DataPlaneIngestor.parse_frame(data)

            asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

        except Exception as e:
            logger.error(f"UDP stream error from {addr}: {e}")

Methods:

`init(sinks)`

Initializes the UDP protocol.

Parameters:

Name	Type	Description	Default
`sinks`	`List[XovisSink]`	A list of attached telemetry sinks.	required

Source code in src/xovis/datapush/udp_server.py

def __init__(self, sinks: list[XovisSink]):
    """
    Initializes the UDP protocol.

    Args:
        sinks (List[XovisSink]): A list of attached telemetry sinks.
    """
    self.sinks = sinks
    self.transport: Optional[asyncio.DatagramTransport] = None

`connection_made(transport)`

Called when the UDP socket is ready.

Parameters:

Name	Type	Description	Default
`transport`	`DatagramTransport`	The transport for the socket.	required

Source code in src/xovis/datapush/udp_server.py

def connection_made(self, transport: asyncio.DatagramTransport):
    """
    Called when the UDP socket is ready.

    Args:
        transport (asyncio.DatagramTransport): The transport for the socket.
    """
    self.transport = transport
    peer = transport.get_extra_info("peername")
    logger.debug(f"UDP socket ready: {peer}")

`datagram_received(data, addr)`

Hot path ingestion for UDP datagrams.

Parses the incoming byte packet as a JSON object and routes it to the attached sinks.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	The raw packet data.	required
`addr`	`tuple`	The source address of the packet.	required

Source code in src/xovis/datapush/udp_server.py

def datagram_received(self, data: bytes, addr: tuple):
    """
    Hot path ingestion for UDP datagrams.

    Parses the incoming byte packet as a JSON object and routes it to
    the attached sinks.

    Args:
        data (bytes): The raw packet data.
        addr (tuple): The source address of the packet.
    """
    # Check if we should log to studio debug log
    studio_debug = False
    if hasattr(self, "sinks"):
        for sink in self.sinks:
            if hasattr(sink, "debug") and sink.debug:
                studio_debug = True
                break

    if studio_debug:
        try:
            with open("xovis_studio_debug.log", "a", encoding="utf-8") as f:
                f.write(f"--- UDP datagram received from {addr}: {len(data)} bytes ---\n")
        except Exception:
            pass

    try:
        logger.debug(f"Received UDP datagram from {addr}: {len(data)} bytes")
        frame = DataPlaneIngestor.parse_frame(data)

        asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

    except Exception as e:
        logger.error(f"UDP stream error from {addr}: {e}")

`XovisUDPServer`

High-performance UDP Ingestion Engine for Xovis Telemetry.

Operates within the Data Plane to provide low-latency telemetry ingestion. Coordinates the lifecycle of the UDP socket and the underlying protocol.

Source code in src/xovis/datapush/udp_server.py

class XovisUDPServer:
    """
    High-performance UDP Ingestion Engine for Xovis Telemetry.

    Operates within the Data Plane to provide low-latency telemetry ingestion.
    Coordinates the lifecycle of the UDP socket and the underlying protocol.
    """

    def __init__(self):
        """
        Initializes the XovisUDPServer.
        """
        self.sinks: list[XovisSink] = []
        self.transport: Optional[asyncio.DatagramTransport] = None
        self.protocol: Optional[XovisUDPProtocol] = None

    def attach_sink(self, sink: XovisSink) -> "XovisUDPServer":
        """
        Attaches a telemetry consumer to the server.

        Args:
            sink (XovisSink): An object implementing the XovisSink protocol.

        Returns:
            XovisUDPServer: The server instance for method chaining.
        """
        self.sinks.append(sink)
        return self

    async def stop(self):
        """
        Gracefully tears down the UDP server.
        """
        if self.transport:
            self.transport.close()
            self.transport = None
            logger.info("Xovis UDP Server stopped")

    async def start(self, host: str = "0.0.0.0", port: int = 9002):  # nosec B104
        """
        Starts the UDP server and begins listening for datagrams.

        Args:
            host (str, optional): The network interface to bind to. Defaults to "0.0.0.0".
            port (int, optional): The UDP port to listen on. Defaults to 9002.

        Raises:
            OSError: If binding fails.
        """
        loop = asyncio.get_running_loop()
        self.transport, self.protocol = await loop.create_datagram_endpoint(lambda: XovisUDPProtocol(self.sinks), local_addr=(host, port))
        logger.info(f"Xovis UDP Server listening on {host}:{port}")

        try:
            while True:
                await asyncio.sleep(3600)
        except asyncio.CancelledError:
            if self.transport:
                self.transport.close()
            logger.info("Xovis UDP Server stopped")

Methods:

`init()`

Initializes the XovisUDPServer.

Source code in src/xovis/datapush/udp_server.py

def __init__(self):
    """
    Initializes the XovisUDPServer.
    """
    self.sinks: list[XovisSink] = []
    self.transport: Optional[asyncio.DatagramTransport] = None
    self.protocol: Optional[XovisUDPProtocol] = None

`attach_sink(sink)`

Attaches a telemetry consumer to the server.

Parameters:

Name	Type	Description	Default
`sink`	`XovisSink`	An object implementing the XovisSink protocol.	required

Returns:

Name	Type	Description
`XovisUDPServer`	`XovisUDPServer`	The server instance for method chaining.

Source code in src/xovis/datapush/udp_server.py

def attach_sink(self, sink: XovisSink) -> "XovisUDPServer":
    """
    Attaches a telemetry consumer to the server.

    Args:
        sink (XovisSink): An object implementing the XovisSink protocol.

    Returns:
        XovisUDPServer: The server instance for method chaining.
    """
    self.sinks.append(sink)
    return self

`start(host='0.0.0.0', port=9002)` `async`

Starts the UDP server and begins listening for datagrams.

Parameters:

Name	Type	Description	Default
`host`	`str`	The network interface to bind to. Defaults to "0.0.0.0".	`'0.0.0.0'`
`port`	`int`	The UDP port to listen on. Defaults to 9002.	`9002`

Raises:

Type	Description
`OSError`	If binding fails.

Source code in src/xovis/datapush/udp_server.py

async def start(self, host: str = "0.0.0.0", port: int = 9002):  # nosec B104
    """
    Starts the UDP server and begins listening for datagrams.

    Args:
        host (str, optional): The network interface to bind to. Defaults to "0.0.0.0".
        port (int, optional): The UDP port to listen on. Defaults to 9002.

    Raises:
        OSError: If binding fails.
    """
    loop = asyncio.get_running_loop()
    self.transport, self.protocol = await loop.create_datagram_endpoint(lambda: XovisUDPProtocol(self.sinks), local_addr=(host, port))
    logger.info(f"Xovis UDP Server listening on {host}:{port}")

    try:
        while True:
            await asyncio.sleep(3600)
    except asyncio.CancelledError:
        if self.transport:
            self.transport.close()
        logger.info("Xovis UDP Server stopped")

`stop()` `async`

Gracefully tears down the UDP server.

Source code in src/xovis/datapush/udp_server.py

async def stop(self):
    """
    Gracefully tears down the UDP server.
    """
    if self.transport:
        self.transport.close()
        self.transport = None
        logger.info("Xovis UDP Server stopped")

`xovis.datapush.tcp_client`

Xovis SDK - Data Plane TCP Client

This module implements a high-speed active TCP ingestion client. It connects to Xovis sensors configured in SERVER mode (ports 49156/49159) and extracts concatenated JSON frames using a zero-copy sliding buffer, adhering strictly to the Data Plane's non-blocking architecture.

Classes

`XovisTCPClient`

Active TCP Ingestion Client for Xovis Telemetry.

Initiates and persists a connection to a Xovis sensor running in SERVER mode. Utilizes a sliding string buffer to slice out continuous JSON payloads without network delimiters, routing them to attached sinks.

Source code in src/xovis/datapush/tcp_client.py

class XovisTCPClient:
    """
    Active TCP Ingestion Client for Xovis Telemetry.

    Initiates and persists a connection to a Xovis sensor running in SERVER mode.
    Utilizes a sliding string buffer to slice out continuous JSON payloads without
    network delimiters, routing them to attached sinks.
    """

    def __init__(self, host: str, port: int = 49156, reconnect_interval: float = 5.0):
        """
        Initializes the XovisTCPClient.

        Args:
            host (str): The IP address or hostname of the Xovis sensor.
            port (int, optional): The TCP port the sensor is listening on.
                Defaults to 49156 (Singlesensor) or 49159 (Multisensor).
            reconnect_interval (float, optional): Seconds to wait before attempting
                to reconnect after a drop. Defaults to 5.0.
        """
        self.host = host
        self.port = port
        self.reconnect_interval = reconnect_interval
        self.sinks: list[XovisSink] = []
        self._running = False

    def attach_sink(self, sink: XovisSink) -> "XovisTCPClient":
        """
        Attaches a telemetry consumer to the client.

        Args:
            sink (XovisSink): An object implementing the XovisSink protocol.

        Returns:
            XovisTCPClient: The client instance for method chaining.
        """
        self.sinks.append(sink)
        return self

    async def start(self) -> None:
        """
        Starts the client connection loop.

        Continuously attempts to connect to the sensor and read the stream.
        If the connection drops or the sensor reboots, it will automatically
        back off and reconnect.
        """
        self._running = True
        logger.info(f"Xovis TCP Client starting connection loop to {self.host}:{self.port}")

        while self._running:
            try:
                await self._connect_and_read()
            except (ConnectionRefusedError, TimeoutError, OSError) as e:
                logger.warning(f"Connection to {self.host}:{self.port} failed: {e}. Retrying in {self.reconnect_interval}s...")
            except Exception as e:
                logger.error(f"Unexpected error in TCP client for {self.host}:{self.port}: {e}")

            if self._running:
                await asyncio.sleep(self.reconnect_interval)

    async def stop(self) -> None:
        """Terminates the active connection loop."""
        self._running = False
        logger.info(f"Xovis TCP Client stopping connection to {self.host}:{self.port}")

    async def _connect_and_read(self) -> None:
        """
        Maintains the active socket and parses the byte stream.

        Uses the `raw_decode` sliding buffer to slice complete JSON objects
        out of the fragmented MTU network chunks.
        """
        reader, writer = await asyncio.open_connection(self.host, self.port)
        logger.info(f"Successfully connected to Xovis sensor at {self.host}:{self.port}")

        buffer = ""
        decoder = json.JSONDecoder()

        try:
            while self._running:
                chunk = await reader.read(8192)
                if not chunk:
                    logger.warning(f"TCP stream closed by sensor {self.host}:{self.port}")
                    break

                buffer += chunk.decode("utf-8", errors="ignore")

                while buffer:
                    buffer = buffer.lstrip()
                    if not buffer:
                        break

                    try:
                        frame, index = decoder.raw_decode(buffer)
                        buffer = buffer[index:]

                        asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

                    except json.JSONDecodeError:
                        if not buffer.startswith(("{", "[")):
                            buffer = buffer[1:]
                            continue
                        break

        finally:
            writer.close()
            await writer.wait_closed()
            logger.info(f"Disconnected from {self.host}:{self.port}")

Methods:

`init(host, port=49156, reconnect_interval=5.0)`

Initializes the XovisTCPClient.

Parameters:

Name	Type	Description	Default
`host`	`str`	The IP address or hostname of the Xovis sensor.	required
`port`	`int`	The TCP port the sensor is listening on. Defaults to 49156 (Singlesensor) or 49159 (Multisensor).	`49156`
`reconnect_interval`	`float`	Seconds to wait before attempting to reconnect after a drop. Defaults to 5.0.	`5.0`

Source code in src/xovis/datapush/tcp_client.py

def __init__(self, host: str, port: int = 49156, reconnect_interval: float = 5.0):
    """
    Initializes the XovisTCPClient.

    Args:
        host (str): The IP address or hostname of the Xovis sensor.
        port (int, optional): The TCP port the sensor is listening on.
            Defaults to 49156 (Singlesensor) or 49159 (Multisensor).
        reconnect_interval (float, optional): Seconds to wait before attempting
            to reconnect after a drop. Defaults to 5.0.
    """
    self.host = host
    self.port = port
    self.reconnect_interval = reconnect_interval
    self.sinks: list[XovisSink] = []
    self._running = False

`_connect_and_read()` `async`

Maintains the active socket and parses the byte stream.

Uses the raw_decode sliding buffer to slice complete JSON objects out of the fragmented MTU network chunks.

Source code in src/xovis/datapush/tcp_client.py

async def _connect_and_read(self) -> None:
    """
    Maintains the active socket and parses the byte stream.

    Uses the `raw_decode` sliding buffer to slice complete JSON objects
    out of the fragmented MTU network chunks.
    """
    reader, writer = await asyncio.open_connection(self.host, self.port)
    logger.info(f"Successfully connected to Xovis sensor at {self.host}:{self.port}")

    buffer = ""
    decoder = json.JSONDecoder()

    try:
        while self._running:
            chunk = await reader.read(8192)
            if not chunk:
                logger.warning(f"TCP stream closed by sensor {self.host}:{self.port}")
                break

            buffer += chunk.decode("utf-8", errors="ignore")

            while buffer:
                buffer = buffer.lstrip()
                if not buffer:
                    break

                try:
                    frame, index = decoder.raw_decode(buffer)
                    buffer = buffer[index:]

                    asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

                except json.JSONDecodeError:
                    if not buffer.startswith(("{", "[")):
                        buffer = buffer[1:]
                        continue
                    break

    finally:
        writer.close()
        await writer.wait_closed()
        logger.info(f"Disconnected from {self.host}:{self.port}")

`attach_sink(sink)`

Attaches a telemetry consumer to the client.

Parameters:

Name	Type	Description	Default
`sink`	`XovisSink`	An object implementing the XovisSink protocol.	required

Returns:

Name	Type	Description
`XovisTCPClient`	`XovisTCPClient`	The client instance for method chaining.

Source code in src/xovis/datapush/tcp_client.py

def attach_sink(self, sink: XovisSink) -> "XovisTCPClient":
    """
    Attaches a telemetry consumer to the client.

    Args:
        sink (XovisSink): An object implementing the XovisSink protocol.

    Returns:
        XovisTCPClient: The client instance for method chaining.
    """
    self.sinks.append(sink)
    return self

`start()` `async`

Starts the client connection loop.

Continuously attempts to connect to the sensor and read the stream. If the connection drops or the sensor reboots, it will automatically back off and reconnect.

Source code in src/xovis/datapush/tcp_client.py

async def start(self) -> None:
    """
    Starts the client connection loop.

    Continuously attempts to connect to the sensor and read the stream.
    If the connection drops or the sensor reboots, it will automatically
    back off and reconnect.
    """
    self._running = True
    logger.info(f"Xovis TCP Client starting connection loop to {self.host}:{self.port}")

    while self._running:
        try:
            await self._connect_and_read()
        except (ConnectionRefusedError, TimeoutError, OSError) as e:
            logger.warning(f"Connection to {self.host}:{self.port} failed: {e}. Retrying in {self.reconnect_interval}s...")
        except Exception as e:
            logger.error(f"Unexpected error in TCP client for {self.host}:{self.port}: {e}")

        if self._running:
            await asyncio.sleep(self.reconnect_interval)

`stop()` `async`

Terminates the active connection loop.

Source code in src/xovis/datapush/tcp_client.py

async def stop(self) -> None:
    """Terminates the active connection loop."""
    self._running = False
    logger.info(f"Xovis TCP Client stopping connection to {self.host}:{self.port}")

`xovis.datapush.mqtt_client`

Xovis SDK - Data Plane MQTT Client

This module implements a high-performance MQTT ingestion client for Xovis telemetry. It utilizes aiomqtt to subscribe to broker topics, process JSON payloads, and dispatch them to attached sinks while natively handling network backpressure.

Classes

`XovisMQTTClient`

Active MQTT Ingestion Client for Xovis Telemetry.

Connects to an MQTT broker, subscribes to the configured telemetry topic, and routes incoming JSON payloads to the attached sinks. Supports TLS and username/password authentication.

Source code in src/xovis/datapush/mqtt_client.py

class XovisMQTTClient:
    """
    Active MQTT Ingestion Client for Xovis Telemetry.

    Connects to an MQTT broker, subscribes to the configured telemetry topic,
    and routes incoming JSON payloads to the attached sinks. Supports TLS
    and username/password authentication.
    """

    def __init__(
        self,
        host: str,
        topic: str,
        port: int = 1883,
        username: Optional[str] = None,
        password: Optional[str] = None,
        use_ssl: bool = False,
    ):
        """
        Initializes the XovisMQTTClient.

        Args:
            host (str): The IP address or hostname of the MQTT broker.
            topic (str): The MQTT topic to subscribe to.
            port (int, optional): The broker port. Defaults to 1883 (or 8883 if SSL is True).
            username (Optional[str], optional): Authentication username.
            password (Optional[str], optional): Authentication password.
            use_ssl (bool, optional): Whether to use TLS encryption. Defaults to False.
        """
        self.host = host
        self.port = 8883 if use_ssl and port == 1883 else port
        self.topic = topic
        self.username = username
        self.password = password
        self.use_ssl = use_ssl
        self.sinks: list[XovisSink] = []
        self._running = False

    def attach_sink(self, sink: XovisSink) -> "XovisMQTTClient":
        """
        Attaches a telemetry consumer to the client.

        Args:
            sink (XovisSink): An object implementing the XovisSink protocol.

        Returns:
            XovisMQTTClient: The client instance for method chaining.
        """
        self.sinks.append(sink)
        return self

    async def start(self) -> None:
        """
        Starts the MQTT client loop.

        Connects to the broker, subscribes to the topic, and listens for messages.
        `aiomqtt` automatically handles reconnections under the hood.
        """
        self._running = True

        tls_context = ssl.create_default_context() if self.use_ssl else None

        logger.info(f"Xovis MQTT Client connecting to {self.host}:{self.port} on topic '{self.topic}'")

        try:
            async with aiomqtt.Client(
                hostname=self.host,
                port=self.port,
                username=self.username,
                password=self.password,
                tls_context=tls_context,
            ) as client:
                await client.subscribe(self.topic)
                logger.info(f"Successfully subscribed to {self.topic}")

                async for message in client.messages:
                    if not self._running:
                        break

                    try:
                        frame = DataPlaneIngestor.parse_frame(message.payload)
                        # Filter out non-JSON frames (binary recordings) for MQTT
                        # unless they are explicitly identified as such.
                        # MQTT usually expects structured JSON for telemetry.
                        if "recording_data" in frame and len(frame) == 1:
                            logger.debug("Ignoring non-JSON MQTT payload")
                            continue

                        asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

                    except Exception as e:
                        logger.error(f"Error processing MQTT message: {e}")

        except aiomqtt.MqttError as e:
            logger.error(f"MQTT connection error: {e}")
        except asyncio.CancelledError:
            logger.info("Xovis MQTT Client stopped")
        finally:
            self._running = False

    async def stop(self) -> None:
        """Terminates the active connection loop."""
        self._running = False

Methods:

`init(host, topic, port=1883, username=None, password=None, use_ssl=False)`

Initializes the XovisMQTTClient.

Parameters:

Name	Type	Description	Default
`host`	`str`	The IP address or hostname of the MQTT broker.	required
`topic`	`str`	The MQTT topic to subscribe to.	required
`port`	`int`	The broker port. Defaults to 1883 (or 8883 if SSL is True).	`1883`
`username`	`Optional[str]`	Authentication username.	`None`
`password`	`Optional[str]`	Authentication password.	`None`
`use_ssl`	`bool`	Whether to use TLS encryption. Defaults to False.	`False`

Source code in src/xovis/datapush/mqtt_client.py

def __init__(
    self,
    host: str,
    topic: str,
    port: int = 1883,
    username: Optional[str] = None,
    password: Optional[str] = None,
    use_ssl: bool = False,
):
    """
    Initializes the XovisMQTTClient.

    Args:
        host (str): The IP address or hostname of the MQTT broker.
        topic (str): The MQTT topic to subscribe to.
        port (int, optional): The broker port. Defaults to 1883 (or 8883 if SSL is True).
        username (Optional[str], optional): Authentication username.
        password (Optional[str], optional): Authentication password.
        use_ssl (bool, optional): Whether to use TLS encryption. Defaults to False.
    """
    self.host = host
    self.port = 8883 if use_ssl and port == 1883 else port
    self.topic = topic
    self.username = username
    self.password = password
    self.use_ssl = use_ssl
    self.sinks: list[XovisSink] = []
    self._running = False

`attach_sink(sink)`

Attaches a telemetry consumer to the client.

Parameters:

Name	Type	Description	Default
`sink`	`XovisSink`	An object implementing the XovisSink protocol.	required

Returns:

Name	Type	Description
`XovisMQTTClient`	`XovisMQTTClient`	The client instance for method chaining.

Source code in src/xovis/datapush/mqtt_client.py

def attach_sink(self, sink: XovisSink) -> "XovisMQTTClient":
    """
    Attaches a telemetry consumer to the client.

    Args:
        sink (XovisSink): An object implementing the XovisSink protocol.

    Returns:
        XovisMQTTClient: The client instance for method chaining.
    """
    self.sinks.append(sink)
    return self

`start()` `async`

Starts the MQTT client loop.

Connects to the broker, subscribes to the topic, and listens for messages. aiomqtt automatically handles reconnections under the hood.

Source code in src/xovis/datapush/mqtt_client.py

async def start(self) -> None:
    """
    Starts the MQTT client loop.

    Connects to the broker, subscribes to the topic, and listens for messages.
    `aiomqtt` automatically handles reconnections under the hood.
    """
    self._running = True

    tls_context = ssl.create_default_context() if self.use_ssl else None

    logger.info(f"Xovis MQTT Client connecting to {self.host}:{self.port} on topic '{self.topic}'")

    try:
        async with aiomqtt.Client(
            hostname=self.host,
            port=self.port,
            username=self.username,
            password=self.password,
            tls_context=tls_context,
        ) as client:
            await client.subscribe(self.topic)
            logger.info(f"Successfully subscribed to {self.topic}")

            async for message in client.messages:
                if not self._running:
                    break

                try:
                    frame = DataPlaneIngestor.parse_frame(message.payload)
                    # Filter out non-JSON frames (binary recordings) for MQTT
                    # unless they are explicitly identified as such.
                    # MQTT usually expects structured JSON for telemetry.
                    if "recording_data" in frame and len(frame) == 1:
                        logger.debug("Ignoring non-JSON MQTT payload")
                        continue

                    asyncio.create_task(DataPlaneIngestor.route_to_sinks(frame, self.sinks))

                except Exception as e:
                    logger.error(f"Error processing MQTT message: {e}")

    except aiomqtt.MqttError as e:
        logger.error(f"MQTT connection error: {e}")
    except asyncio.CancelledError:
        logger.info("Xovis MQTT Client stopped")
    finally:
        self._running = False

`stop()` `async`

Terminates the active connection loop.

Source code in src/xovis/datapush/mqtt_client.py

async def stop(self) -> None:
    """Terminates the active connection loop."""
    self._running = False

`xovis.datapush.sinks`

Xovis SDK - Data Plane Sink Interfaces

This module defines the architectural contract for all telemetry consumers within the Data Plane. It provides the XovisSink protocol, ensuring a unified interface for high-frequency frame and event processing.

Classes

`XovisSink`

Bases: Protocol

Base interface for developer integrations attaching to the stream.

Defines the standard protocol for processing incoming telemetry data. Implementations of this protocol are attached to XovisTCPServer, XovisUDPServer, or XovisHTTPServer to receive real-time updates.

Source code in src/xovis/datapush/sinks.py

class XovisSink(Protocol):
    """
    Base interface for developer integrations attaching to the stream.

    Defines the standard protocol for processing incoming telemetry data.
    Implementations of this protocol are attached to `XovisTCPServer`,
    `XovisUDPServer`, or `XovisHTTPServer` to receive real-time updates.
    """

    async def on_frame(self, frame: dict) -> None:
        """
        Triggered for every tracked frame received from the sensor.

        In standard configurations, this occurs approximately 12.5 times per
        second and contains all tracked objects and coordinates.

        Args:
            frame (dict): The parsed telemetry frame payload.
        """
        ...

    async def on_events(self, events: list) -> None:
        """
        Triggered when discrete events are detected by the sensor.

        Captures high-level logic events such as zone entries, exits, or
        line crossings.

        Args:
            events (list): A list of discrete event dictionaries.
        """
        ...

Methods:

`on_events(events)` `async`

Triggered when discrete events are detected by the sensor.

Captures high-level logic events such as zone entries, exits, or line crossings.

Parameters:

Name	Type	Description	Default
`events`	`list`	A list of discrete event dictionaries.	required

Source code in src/xovis/datapush/sinks.py

async def on_events(self, events: list) -> None:
    """
    Triggered when discrete events are detected by the sensor.

    Captures high-level logic events such as zone entries, exits, or
    line crossings.

    Args:
        events (list): A list of discrete event dictionaries.
    """
    ...

`on_frame(frame)` `async`

Triggered for every tracked frame received from the sensor.

In standard configurations, this occurs approximately 12.5 times per second and contains all tracked objects and coordinates.

Parameters:

Name	Type	Description	Default
`frame`	`dict`	The parsed telemetry frame payload.	required

Source code in src/xovis/datapush/sinks.py

async def on_frame(self, frame: dict) -> None:
    """
    Triggered for every tracked frame received from the sensor.

    In standard configurations, this occurs approximately 12.5 times per
    second and contains all tracked objects and coordinates.

    Args:
        frame (dict): The parsed telemetry frame payload.
    """
    ...

Telemetry Data Push Reference

xovis.datapush.tcp_server

Classes

XovisTCPServer

Methods:

__init__()

_handle_client(reader, writer) async

attach_sink(sink)

start(host='0.0.0.0', port=9000) async

stop() async

xovis.datapush.http_server

Classes

XovisHTTPServer

Methods:

_handle_post(request) async

attach_sink(sink)

start(host='0.0.0.0', port=9001) async

stop() async

xovis.datapush.udp_server

Classes

XovisUDPProtocol

Methods:

__init__(sinks)

connection_made(transport)

datagram_received(data, addr)

XovisUDPServer

Methods:

__init__()

attach_sink(sink)

start(host='0.0.0.0', port=9002) async

stop() async

xovis.datapush.tcp_client

Classes

XovisTCPClient

Methods:

__init__(host, port=49156, reconnect_interval=5.0)

_connect_and_read() async

attach_sink(sink)

start() async

stop() async

xovis.datapush.mqtt_client

Classes

XovisMQTTClient

Methods:

__init__(host, topic, port=1883, username=None, password=None, use_ssl=False)

attach_sink(sink)

start() async

stop() async

xovis.datapush.sinks

Classes

XovisSink

Methods:

on_events(events) async

on_frame(frame) async

`xovis.datapush.tcp_server`

`XovisTCPServer`

`init()`

`_handle_client(reader, writer)` `async`

`attach_sink(sink)`

`start(host='0.0.0.0', port=9000)` `async`

`stop()` `async`

`xovis.datapush.http_server`

`XovisHTTPServer`

`_handle_post(request)` `async`

`attach_sink(sink)`

`start(host='0.0.0.0', port=9001)` `async`

`stop()` `async`

`xovis.datapush.udp_server`

`XovisUDPProtocol`

`init(sinks)`

`connection_made(transport)`

`datagram_received(data, addr)`

`XovisUDPServer`

`init()`

`attach_sink(sink)`

`start(host='0.0.0.0', port=9002)` `async`

`stop()` `async`

`xovis.datapush.tcp_client`

`XovisTCPClient`

`init(host, port=49156, reconnect_interval=5.0)`

`_connect_and_read()` `async`

`attach_sink(sink)`

`start()` `async`

`stop()` `async`

`xovis.datapush.mqtt_client`

`XovisMQTTClient`

`init(host, topic, port=1883, username=None, password=None, use_ssl=False)`

`attach_sink(sink)`

`start()` `async`

`stop()` `async`

`xovis.datapush.sinks`

`XovisSink`

`on_events(events)` `async`

`on_frame(frame)` `async`