serial_session.py
Async, transport-neutral ownership of one OpenAliro serial console.
Overview
Async, transport-neutral ownership of one OpenAliro serial console.
The session is deliberately independent of pyserial and Home Assistant. A
runtime adapter provides an opened byte-stream; this module serializes shell
commands, parses only the approved observations, and never retains raw console
lines. The device can be idle after `aliro frames on`: a stream acknowledgement
is therefore the capability probe, not the first range reading.
depends on compatibility.py models.py parser.py · used by __init__.py agent.py cli.py
API
Cclass SerialConnection(Protocol)
The small async byte-stream contract needed by `SerialSession`.
FSerialConnection.readline(self) -> bytes
Return one newline-delimited serial line or `b""` on disconnect.
SerialSession._read_loopFSerialConnection.write(self, data: bytes) -> None
Queue bytes for delivery to the console.
SerialSession._commandFSerialConnection.close(self) -> None
Release the serial device.
SerialSession.close, SerialSession.maintain, SerialSession.startCclass SessionState(str, Enum)
Externally visible lifecycle states with no raw device details.
Cclass SerialSessionError(RuntimeError)
A safe, user-facing serial session failure.
SerialSession._command, SerialSession._read_loop, SerialSession.close, SerialSession.poll_compatibility_range, SerialSession.startCclass _ResponseHandler(Protocol)
Protocol for command response handlers: feed one line of console output and return a tuple of (is_complete, result), where is_complete indicates the full response has arrived.
F_ResponseHandler.feed(self, line: str) -> tuple[bool, object]
Return whether a command response is complete and its safe result.
SerialSession._feed_responseCclass _ContainsResponse
Handler that returns true and a provided result as soon as the expected substring appears anywhere in a line, ignoring ANSI escape codes.
SerialSession.startF_ContainsResponse.__init__(self, expected: str, result: Result) -> None
Record the substring to match and the result to return when the match is found.
F_ContainsResponse.feed(self, line: str) -> tuple[bool, object]
Return whether the expected substring is present in the line after stripping ANSI codes, and the pre-set result.
Cclass _StreamResponse
Accept the firmware's frames acknowledgement and retain its actual mode.
SerialSession.startF_StreamResponse.feed(self, line: str) -> tuple[bool, object]
Return whether the ACKNOWLEDGEMENT text is present in the line, and if so whether the response indicates the stream is on or off.
Cclass _RangeResponse
Handler that parses the multiline output of the aliro range command and returns true with the parsed RangeReading when available, or true with None when the parser confirms the response is finished.
SerialSession.poll_compatibility_rangeF_RangeResponse.__init__(self) -> None
Initialize the handler and prime the internal range response parser.
F_RangeResponse.feed(self, line: str) -> tuple[bool, object]
Feed a line to the internal parser; return true with a RangeReading if a complete range measurement is available, true with None if the parser confirms the response finished, or false if more input is needed.
Cclass SerialSession
Read console observations while issuing one shell command at a time.
FSerialSession.__init__(self, connection_factory: ConnectionFactory, *, command_timeout: float=3.0, observation_queue_size: int=256) -> None
Initialize a serial session with the given connection factory and optional command and queue limits. Raises ValueError if command_timeout or observation_queue_size is not positive.
FSerialSession.state(self) -> SessionState
Return the current lifecycle state.
FSerialSession.observations(self) -> asyncio.Queue[Observation]
Expose parsed observations without exposing the raw console.
FSerialSession.start(self) -> SessionState
Open, probe, and prepare the console without changing lock state.
SerialSession.maintain · calls SerialConnection.close, SerialSession._command, SerialSession._read_loop, SerialSessionError, _ContainsResponse, _StreamResponseFSerialSession.poll_compatibility_range(self) -> Optional[CompatibilityRangeReading]
Read one `aliro range` response while preserving unsolicited events.
SerialSession._command, SerialSessionError, _RangeResponseFSerialSession.maintain(self, stop_event: asyncio.Event, *, retry_delay: float=1.0) -> None
Reconnect until stopped, with a bounded caller-selected delay.
SerialConnection.close, SerialSession._wait_for_disconnect_or_stop, SerialSession.startFSerialSession._wait_for_disconnect_or_stop(self, stop_event: asyncio.Event) -> None
Wake promptly for either a transport loss or a caller-requested stop.
SerialSession.maintainFSerialSession.close(self) -> None
Stop I/O, fail any pending command, and close the owned transport.
SerialConnection.close, SerialSession._fail_pending, SerialSessionErrorFSerialSession._command(self, command: str, handler: _ResponseHandler) -> object
Send a command string to the serial console, invoke the given response handler on each line of reply, and return the handler's result when the response is complete. Raises SerialSessionError if the connection is closed or the command times out.
SerialSession.poll_compatibility_range, SerialSession.start · calls SerialConnection.write, SerialSessionErrorFSerialSession._read_loop(self) -> None
Coroutine: read lines from the serial console indefinitely until disconnected. Parse each line for observations (distance/access events) and enqueue them; feed unparsed lines to the current command's response handler. On disconnect or queue overflow, fail all pending responses and mark the session disconnected.
SerialSession.start · calls SerialConnection.readline, SerialSession._fail_pending, SerialSession._feed_response, SerialSessionErrorFSerialSession._feed_response(self, line: str) -> None
Feed a line of console output to the current command's response handler, and set the result on the pending future if the handler reports the response is complete.
SerialSession._read_loop · calls _ResponseHandler.feedFSerialSession._fail_pending(self, error: SerialSessionError) -> None
Set an exception on any pending command response future to signal that the response will not arrive, used when the connection fails or is closed.
SerialSession._read_loop, SerialSession.close