host.embedded_host¶

Embedded (bare-metal / RTOS) host class.

An EmbeddedHost is a network-reached target whose “OS” is a real-time kernel or bare-metal firmware rather than a POSIX system — Zephyr is the first concrete example. It is exposed through the same Host API as UnixHost (run/oneshot/send/ expect/put/get) so test code does not care whether a target is a Linux box or a microcontroller.

What makes an embedded target different from a Unix host:

One console. A Zephyr device exposes a single shell over telnet. There is no second channel and no stateless exec primitive, so oneshot shares the one persistent session with run and is therefore not concurrency-safe (it is on UnixHost).
No bash. No $?, no command substitution, no scp/ftp/nc. Command framing and file transfer cannot reuse the Unix machinery.
Telnet only. The shell is reached over telnet (optionally through an SSH hop), never SSH directly.

Command execution requires a command frame: a CommandFrame instance that frames each command for the target’s RTOS shell over the plain telnet transport and parses the output/return-code back. There is no default frame — a bare EmbeddedHost raises ValueError at construction if none is supplied (fail loud). The frame is provided by:

a registered OsProfile data bundle (e.g. a command_frame key in an [os_profiles.<name>] settings table), or
a concrete subclass that re-declares the default, or
an explicit constructor argument.

ZephyrHost is the in-tree concrete class: it subclasses EmbeddedHost and declares ZephyrFrame as the default command_frame (along with os_type='zephyr' and os_name='Zephyr'). Zephyr-specific framing and OS naming live on ZephyrHost, not on the base class.

File transfer (get/put) is delegated to EmbeddedFileTransfer, which speaks the device shell only (the console backend uses Zephyr’s fs commands). The interactive bridge (_interact) currently raises NotImplementedError.

class otto.host.embedded_host.EmbeddedHost(ip: str, element: str, os_type: OsType = 'embedded', os_name: str | None = None, os_version: str | None = None, name: str = None, creds: dict[str, str] = <factory>, user: str | None = None, element_id: int | None = None, board: str | None = None, slot: int | None = None, is_virtual: bool = False, term: str = 'telnet', transfer: str = 'console', valid_terms: list[str] = <factory>, valid_transfers: list[str] = <factory>, filesystem: EmbeddedFileSystem = <factory>, command_frame: CommandFrame | None = None, loader: BinaryLoader | None = None, default_dest_dir: Path = <factory>, max_filename_len: int = 255, telnet_options: TelnetOptions = <factory>, snmp: SnmpOptions | None = None, toolchain: Toolchain = <factory>, hop: str | None = None, resources: set[str] = <factory>, interfaces: dict[str, str] = <factory>, products: list['Product'] = <factory>, power_control: PowerController | None = None, log: bool = True, log_stdout: bool = True, _connection_factory: type[ConnectionManager] | None = None, *, _lab: Lab | None = None)¶

Bases: RemoteHost

OS-agnostic bare-metal / RTOS host reached over telnet.

EmbeddedHost carries no OS-specific defaults. A command_frame must be supplied — either via a profile, a subclass (e.g. ZephyrHost), or an explicit constructor argument — or construction raises ValueError (fail loud). ZephyrHost is the in-tree concrete subclass and worked example.

ip : str¶: IP address of the host’s telnet shell.

element : str¶: Network element to which this host belongs.

os_type : OsType¶: Default profile selector for a bare EmbeddedHost. Subclasses (e.g. ZephyrHost) override this to their registered name.

os_name : str | None¶: Kernel/OS name, or None. A bare embedded host carries no OS name; a concrete subclass (e.g. ZephyrHost) sets it.

os_version : str | None¶: OS/kernel version string, or None if unspecified.

name : str¶: Human readable name to represent the host. Automatically generated if not provided.

creds : dict[str, str]¶: Users and their respective passwords. Optional — the Zephyr telnet shell backend has no login step, so this is empty for a stock Zephyr target.

user : str | None¶: User with which to log in, if the shell requires one. Usually unset.

element_id : int | None¶: Network element identifier to which this host belongs.

board : str | None¶: Name of the board type to which this host belongs.

slot : int | None¶: Physical slot number of the board to which this host belongs.

is_virtual : bool¶: Determines whether a host is a VM/emulator (e.g. QEMU) or not.

term : str¶: Active session transport. Embedded hosts speak telnet today; the command frame is transport-independent, so this is not a hard coupling.

transfer : str¶: File-transfer backend. console (default) drives the device shell’s fs commands; tftp is reserved and not yet implemented.

valid_terms : list[str]¶: Closed menu of term backends this host supports (active is term).

valid_transfers : list[str]¶: Closed menu of transfer backends this host supports (active is transfer).

