host.transfer¶

File transfer backends for otto hosts — one package, both host families.

Public API (also re-exported from otto.host): register_transfer_backend, build_transfer_backend, the Rich progress helpers, and the Nc* selector Literals. Importing this package registers every built-in backend.

class otto.host.transfer.BaseFileTransfer(name: str, max_filename_len: int = 255)¶

Bases: ABC

Shared API + progress plumbing for any file-transfer backend.

The public put_files / get_files surface (filename-length validation, shared Rich progress acquisition) is owned by this base. Concrete backends (Unix’s UnixFileTransfer subclasses (ScpFileTransfer, SftpFileTransfer, FtpFileTransfer, NcFileTransfer), embedded’s EmbeddedFileTransfer subclasses (ConsoleFileTransfer, TftpFileTransfer), and any future ones) implement two abstract methods — _run_put and _run_get — both of which receive a TransferProgressFactory and are responsible for invoking it at least once per source file, terminating with bytes_done == bytes_total to mark completion.

The progress-bar capability is enforced at the type system level: abc.abstractmethod refuses to instantiate a subclass that omits either method, so a new backend cannot be defined without supplying a way to report progress. The runtime contract test (TestTransferProgressContract) verifies the factory is actually invoked, not just that the methods exist.

host_families : frozenset[str] = frozenset({})¶: Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

classmethod create(ctx: TransferContext) → BaseFileTransfer¶

Build a transfer backend from a TransferContext.

