embedl_hub.core.component package#

Component base classes, outputs, and provider types.

Re-exports#

Component — Abstract base class for all components.
CompiledModel — Output type for compilation components.
ComponentOutput — Base output type for all components.
MissingOutputError — Raised on missing output fields.
NoProviderError — Raised when no provider is registered.
ProviderType — Enum of built-in provider identifiers.
RunType — Enum of run types for tracking.

class embedl_hub.core.component.CompiledModel(artifact_dir: Path | None, devices: dict[str, DeviceLog], run_log: RunLog | None, path: LoggedArtifact, input: LoggedArtifact | None = None)[source]#

Bases: ComponentOutput

Base class for compiled model outputs.

Every compiler component produces an output that carries at least:

path — a LoggedArtifact pointing at the compiled model.

Optionally:

input — a LoggedArtifact pointing at the original (pre-compilation) model, or None when the source model is not tracked.

Runtime-specific subclasses may add extra fields (e.g. tensor name mappings).

classmethod from_path(path: Path | PurePosixPath, **extra_kwargs: object) → Self[source]#

Create a compiled model output from a model file path.

Builds a minimal LoggedArtifact for the path field. The input field is left as None and no run_log is attached — the component system will create an IMPORT run automatically when this output is passed to another component.

For local paths the file size is read from disk; for remote paths it defaults to 0.

Any additional keyword arguments are forwarded to the constructor (e.g. input_name_mapping on subclasses).

Parameters:

path – Local or remote path to the compiled model file.
extra_kwargs – Extra fields to pass to the constructor.

Returns:

A new instance of this class.

input: LoggedArtifact | None = None#

path: LoggedArtifact#

class embedl_hub.core.component.Component(**kwargs: object)[source]#

Bases: ABC

Abstract base class for all components.

Subclasses define their interface by implementing run with the desired positional and keyword-only parameters. The body should just be ... — the actual work is performed by device-specific implementations registered via provider().

Each subclass must define a run_type class variable indicating which RunType to use when logging runs.

Each subclass must also define __init__ with the same keyword-only parameters as run, calling super().__init__(...) to store them as defaults. The metaclass validates the parameter names match. Values passed to the constructor serve as defaults that can be overridden at each call site.

Example:

class MyCompiler(Component):
    run_type = RunType.COMPILE

    def __init__(
        self,
        *,
        optimize: bool = True,
    ) -> None:
        """Create a new compiler."""
        super().__init__(optimize=optimize)

    def run(
        self,
        ctx: HubContext,
        model: Path,
        *,
        optimize: bool = True,
    ) -> MyOutput:
        """Compile *model*."""
        ...

@MyCompiler.provider(ProviderType.EMBEDL_ONNXRUNTIME)
def _compile_cli(
    ctx: HubContext,
    model: Path,
    *,
    optimize: bool = True,
) -> MyOutput:
    # *optimize* is already resolved
    ...

compiler = MyCompiler(optimize=False)   # default optimize=False
compiler.run(ctx, some_model)               # uses optimize=False
compiler.run(ctx, some_model, optimize=True) # overrides to True

property name: str#

The display name of this component instance.

Defaults to the class name when not provided at construction time. Used as the run name in the tracking system.

classmethod provider(provider_type: str) → Callable[source]#

Decorator that registers a function as a provider.

Usage:

@MyComponent.provider(ProviderType.EMBEDL_ONNXRUNTIME)
def _run_cli(ctx: HubContext, ...) -> OutputType:
    ...

The decorated function must have the same positional and keyword-only parameters as the component’s run (excluding self). Keyword argument values will be pre-resolved by the component system (call-time overrides beat constructor defaults).

Parameters:: provider_type – A ProviderType member or equivalent string (e.g. ProviderType.EMBEDL_ONNXRUNTIME).
Returns:: A decorator that registers the function and returns it unchanged.

classmethod providers() → list[str][source]#

Return the registered provider type names.

Returns:: A list of provider-type strings.

run_type: ClassVar[RunType]#

class embedl_hub.core.component.ComponentOutput(artifact_dir: Path | None, devices: dict[str, DeviceLog], run_log: RunLog | None)[source]#

Bases: object

Base class for component outputs with typed access to run data.

Subclasses are frozen dataclasses whose fields are annotated as LoggedParam, LoggedMetric, or LoggedArtifact. They can be constructed manually or from a RunLog via from_run_log().

Example:

@dataclass(frozen=True)
class MyOutput(ComponentOutput):
    accuracy: LoggedMetric
    learning_rate: LoggedParam
    model: LoggedArtifact

