agent.py
Runnable standalone-agent orchestration over the shared serial library.
Overview
Runnable standalone-agent orchestration over the shared serial library.
depends on config.py models.py mqtt.py serial_session.py serial_transport.py · used by __init__.py cli.py
API
Cclass AgentError(RuntimeError)
A safe agent failure with no raw serial, broker, or console details.
doctor, probe_device, run_deviceCclass DoctorDeviceResult
One diagnostics-safe device result.
doctorF_publisher(config: AgentConfig, device: DeviceConfig) -> MqttPublisher
Create an MQTT publisher from the agent and device configuration.
Fsession_for_device(device: DeviceConfig) -> SerialSession
Build a reconnecting serial session that resolves `auto` on every open.
probe_deviceFconnection_factory()
Factory coroutine to open a serial connection to the device: resolve the serial port from device.serial_port and device.serial_identity, then open the connection at device.baud.
Fdoctor(config: AgentConfig, *, publisher_factory: PublisherFactory=_publisher, session_factory: Callable[[DeviceConfig], SerialSession]=session_for_device) -> tuple[DoctorDeviceResult, ...]
Validate serial protocol and broker connectivity without publishing events.
AgentError, DoctorDeviceResultFprobe_device(device: DeviceConfig) -> SessionState
Probe one selected port before configuration persists any settings.
AgentError, session_for_deviceCclass _DistanceThrottle
Rate-limit distance publishes without hiding real movement.
Ranging emits one reading per block, roughly every 192 ms, which is far more than Home Assistant needs. Publish at most once per interval, but let a large change through immediately so an approach or retreat is never delayed.
run_deviceF_DistanceThrottle.__init__(self, *, min_interval: float=DISTANCE_MIN_INTERVAL_S, significant_change_mm: int=DISTANCE_SIGNIFICANT_CHANGE_MM, clock: Callable[[], float]=time.monotonic) -> None
Initialize the distance throttle with optional minimum publish interval in seconds, minimum significant change in millimeters, and a clock function. Default min_interval is DISTANCE_MIN_INTERVAL_S, default significant_change_mm is DISTANCE_SIGNIFICANT_CHANGE_MM, and default clock is time.monotonic.
F_DistanceThrottle.allows(self, distance_mm: int) -> bool
Report whether this reading should be published, recording it if so.
run_deviceFrun_device(config: AgentConfig, device: DeviceConfig, stop_event: asyncio.Event, *, publisher_factory: PublisherFactory=_publisher, session_factory: Callable[[DeviceConfig], SerialSession]=session_for_device) -> None
Publish one device's approved observations until a caller stops the agent.
run · calls AgentError, _DistanceThrottle, _DistanceThrottle.allowsFrun(config: AgentConfig, stop_event: Optional[asyncio.Event]=None) -> None
Run every configured lock concurrently until the supplied event is set.
run_device