The uniform construction seam (WS#4). Concrete backends override this to run their exact construction against the ctx fields they need. Not an abstractmethod deliberately: only registered built-ins are ever constructed through create, and test doubles that subclass BaseFileTransfer only to exercise the progress contract must not be forced to implement it.

async put_files(src_files: list[Path], dest_dir: Path, show_progress: bool = True) → tuple[Status, str]¶

async get_files(src_files: list[Path], dest_dir: Path, show_progress: bool = True) → tuple[Status, str]¶

class otto.host.transfer.TransferContext(transfer: str, host_name: str, max_filename_len: int = 255, exec_cmd: collections.abc.Callable[..., Coroutine[Any, Any, CommandStatus]] | None = None, connections: ConnectionManager | None = None, nc_options: NcOptions | None = None, scp_options: ScpOptions | None = None, get_local_ip: collections.abc.Callable[[], str] | None = None, filesystem: EmbeddedFileSystem | None = None)¶

Bases: object

Construction inputs a host provides to build its transfer backend via BaseFileTransfer.create(). The frozen public seam for custom transfer backends. Carries the union of what any family’s built-ins receive at their call sites; a unix backend reads the unix fields, an embedded backend the embedded ones. Selector validation (host-family applicability) runs before construction, so a backend never sees a ctx missing the fields it needs.

transfer : str¶

host_name : str¶

max_filename_len : int = 255¶

exec_cmd : Callable[..., Coroutine[Any, Any, CommandStatus]] | None = None¶

connections : ConnectionManager | None = None¶

nc_options : NcOptions | None = None¶

scp_options : ScpOptions | None = None¶

get_local_ip : Callable[[], str] | None = None¶

filesystem : EmbeddedFileSystem | None = None¶

otto.host.transfer.validate_filename_lengths(files: list[Path], limit: int, host_name: str) → tuple[Status, str]¶

Reject files whose basename exceeds the host’s filesystem cap.

Shared by UnixFileTransfer (Unix) and EmbeddedFileTransfer (embedded) so every backend surfaces the same self-explaining error. Without this guard the failure modes are:

Unix SCP/SFTP/FTP: server returns File name too long (errno 36), mid-transfer, after the local file is already read.
Embedded FAT (8.3, no LFN) or LittleFS over NAME_MAX: device fails fs_open with -ENOENT, giving no hint that the name was the problem.

Returns (Status.Success, '') when every basename fits.

otto.host.transfer.TransferProgressHandler¶: alias of Callable[[str, str, int, int], None]

otto.host.transfer.TransferProgressFactory¶: alias of Callable[[], Callable[[str, str, int, int], None]]

otto.host.transfer.make_rich_progress_handler(progress: Progress, host_name: str) → TransferProgressHandler¶

Return a TransferProgressHandler that drives the given Rich Progress bar.

One task is created per source file, detected by a change in src_path. The caller is responsible for the Progress context (entering and exiting it).

Example:

with make_transfer_progress() as progress:
    handler = make_rich_progress_handler(progress, host_name=host.hostname)
    status, err = await host.get(files, dest, progress_handler=handler)

otto.host.transfer.make_rich_progress_factory(progress: Progress, host_name: str) → TransferProgressFactory¶

Return a factory that creates a fresh TransferProgressHandler per file.

Each call to the returned factory produces an independent handler with its own closure state, so concurrent transfers don’t share progress tracking.

Example:

with make_transfer_progress() as progress:
    factory = make_rich_progress_factory(progress, host_name=host.name)
    status, err = await host.put(files, dest)

otto.host.transfer.make_transfer_progress() → Progress¶: Return a pre-configured Rich Progress suited for file transfers.

otto.host.transfer.register_transfer_backend(name: str, cls: type[BaseFileTransfer]) → None¶

Make a custom transfer backend available to lab data under name.

Call from an init module listed in .otto/settings.toml. The backend must declare a non-empty BaseFileTransfer.host_families; otherwise it could never validate against any host and is rejected here.

otto.host.transfer.build_transfer_backend(name: str) → type[BaseFileTransfer]¶

Return the transfer-backend class registered under name.

Raises¶

ValueError: If name is not registered; the message lists the registered names.

class otto.host.transfer.NcFileTransfer(connections: ConnectionManager, name: str, transfer: str, nc_options: NcOptions, get_local_ip: Callable[[], str], exec_cmd: Callable[[...], Coroutine[Any, Any, CommandStatus]], max_filename_len: int = 255)¶

Bases: UnixFileTransfer

Handles netcat file transfers for a UnixHost.

Receives injectable callables for open_session and oneshot so it can be tested without real connections.

Inherits put_files / get_files from BaseFileTransfer and unix scaffolding (_connections, _exec_cmd, _warmup_for_transfer) from UnixFileTransfer; implements the abstract _run_put / _run_get as direct calls to _put_files_nc / _get_files_nc.

host_families : frozenset[str] = frozenset({'unix'})¶: Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

classmethod create(ctx: TransferContext) → NcFileTransfer¶

Build a transfer backend from a TransferContext.

async prepare() → None¶

Resolve port + listener strategies in a single round-trip.

Runs the shared _STRATEGY_PROBE script through _control_run so the port and listener strategies are resolved up front rather than lazily at first-transfer time. Idempotent — a second call with both strategies already cached is a no-op.

Callers use _warmup_for_transfer to run this concurrently with exec-pool warming; direct callers can invoke prepare() alone.

If the probe itself fails (non-zero exit, malformed output), the caches stay unset and the lazy cascades in _find_free_port_auto / _resolve_listener_strategy still kick in as fallbacks.

class otto.host.transfer.UnixFileTransfer(connections: ConnectionManager, name: str, exec_cmd: Callable[[...], Coroutine[Any, Any, CommandStatus]], max_filename_len: int = 255)¶

Bases: BaseFileTransfer

Common unix scaffolding shared by all SSH/Telnet transfer backends.

Stores the two mandatory unix fields (_connections, _exec_cmd), provides _warmup_for_transfer (concurrent strategy-probe + pool warming), and supplies a no-op prepare() that subclasses override when they need a real probe.

host_families : frozenset[str] = frozenset({'unix'})¶: Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

async prepare() → None¶: No-op default — subclasses override to run a strategy probe.

class otto.host.transfer.EmbeddedFileTransfer(name: str, exec_cmd: Callable[[...], Coroutine[Any, Any, CommandStatus]] | None = None, filesystem: EmbeddedFileSystem | None = None, max_filename_len: int = 255, transfer: str | None = None)¶

Bases: BaseFileTransfer

Shared base for embedded file-transfer backends (console, tftp).

Subclasses BaseFileTransfer, inheriting its put_files / get_files API (filename-length validation, shared Rich progress acquisition). Holds the fields common to all embedded selectors: the shell runner (exec_cmd), the on-device filesystem model, and the idempotent mount-done flag. Concrete subclasses implement _run_put / _run_get.

host_families : frozenset[str] = frozenset({'embedded'})¶: Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

classmethod create(ctx: TransferContext) → EmbeddedFileTransfer¶

Construct from a TransferContext.

Subclasses override this to read selector-specific ctx fields.

class otto.host.transfer.ConsoleFileTransfer(name: str, exec_cmd: Callable[[...], Coroutine[Any, Any, CommandStatus]], filesystem: EmbeddedFileSystem | None = None, max_filename_len: int = 255, transfer: str | None = None)¶

Bases: EmbeddedFileTransfer

File transfer for an embedded host, over the device shell only.

Subclasses EmbeddedFileTransfer, inheriting its put_files / get_files API (filename-length validation, shared Rich progress acquisition). Implements the abstract _run_put / _run_get against the device’s fs shell. The shell command runner is injected as exec_cmd so the class is testable against a fake shell with no real connection.

classmethod create(ctx: TransferContext) → ConsoleFileTransfer¶

Construct from a TransferContext.

Subclasses override this to read selector-specific ctx fields.

class otto.host.transfer.TftpFileTransfer(name: str, exec_cmd: Callable[[...], Coroutine[Any, Any, CommandStatus]] | None = None, filesystem: EmbeddedFileSystem | None = None, max_filename_len: int = 255, transfer: str | None = None)¶

Bases: EmbeddedFileTransfer

Reserved: TFTP transfer for embedded hosts is not yet implemented.

classmethod create(ctx: TransferContext) → TftpFileTransfer¶

Construct from a TransferContext.

Subclasses override this to read selector-specific ctx fields.

class otto.host.transfer.ScpFileTransfer(connections: ConnectionManager, name: str, exec_cmd: Callable[[...], Coroutine[Any, Any, CommandStatus]], scp_options: ScpOptions, max_filename_len: int = 255)¶

Bases: UnixFileTransfer

SCP file transfer backend for UnixHost.

Inherits put_files / get_files from BaseFileTransfer and unix scaffolding (_connections, _exec_cmd, _warmup_for_transfer) from UnixFileTransfer; implements _run_put / _run_get directly for the SCP protocol.

host_families : frozenset[str] = frozenset({'unix'})¶: Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

classmethod create(ctx: TransferContext) → ScpFileTransfer¶

Build a transfer backend from a TransferContext.

class otto.host.transfer.SftpFileTransfer(connections: ConnectionManager, name: str, exec_cmd: Callable[[...], Coroutine[Any, Any, CommandStatus]], max_filename_len: int = 255)¶

Bases: UnixFileTransfer

SFTP file transfer backend for UnixHost.

Inherits put_files / get_files from BaseFileTransfer and unix scaffolding (_connections, _exec_cmd, _warmup_for_transfer) from UnixFileTransfer; implements _run_put / _run_get directly for the SFTP protocol.

host_families : frozenset[str] = frozenset({'unix'})¶: Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

classmethod create(ctx: TransferContext) → SftpFileTransfer¶

Build a transfer backend from a TransferContext.

class otto.host.transfer.FtpFileTransfer(connections: ConnectionManager, name: str, exec_cmd: collections.abc.Callable[..., Coroutine[Any, Any, CommandStatus]], max_filename_len: int = 255)¶

Bases: UnixFileTransfer

FTP file transfer backend for UnixHost.

Inherits put_files / get_files from BaseFileTransfer and unix scaffolding (_connections, _exec_cmd, _warmup_for_transfer) from UnixFileTransfer; implements _run_put / _run_get directly for the FTP protocol.

host_families : frozenset[str] = frozenset({'unix'})¶: Host-family selectors this backend serves — a subset of {'unix', 'embedded'}. Subclasses declare it; the spec field_validator rejects a backend on a host of the wrong family. A backend with an empty set can never validate and is rejected at registration.

classmethod create(ctx: TransferContext) → FtpFileTransfer¶

Build a transfer backend from a TransferContext.