embedl_hub.core.device package#

Device abstraction, management, and configuration.

Re-exports#

CommandResult — Protocol for command execution results.
CommandRunner — Protocol for running commands on a device.
Device — Concrete device representation.
DeviceManager — Factory for creating device instances.
DeviceSpec — Hardware specification for a device.
SSHConfig — SSH connection configuration.
SSHCommandRunner — SSH-based command runner.
SSHConnectionError — Raised on SSH connection failure.
ProviderConfig — Base class for provider configurations.
EmbedlONNXRuntimeConfig — Config for the Embedl ONNX Runtime provider.
TrtexecConfig — Config for the TensorRT (trtexec) provider.

class embedl_hub.core.device.CommandResult(*args, **kwargs)[source]#

Bases: Protocol

Protocol for the result of executing a command on a device.

stderr: str | None#

stdout: str | None#

class embedl_hub.core.device.CommandRunner(*args, **kwargs)[source]#

Bases: Protocol

Protocol for executing commands and transferring files on a device.

get(source: PurePosixPath, destination: Path) → None[source]#

put(source: Path, destination: PurePosixPath) → None[source]#

run(command: Sequence[str], cwd: PurePosixPath | None = None, *, hide: bool = False) → CommandResult[source]#

class embedl_hub.core.device.Device(name: str, runner: CommandRunner | None, spec: DeviceSpec, provider_type: str, provider_config: ProviderConfig | None = None, provider_config_overrides: dict[type[Component], ProviderConfig] = <factory>)[source]#

Bases: object

Represents a target device for component execution.

Directory and artifact management is delegated to an internal RemoteArtifactManager. The most commonly used properties (artifact_base_dir, artifact_dir, last_download_dir) and the download_artifact_dir method are exposed as public delegates for convenience.

Parameters:

name – A human-readable label for this device (e.g. 'main', 'jetson-orin'). Used as the dictionary key in HubContext.devices and shown in console output.
runner – Optional command runner for executing commands on the device.
spec – The device specification (platform and environment).
provider_type – The provider type identifier for this device.
provider_config – Default ProviderConfig used by all components unless overridden per-component.
provider_config_overrides – Optional per-component overrides keyed by the concrete Component subclass.

property artifact_base_dir: PurePosixPath | None#: Get the remote base directory for artifacts.

property artifact_dir: PurePosixPath | None#: Get the artifact directory for the current run.

connect() → Generator[Self, None, None][source]#

Connect the device, establishing a runner connection if supported.

Yields:: This device instance with an active connection.

create_device_log(name: str) → DeviceLog[source]#

Create a DeviceLog snapshot of this device’s current state.

Parameters:: name – Logical name for this device in the run record.
Returns:: A frozen DeviceLog capturing the current artifact_dir, device_spec, provider_type, provider_config, and downloaded_artifact_dir.

download_artifact_dir(destination: Path) → Path[source]#

Download the current remote artifact directory to a local path.

Parameters:

destination – The local directory to download into.

Returns:

The local path where the artifacts were downloaded.

Raises:

RuntimeError – If not inside a device context.
RuntimeError – If there is no command runner.
RuntimeError – If there is no remote artifact directory.

get_provider_config(config_type: type[_T], component_cls: type[Component]) → _T | None[source]#

Resolve the provider config for a specific component class.

Walks the component_cls MRO looking for a matching override in provider_config_overrides. If none is found, falls back to provider_config. In both cases the config must be an instance of config_type; if it is not, None is returned.

This method is generic on config_type: the return type is T | None where T is the concrete config class passed in, giving both runtime validation and static type narrowing:

cfg = dev.get_provider_config(TrtexecConfig, TensorRTCompiler)
# type: TrtexecConfig | None

Parameters:

config_type – The expected concrete ProviderConfig subclass.
component_cls – The component class requesting the config.

Returns:

The resolved config, or None if no matching config is found.

property is_active: bool#: Whether this device is inside a connect() context.

property last_download_dir: Path | None#: Get the local path of the most recent artifact download.

name: str#

provider_config: ProviderConfig | None = None#

provider_config_overrides: dict[type[Component], ProviderConfig]#

provider_type: str#

runner: CommandRunner | None#

spec: DeviceSpec#

class embedl_hub.core.device.DeviceManager[source]#

Bases: object

Manager for creating device instances.

classmethod get_aws_device(device_name: str, *, name: str = 'main') → Device[source]#

Create a device for the Embedl device cloud (AWS Device Farm).

Parameters:

device_name – The Embedl device cloud device name.
name – Human-readable label for this device.

Returns:

A Device instance configured for AWS Device Farm.

classmethod get_embedl_onnxruntime_device(config: SSHConfig, *, name: str = 'main', provider_config: EmbedlONNXRuntimeConfig = EmbedlONNXRuntimeConfig(embedl_onnxruntime_path=PurePosixPath('embedl-onnxruntime'), cli_args=()), overrides: dict[type[Component], EmbedlONNXRuntimeConfig] | None = None) → Device[source]#

Create an SSH device for ONNX Runtime compilation.

Parameters:

config – SSH connection configuration.
name – Human-readable label for this device.
provider_config – Typed configuration for the embedl-onnxruntime CLI provider.
overrides – Optional per-component config overrides keyed by concrete component class.

Returns:

A Device instance configured for remote ORT compilation.

classmethod get_qai_hub_device(device_name: str, *, name: str = 'main') → Device[source]#

Create a device for Qualcomm AI Hub.

Parameters:

device_name – The QAI Hub device name (e.g., ‘Samsung Galaxy S24’).
name – Human-readable label for this device.

Returns:

A Device instance configured for QAI Hub.

classmethod get_ssh_device(config: SSHConfig, *, name: str = 'main', provider_type: str = ProviderType.EMBEDL_ONNXRUNTIME) → Device[source]#