# Construct manually:
output = MyOutput(
    accuracy=LoggedMetric(...),
    learning_rate=LoggedParam(...),
    model=LoggedArtifact(...),
)

# Or construct from a run log:
output = MyOutput.from_run_log(run_log)
print(output.accuracy.value)

If a field is defined in the class but no matching logged entity with that name exists in the RunLog, a MissingOutputError is raised.

The run_log field holds a reference to the RunLog the output was created from, or None if the output was constructed manually.

artifact_dir: Path | None#

devices: dict[str, DeviceLog]#

download_device_artifacts(ctx: HubContext, device_name: str, target_dir: Path | None = None) → Self[source]#

Download a device’s remote artifact directory.

Uses the artifact_dir recorded in the device’s DeviceLog and the live runner from ctx.devices to transfer the remote directory. The device connection must still be open (i.e. the caller must still be inside with ctx:).

Parameters:

ctx – The active hub context with device connections.
device_name – Name of the device whose artifacts to download.
target_dir – Local directory to download into. If None, defaults to <artifact_dir>/../remote_<device_name>.

Returns:

A new ComponentOutput with the device’s downloaded_artifact_dir updated.

Raises:

RuntimeError – If the device is not found in the output or context, has no remote artifact directory, or has no command runner.

classmethod from_current_run(ctx: HubContext, **extra_kwargs: object) → Self[source]#

Construct an output instance from the current run in the context.

Any additional keyword arguments are forwarded to the constructor for fields that are not tracked in the run log (e.g. raw data dictionaries).

Parameters:

ctx – The execution context with an active run.
extra_kwargs – Extra fields to pass to the constructor.

Returns:

An instance of the output class with fields populated.

Raises:

MissingOutputError – If a required field is not in the run log.
RuntimeError – If no run is currently active.

classmethod from_run_log(run_log: RunLog, **extra_kwargs: object) → Self[source]#

Construct an output instance from a RunLog.

Matches fields by name from the RunLog. Names logged with a $ prefix are matched to fields without the prefix (e.g. $path → path). Raises MissingOutputError if a required field is not found. Raises ValueError if both $name and name exist in the run log.

Fields typed as LoggedX | None (or equivalently Optional[LoggedX]) are optional: if the corresponding entry is missing from the run log the field is set to None instead of raising.

Any additional keyword arguments are forwarded to the constructor for fields that are not tracked in the run log (e.g. raw data dictionaries).

Parameters:

run_log – The RunLog to extract data from.
extra_kwargs – Extra fields to pass to the constructor.

Returns:

An instance of the output class with fields populated.

Raises:

MissingOutputError – If a required field is not in the run log.
ValueError – If both prefixed and unprefixed versions exist.

classmethod from_yaml(yaml_path: Path, **extra_kwargs: object) → Self[source]#

Construct an output instance from a run.yaml file.

This is a convenience wrapper that combines RunLog.from_yaml() and from_run_log() into a single call:

output = CompiledModel.from_yaml(run_dir / "run.yaml")

Any additional keyword arguments are forwarded to the constructor for fields that are not tracked in the run log (e.g. raw data dictionaries).

Parameters:

yaml_path – Path to a run.yaml metadata file.
extra_kwargs – Extra fields to pass to the constructor.

Returns:

An instance of the output class with fields populated.

Raises:

FileNotFoundError – If yaml_path does not exist.
ValueError – If the YAML version is unsupported.
MissingOutputError – If a required field is not in the run log.

run_log: RunLog | None#

exception embedl_hub.core.component.MissingOutputError(field_name: str, field_type: type)[source]#

Bases: Exception

Raised when a required output field is not found in the run log.

exception embedl_hub.core.component.NoProviderError[source]#

Bases: NotImplementedError

Raised when no provider is registered for the active device type.

class embedl_hub.core.component.ProviderType(value)[source]#

Bases: str, Enum

Well-known provider types.

Inherits from str so that values compare naturally with plain strings and can be used as dictionary keys alongside raw strings.

AWS = 'aws'#: Embedl device cloud (AWS Device Farm).

EMBEDL_ONNXRUNTIME = 'embedl_onnxruntime'#: Remote execution via embedl-onnxruntime over SSH.

LOCAL = 'local'#: Local execution — no remote device required.

QAI_HUB = 'qai_hub'#: Qualcomm AI Hub cloud execution.

TRTEXEC = 'trtexec'#: Remote execution via trtexec over SSH.

class embedl_hub.core.component.RunType(value)[source]#

Bases: Enum

COMPILE = 'COMPILE'#

EVAL = 'EVAL'#

GRAPH = 'GRAPH'#

INFERENCE = 'INFERENCE'#

PROFILE = 'PROFILE'#