filesystem : EmbeddedFileSystem¶

On-device filesystem variant — e.g. FatRamFileSystem, LittleFsFileSystem, or NoFileSystem (the default). Carries the mount path, the optional fs mount command, and the command-formation hooks the transfer code and the embedded monitor’s disk parser drive. See otto.host.embedded_filesystem.

Lab data declares the variant by string in the filesystem field; the storage factory resolves the string to a class. Projects can register custom variants via otto.host.embedded_filesystem.register_filesystem().

command_frame : CommandFrame | None¶

Shell-framing dialect for this target’s console — how a command is wrapped in sentinels and how output/retcode are parsed back. There is NO default: a bare embedded host carries no dialect, so a frame is required — supplied either by a profile/subclass (e.g. ZephyrHost) or as an explicit value. A frame-less EmbeddedHost fails loud at construction.

Lab data declares the dialect by string in the command_frame field (e.g. a Zephyr 2.7 build that reports its retcode inline would name a project-registered frame); the storage factory resolves the string to an instance. Projects can register custom dialects via otto.host.command_frame.register_command_frame(). The dialect is independent of the transport, so it is handed straight to the SessionManager.

loader : BinaryLoader | None¶: Binary-load strategy for this target’s runtime (e.g. Zephyr LLEXT). Unlike command_frame it is optional — many embedded hosts never load binaries. Lab data declares it by string in the loader field (e.g. "llext-hex"); __post_init__ resolves the string to an instance. load() / unload() fail loud (ValueError) when it is None. Projects register custom loaders via otto.host.binary_loader.register_binary_loader().

default_dest_dir : Path¶: Default landing directory for put / get when the caller supplies an empty or relative dest_dir. When left at the default (an empty Path()), __post_init__ resolves it to filesystem.mount so generic fan-out callers like do_for_all_hosts don’t have to branch on host type. Override in lab data to land transfers somewhere other than the FS root. See default_dest_dir.

max_filename_len : int¶: Upper bound on the basename length (including extension) accepted by the target’s filesystem. Defaults to 255 — the Linux NAME_MAX, also the typical LittleFS ceiling. Override per-host when the firmware enforces a tighter limit (e.g. 32 for a Zephyr build that sets CONFIG_FS_FATFS_MAX_LFN=32 / CONFIG_FS_LITTLEFS_NAME_MAX=32, or 12 for a stock FAT 8.3 build without LFN support). See max_filename_len.

telnet_options : TelnetOptions¶: Connection options for the telnet shell (port, cols/rows, etc.).

snmp : SnmpOptions | None¶: Optional SNMP polling config (lab snmp block). When set, otto’s monitor collects this host’s metrics over SNMP — a separate channel from the single telnet console — instead of running shell commands. See SnmpOptions.

toolchain : Toolchain¶: Cross-toolchain for this bed’s products. Used by the coverage pipeline to select the correct gcov/lcov. The host is the test bed, so it owns the toolchain matching its target ABI — a Zephyr 3.7 bed and a 4.4 bed declare different SDKs. Defaults to system-installed tools.

hop : str | None¶: Host ID of the intermediate SSH hop used to reach this host, or None.

resources : set[str]¶: Names of resources required to use this host.

interfaces : dict[str, str]¶: Named secondary interface addresses (see interfaces). Resolve with address_for().

products : list['Product']¶: Software-under-test deployed to this host. Default empty. See products.

power_control : PowerController | None¶: Pluggable power backend. Lab data declares it by string (a config-free controller type) or a [power] table ({type, on_cmd, off_cmd, ...}); __post_init__ coerces it to an instance. None → power()/reboot(hard=True) fail loud. See power_control.

log : bool¶: Whether this host should log its output to stdout and log files.

log_stdout : bool¶: Whether this host should log its output to stdout.

id : str¶: Unique identifier for this host.

async verify_connection() → CommandStatus¶: Attempt to open the telnet shell without running commands (dry-run).

async close() → None¶

async oneshot(cmd: str, timeout: float | None = None, log: bool = True) → CommandStatus¶

Run a single command on the embedded host.

Unlike oneshot(), this is not concurrency-safe: an embedded target exposes a single console with no stateless exec primitive, so oneshot runs on the same persistent session as run. It exists for API parity; use run for stateful workflows.

async open_session(name: str) → HostSession¶

Open a named persistent shell session.

Note: an embedded target has a single console. Opening a second named session opens a second telnet connection to the device, which most RTOS shell backends do not accept concurrently. Prefer the default session via run.

async send(text: str, log: bool = True) → None¶: Send raw text to the host’s persistent session.

async expect(pattern: str | Pattern[str], timeout: float = 10.0) → str¶: Wait for a pattern in the host’s session output stream.

