cli.py
Small, non-interactive HA=1 staging CLI for safe offline operations.
Overview
Small, non-interactive HA=1 staging CLI for safe offline operations.
depends on agent.py config.py models.py parser.py serial_session.py serial_transport.py · used by __main__.py
flowchart TD main --> _configure main --> _observation_dict main --> _parser main --> _read_capture
API
F_observation_dict(observation: Observation) -> dict[str, object]
Convert an Observation union (DistanceReading or AccessEvent) to a dict with a kind field and all dataclass fields. Raises TypeError if the observation is not a recognized type.
mainF_read_capture(path: Path) -> list[Observation]
Read a UTF-8 sanitized capture file, parse it line by line into Observation objects, and raise ValueError if the file cannot be read, exceeds MAX_REPLAY_BYTES, or contains invalid UTF-8.
mainF_parser() -> argparse.ArgumentParser
Build an argument parser for the openaliro-ha CLI tool with subcommands for version, replay, doctor, configure, and run; the configure subcommand accepts optional device discovery and MQTT configuration flags.
mainF_prompt(input_fn: object, message: str, *, default: str | None=None) -> str
Return the user's input after stripping whitespace, or the default value if the input is empty and a default was provided.
_configure, _new_mqtt_config, _select_portF_candidate_label(index: int, port: SerialPort) -> str
Describe a candidate without echoing its device path or USB serial.
_select_portF_probe_candidate(port: SerialPort) -> Optional[str]
Return None when a candidate answers, else why it did not.
The reason is stripped of device paths so the picker keeps hiding them.
_select_portF_select_port(input_fn: object, output_fn: object) -> SerialPort
Probe all discoverable serial interfaces with USB identity and select one for the OpenAliro console. List outcomes (ready or probe failure reason), auto-select if exactly one is ready, otherwise prompt for index. Raises ConfigError if no candidates exist or selection is invalid.
_configure · calls _candidate_label, _probe_candidate, _promptF_new_mqtt_config(input_fn: object) -> MqttConfig
Interactively prompt for MQTT configuration: host, TLS port, CA certificate path, username. If username is given, prompt for password environment variable. If not, require explicit "ALLOW ANONYMOUS MQTT" confirmation. Return a populated MqttConfig struct.
_configure · calls _promptF_flag(arguments: argparse.Namespace, name: str) -> Optional[object]
Read an optional configure flag, tolerating callers that omit it.
_configure, _flag_mqtt_configF_flag_mqtt_config(arguments: argparse.Namespace) -> MqttConfig
Build an MQTT config from flags so a setup script needs no prompts.
_configure · calls _flagF_configure(arguments: argparse.Namespace, *, input_fn: object=input, output_fn: object=print) -> int
Create or add one device, from flags when given and prompts otherwise.
main · calls _flag, _flag_mqtt_config, _new_mqtt_config, _prompt, _select_portFmain(argv: Sequence[str] | None=None) -> int
Run an HA=1-gated offline command without exposing raw capture text.
_configure, _observation_dict, _parser, _read_capture