Listener Monitor

Create a small utility that monitors live table data and turns network updates into structured events for downstream consumers.

Capabilities

Track connection status

• When monitoring starts disconnected and a connection state of connected=True is delivered, a connection event is recorded with the provided info payload. @test
• When the connection drops after being connected, a connection event with connected=False is recorded. @test

Observe entry changes

• After starting to watch the SmartDashboard table, creating the key speed with value 3.5 emits a single entry event containing the full path /SmartDashboard/speed, the numeric value, and is_new=True. @test
• Updating an existing watched key records an entry event with is_new=False while reflecting the latest value. @test

Discover subtables

• When a new subtable Vision is created under a watched root, a subtable event records parent="SmartDashboard" and name="Vision" once; subsequent entry additions inside that subtable do not duplicate the discovery event. @test

Cleanup and unsubscribe

• After stop() is called, further entry or connection changes no longer append to the event log. @test

Implementation

@generates

API

from typing import Callable, List, Dict, Any

class ListenerMonitor:
    def __init__(self, server: str = "localhost"):
        ...

    def start(
        self,
        watch_tables: List[str],
        on_entry: Callable[[str, Any, bool], None],
        on_subtable: Callable[[str, str], None],
        on_connection: Callable[[bool, Dict[str, Any]], None],
    ) -> None:
        """
        Connects to the table server and registers listeners for the given root tables.
        on_entry receives (full_path, value, is_new).
        on_subtable receives (parent_table, subtable_name).
        on_connection receives (connected, info).
        Every listener invocation is also appended to the internal event log.
        """

    def stop(self) -> None:
        """Unregisters all listeners and disconnects/cleans up resources."""

    def get_events(self) -> List[Dict[str, Any]]:
        """
        Returns a chronological list of captured events describing entries, subtables, and connections.
        Each event dict includes a 'type' field (entry|subtable|connection) and type-specific payload:
        - entry events include keys 'path', 'value', and 'is_new'
        - subtable events include keys 'parent' and 'name'
        - connection events include keys 'connected' and 'info'
        """

Dependencies { .dependencies }

pynetworktables { .dependency }

Provides realtime table, entry, subtable, and connection listener APIs.