async get(src_files: ~types.Annotated[list[~pathlib.Path] | ~pathlib.Path, ~otto.utils.Arg(variadic=True, elem_type=~pathlib.Path, name=None, help=Remote file(s) to download.)], dest_dir: ~pathlib.Path, show_progress: ~typing.Annotated[bool, <otto.utils._Exclude object at 0x7fcc41e8fea0>] = True) → tuple[Status, str]¶

Transfer files from the embedded host to the local machine.

Delegates to EmbeddedFileTransfer, which speaks the device shell (the console backend uses Zephyr’s fs commands). Transfers are sequential — an embedded target has a single console.

async put(src_files: ~types.Annotated[list[~pathlib.Path] | ~pathlib.Path, ~otto.utils.Arg(variadic=True, elem_type=~pathlib.Path, name=None, help=Local file(s) to upload.)], dest_dir: ~pathlib.Path, show_progress: ~typing.Annotated[bool, <otto.utils._Exclude object at 0x7fcc41e8fea0>] = True) → tuple[Status, str]¶

Transfer files from the local machine to the embedded host.

Delegates to EmbeddedFileTransfer (the console backend writes via Zephyr’s chunked fs write). Transfers are sequential — an embedded target has a single console.

dest_dir is resolved against default_dest_dir so a generic Path() from a fan-out caller lands on the host’s mounted filesystem (e.g. /RAM: on a FAT target) rather than on Zephyr’s bare /, which has no FS and rejects opens with -ENOENT.

async exists(path: str | Path) → bool¶: Return True when path exists on the device (via fs ls).

async ls(path: Annotated[str | Path, Arg(variadic=False, elem_type=None, name=None, help=None)] = '.', all: bool = False) → list[str]¶: List entry names in path via the device fs ls former.

async rm(path: str | Path, recursive: bool = False, force: bool = False) → tuple[Status, str]¶: Remove path via the device fs rm former (flags ignored).

async mkdir(path: str | Path, parents: bool = True) → tuple[Status, str]¶

async cp(src: str | Path, dst: str | Path, recursive: bool = False) → tuple[Status, str]¶

async mv(src: str | Path, dst: str | Path) → tuple[Status, str]¶

async read_file(path: str | Path) → str¶

async write_file(path: str | Path, data: str, append: bool = False) → tuple[Status, str]¶

async load(file: Path, name: str, show_progress: bool = False, timeout: float | None = 120.0) → tuple[Status, str]¶

Load a binary into the device runtime via the host’s binary loader.

Distinct from put() (a file transfer to a mounted filesystem): load pushes a binary into the target’s loader (e.g. Zephyr LLEXT’s llext load_hex), with no destination file. The payload is read from file, formatted into the device command by the loader, and sent with log=False so the (large) encoded payload never reaches the console or log. Returns (Status, str) like put()/get(); the str carries the device’s failure text on error.

show_progress is off by default (the bar only renders in interactive / otto run; under otto test output is captured). When enabled it drives a transfer-style Rich bar from the paced telnet write of the payload — the only measurable progress (the device’s relocation emits no incremental signal). Fails loud (ValueError) if the host declares no loader.

async unload(name: str, timeout: float | None = 20.0) → tuple[Status, str]¶

Unload name from the device runtime, draining to full eviction.

Some loaders (LLEXT) refcount a resident binary, so one unload may only decrement it. unload loops the loader’s unload command until is_fully_unloaded() reports the binary gone (bounded by loader.max_unload_rounds). Idempotent: unloading something not loaded succeeds on the first round. Returns (Status, str); fails loud (ValueError) if no loader is declared.

class otto.host.embedded_host.ZephyrHost(ip: str, element: str, os_type: OsType = 'zephyr', os_name: str | None = 'Zephyr', os_version: str | None = None, name: str = None, creds: dict[str, str] = <factory>, user: str | None = None, element_id: int | None = None, board: str | None = None, slot: int | None = None, is_virtual: bool = False, term: str = 'telnet', transfer: str = 'console', valid_terms: list[str] = <factory>, valid_transfers: list[str] = <factory>, filesystem: EmbeddedFileSystem = <factory>, command_frame: CommandFrame = <factory>, loader: BinaryLoader | None = None, default_dest_dir: Path = <factory>, max_filename_len: int = 255, telnet_options: TelnetOptions = <factory>, snmp: SnmpOptions | None = None, toolchain: Toolchain = <factory>, hop: str | None = None, resources: set[str] = <factory>, interfaces: dict[str, str] = <factory>, products: list['Product'] = <factory>, power_control: PowerController | None = None, log: bool = True, log_stdout: bool = True, _connection_factory: type[ConnectionManager] | None = None, *, _lab: Lab | None = None)¶

