Poller

A wrapper of threading.Thread to poll a function periodically.

class duetector.tools.poller.Poller(config: dict[str, Any] | None = None, *args, **kwargs)

Bases: Configuable

A wrapper for threading.Thread

Special config:

• interval_ms: Polling interval in milliseconds

property call_when_shutdown

Whether to call func when shutdown

config_scope: str | None = 'poller'

Config scope for this poller.

default_config = {'call_when_shutdown': True, 'interval_ms': 500}

Default config for this poller.

property interval_ms

Polling interval in milliseconds

shutdown()

Shutdown poller thread. It’s safe to call this method multiple times.

After shutdown, wait should be called to wait for the thread to exit.

start(func, *args, **kwargs)

Start a poller thread, until shutdown is called.

Exceptions:

• RuntimeError: If poller thread is already started.

wait(timeout_ms=None)

Wait for poller thread to exit.

Call this method after shutdown.