Create a device with an SSH command runner.

Parameters:

config – SSH connection configuration.
name – Human-readable label for this device.
provider_type – The provider type identifier (e.g. ProviderType.EMBEDL_ONNXRUNTIME, ProviderType.TRTEXEC).

Returns:

A Device instance configured for SSH access.

classmethod get_tensorrt_device(config: SSHConfig, *, name: str = 'main', provider_config: TrtexecConfig = TrtexecConfig(trtexec_path=PurePosixPath('trtexec'), trtexec_cli_args=()), overrides: dict[type[Component], TrtexecConfig] | None = None) → Device[source]#

Create an SSH device for TensorRT compilation.

Parameters:

config – SSH connection configuration.
name – Human-readable label for this device.
provider_config – Typed configuration for the trtexec CLI provider.
overrides – Optional per-component config overrides keyed by concrete component class.

Returns:

A Device instance configured for remote TensorRT compilation.

class embedl_hub.core.device.DeviceSpec(device_name: str | None = None)[source]#

Bases: object

Specification for a target device.

Parameters:: device_name – Optional name identifier for the device (e.g. 'Samsung Galaxy S24').

device_name: str | None = None#

class embedl_hub.core.device.EmbedlONNXRuntimeConfig(embedl_onnxruntime_path: PurePosixPath = PurePosixPath('embedl-onnxruntime'), cli_args: tuple[str, ...]=<factory>)[source]#

Bases: ProviderConfig

Configuration for devices using the embedl-onnxruntime CLI.

Instances are attached to a Device via provider_config (default for all components) and optionally overridden per-component in provider_config_overrides.

Parameters:

embedl_onnxruntime_path – Path to the embedl-onnxruntime executable on the remote system. Defaults to "embedl-onnxruntime" (assumes it is on $PATH).
cli_args – Additional device-specific CLI arguments forwarded verbatim to the embedl-onnxruntime tool.

cli_args: tuple[str, ...]#

embedl_onnxruntime_path: PurePosixPath = PurePosixPath('embedl-onnxruntime')#

class embedl_hub.core.device.ProviderConfig[source]#

Bases: object

Marker base class for typed provider configurations.

Each provider defines a frozen dataclass that inherits from this class. The Device.get_provider_config() method uses isinstance() against a concrete subclass to narrow the type at runtime and satisfy static type-checkers.

class embedl_hub.core.device.SSHCommandRunner(config: SSHConfig)[source]#

Bases: CommandRunner, Connectable

SSH-based command runner for remote device execution.

Implements the CommandRunner and Connectable protocols using asyncssh for SSH connections. Call connect() to obtain a context manager that establishes the connection.

Example:

runner = SSHCommandRunner(config)
with runner.connect():
    runner.run(["ls", "-la"])

connect() → Generator[Self, None, None][source]#

Establish an SSH connection for the duration of the context.

Yields:: This runner instance with an active connection.
Raises:: RuntimeError – If a connection is already active.

device_name: str | None#: Human-readable device name, set by the owning Device.

get(source: PurePosixPath, destination: Path) → None[source]#

Download a file from the remote device.

Parameters:

source – The remote file path to download.
destination – The local destination path.

Raises:

SSHConnectionError – If no connection is established.

property is_active: bool#: Whether an SSH connection is currently established.

put(source: Path, destination: PurePosixPath) → None[source]#

Upload a file to the remote device.

Parameters:

source – The local file path to upload.
destination – The remote destination path.

Raises:

SSHConnectionError – If no connection is established.

run(command: Sequence[str], cwd: PurePosixPath | None = None, *, hide: bool = False) → CommandResult[source]#

Execute a command on the remote device.

Parameters:

command – The command and arguments to execute.
cwd – Optional working directory for command execution.
hide – If True, print only a compact one-liner instead of the full bordered box with streamed output. Use this for short utility commands (e.g. mkdir, rm) that don’t produce meaningful output.

Returns:

The completed process result.

Raises:

SSHConnectionError – If no connection is established.
asyncssh.ProcessError – If the command fails.

class embedl_hub.core.device.SSHConfig(host: str, username: str, port: int = 22, password: str | None = None, private_key_path: Path | None = None, known_hosts: Path | None = None)[source]#

Bases: object

Configuration for SSH connections.

Parameters:

host – The hostname or IP address of the remote device.
port – The SSH port (default: 22).
username – The username for authentication.
password – Optional password for authentication. Using password authentication is discouraged for security reasons and will raise a warning. Prefer key-based authentication instead.
private_key_path – Optional path to private key file. If not provided, asyncssh will attempt to use available authentication methods.
known_hosts – Path to known_hosts file, or None to disable checking.

host: str#

known_hosts: Path | None = None#

password: str | None = None#

port: int = 22#

private_key_path: Path | None = None#

username: str#

exception embedl_hub.core.device.SSHConnectionError[source]#

Bases: Exception

Raised when an SSH operation is attempted without an established connection.

class embedl_hub.core.device.TrtexecConfig(trtexec_path: PurePosixPath = PurePosixPath('trtexec'), trtexec_cli_args: tuple[str, ...]=<factory>)[source]#

Bases: ProviderConfig

Configuration for devices using the trtexec CLI.

Instances are attached to a Device via provider_config (default for all components) and optionally overridden per-component in provider_config_overrides.

Parameters:

trtexec_path – Path to the trtexec executable on the remote system. Defaults to "trtexec" (assumes it is on $PATH).
trtexec_cli_args – Additional device-specific CLI arguments forwarded verbatim to trtexec. These are appended to any component-level arguments.

trtexec_cli_args: tuple[str, ...]#

trtexec_path: PurePosixPath = PurePosixPath('trtexec')#