Bases: EmbeddedHost

A Zephyr RTOS host — the concrete, registered embedded host.

This is the worked example for shipping a host subclass: it re-declares the Zephyr-specific field defaults that EmbeddedHost no longer assumes, and is registered under os_type: "zephyr" via otto.host.os_profile.register_host_class(). External repositories register their own EmbeddedHost/UnixHost subclasses the same way (from an init module listed in .otto/settings.toml), and may layer per-build OsProfile data bundles over them.

ip : str¶: IP address of the host’s telnet shell.

element : str¶: Network element to which this host belongs.

os_version : str | None¶: OS/kernel version string, or None if unspecified.

name : str¶: Human readable name to represent the host. Automatically generated if not provided.

creds : dict[str, str]¶: Users and their respective passwords. Optional — the Zephyr telnet shell backend has no login step, so this is empty for a stock Zephyr target.

user : str | None¶: User with which to log in, if the shell requires one. Usually unset.

element_id : int | None¶: Network element identifier to which this host belongs.

board : str | None¶: Name of the board type to which this host belongs.

slot : int | None¶: Physical slot number of the board to which this host belongs.

is_virtual : bool¶: Determines whether a host is a VM/emulator (e.g. QEMU) or not.

term : str¶: Active session transport. Embedded hosts speak telnet today; the command frame is transport-independent, so this is not a hard coupling.

transfer : str¶: File-transfer backend. console (default) drives the device shell’s fs commands; tftp is reserved and not yet implemented.

valid_terms : list[str]¶: Closed menu of term backends this host supports (active is term).

valid_transfers : list[str]¶: Closed menu of transfer backends this host supports (active is transfer).

filesystem : EmbeddedFileSystem¶

loader : BinaryLoader | None¶: Binary-load strategy for this target’s runtime (e.g. Zephyr LLEXT). Unlike command_frame it is optional — many embedded hosts never load binaries. Lab data declares it by string in the loader field (e.g. "llext-hex"); __post_init__ resolves the string to an instance. load() / unload() fail loud (ValueError) when it is None. Projects register custom loaders via otto.host.binary_loader.register_binary_loader().

default_dest_dir : Path¶: Default landing directory for put / get when the caller supplies an empty or relative dest_dir. When left at the default (an empty Path()), __post_init__ resolves it to filesystem.mount so generic fan-out callers like do_for_all_hosts don’t have to branch on host type. Override in lab data to land transfers somewhere other than the FS root. See default_dest_dir.

max_filename_len : int¶: Upper bound on the basename length (including extension) accepted by the target’s filesystem. Defaults to 255 — the Linux NAME_MAX, also the typical LittleFS ceiling. Override per-host when the firmware enforces a tighter limit (e.g. 32 for a Zephyr build that sets CONFIG_FS_FATFS_MAX_LFN=32 / CONFIG_FS_LITTLEFS_NAME_MAX=32, or 12 for a stock FAT 8.3 build without LFN support). See max_filename_len.

telnet_options : TelnetOptions¶: Connection options for the telnet shell (port, cols/rows, etc.).

snmp : SnmpOptions | None¶: Optional SNMP polling config (lab snmp block). When set, otto’s monitor collects this host’s metrics over SNMP — a separate channel from the single telnet console — instead of running shell commands. See SnmpOptions.

toolchain : Toolchain¶: Cross-toolchain for this bed’s products. Used by the coverage pipeline to select the correct gcov/lcov. The host is the test bed, so it owns the toolchain matching its target ABI — a Zephyr 3.7 bed and a 4.4 bed declare different SDKs. Defaults to system-installed tools.

hop : str | None¶: Host ID of the intermediate SSH hop used to reach this host, or None.

resources : set[str]¶: Names of resources required to use this host.

interfaces : dict[str, str]¶: Named secondary interface addresses (see interfaces). Resolve with address_for().

products : list['Product']¶: Software-under-test deployed to this host. Default empty. See products.

power_control : PowerController | None¶: Pluggable power backend. Lab data declares it by string (a config-free controller type) or a [power] table ({type, on_cmd, off_cmd, ...}); __post_init__ coerces it to an instance. None → power()/reboot(hard=True) fail loud. See power_control.

log : bool¶: Whether this host should log its output to stdout and log files.

log_stdout : bool¶: Whether this host should log its output to stdout.

id : str¶: Unique identifier for this host.

os_type : OsType¶: Profile selector recorded on the host. zephyr for this class.

os_name : str | None¶: Kernel/OS name — Zephyr for this class.

command_frame : CommandFrame¶: Stock Zephyr retval shell framing (3.7 / 4.4 LTS).