Network Update Synchronizer

Coordinate configurable NetworkTables update timing with explicit flush and listener draining controls for deterministic telemetry publishing. Values are staged locally and propagated on the configured cadence unless explicitly flushed.

Capabilities

Configurable cadence

• Accepts a default update interval of 0.1 seconds and automatically pushes staged values on that cadence; reconfiguring the interval updates the pacing @test
• Rejects non-positive intervals with a clear error @test

Manual flush of staged entries

• After staging values, a manual flush pushes pending changes immediately instead of waiting for the next interval @test

Listener drain for deterministic handoff

• Provides a blocking call that waits for entry and connection listener queues to empty before continuing @test
• Returns False when the queues fail to empty within the timeout while leaving staged data intact @test

Graceful shutdown sequencing

• Shutdown triggers a final flush, drains listener queues with the provided timeout, and reports whether the drain completed @test

Implementation

@generates

API

class NetworkUpdateSynchronizer:
    def __init__(self, interval_seconds: float = 0.1): ...
    def configure_interval(self, interval_seconds: float) -> None: ...
    def stage_value(self, path: str, value: object) -> None: ...
    def flush_now(self) -> None: ...
    def drain_listeners(self, timeout_seconds: float) -> bool: ...
    def shutdown(self, timeout_seconds: float = 1.0) -> bool: ...

Dependencies { .dependencies }

pynetworktables { .dependency }

Provides network table connectivity, configurable network update rates, manual flushing of pending network traffic, and listener queue draining.