blueye.sdk.cli

commands

First-party command registry for the blueye CLI.

Every built-in command lives in its own package under blueye/sdk/cli/commands/ and exposes a module-level COMMAND: CommandSpec. Registering a new command means adding it to :func:all_commands — nothing else in the CLI changes.

Invariants command packages must uphold:

• The package (and everything it imports at module level) must be importable with zero optional extras installed; heavy imports (onnx, rich, questionary, ...) belong inside run. This keeps blueye --help working before the [cli] extra is installed.
• add_parser uses only argparse.
• User-facing failures raise :class:blueye.sdk.cli.errors.CliError; main turns them into a clean message and exit code 1.

The CommandSpec contract is also the intended payload for future pip-installable plugins (a blueye.cli entry-point group): an external distribution would expose the same object, and the registry would grow a second discovery source.

Modules:

• bundle_model –

  The blueye bundle-model command: bundle an ONNX model into a BlueyeCV package.

• logs –

  The blueye logs command: list and download dive logs from the drone.

• models –

  The blueye models command: manage the CV models installed on the drone.

• tools –

  The blueye tools command: manage third-party CLI tools.

Classes:

• CommandSpec –

  One first-party blueye subcommand.

Functions:

• all_commands –

  Return every built-in command, in the order shown in blueye --help.

CommandSpec dataclass

CommandSpec(
    name: str,
    help: str,
    requires: tuple[str, ...],
    add_parser: Callable[[_SubParsersAction], None],
    run: Callable[[Namespace], int],
)

One first-party blueye subcommand.

Attributes:

• name (str) –

  The subcommand name (e.g. "bundle-model").

• help (str) –

  One-line description shown in blueye --help.

• requires (tuple[str, ...]) –

  Import names of optional dependencies that must be installed before run executes; main gates on these and prints install guidance.

• add_parser (Callable[[_SubParsersAction], None]) –

  Registers the subcommand's arguments on the root subparsers (argparse only, no optional imports).

• run (Callable[[Namespace], int]) –

  Executes the command and returns the process exit code.

all_commands

all_commands() -> tuple[CommandSpec, ...]

Return every built-in command, in the order shown in blueye --help.

Source code in blueye/sdk/cli/commands/__init__.py

def all_commands() -> tuple[CommandSpec, ...]:
    """Return every built-in command, in the order shown in ``blueye --help``."""
    from .bundle_model import COMMAND as bundle_model_command
    from .logs import COMMAND as logs_command
    from .models import COMMAND as models_command
    from .tools import COMMAND as tools_command

    return (bundle_model_command, logs_command, models_command, tools_command)

meta

model_meta.json construction and validation.

Pure, stdlib-only functions implementing the BlueyeCV model package contract (BlueyeCV/docs/model-meta-spec.md). The generated JSON mirrors the field order used by the reference packages so diffs against hand-written files stay readable.

Classes:

• MetaOptions –

  Everything needed to build a model_meta.json, after prompting/flags.

Functions:

• build_meta –

  Build the model_meta.json dict from the collected options.

• default_preprocessing –

  Return (normalize_scale, normalize_mean, normalize_std) defaults per format.

• validate_meta –

  Check a model_meta dict against the BlueyeCV parser's validation rules.

MetaOptions dataclass

MetaOptions(
    name: str = "",
    version: str = "",
    description: str = "",
    author: str = "",
    license: str = "",
    output_format: str = "",
    kind: str = "detection",
    num_classes: int | None = None,
    labels: list[str] = list(),
    input_width: int | None = None,
    input_height: int | None = None,
    grid_size: int | None = None,
    anchors: list[list[float]] = list(),
    confidence_threshold: float = 0.3,
    nms_threshold: float = 0.45,
    one_indexed_classes: bool = False,
    color_order: str = "rgb",
    normalize_scale: float = SCALE_1_OVER_255,
    normalize_mean: list[float] = list(),
    normalize_std: list[float] = list(),
    template_size: int | None = None,
    search_size: int | None = None,
    sot_overrides: dict[str, object] = dict(),
    tracking_algorithm: str = "none",
    tracking_overrides: dict[str, object] = dict(),
    runtime_device: str | None = None,
    runtime_hz: float | None = None,
    runtime_enabled: bool = False,
)

Everything needed to build a model_meta.json, after prompting/flags.

Attributes mirror the spec's blocks; None/empty means "omit or use the default".

build_meta

build_meta(options: MetaOptions) -> dict

Build the model_meta.json dict from the collected options.

Parameters:

• options (MetaOptions) –

  The fully-resolved options (inference results merged with prompt/flag answers).

Returns:

• dict –

  A JSON-serializable dict in the reference packages' field order.

Source code in blueye/sdk/cli/commands/bundle_model/meta.py

def build_meta(options: MetaOptions) -> dict:
    """Build the model_meta.json dict from the collected options.

    Args:
        options: The fully-resolved options (inference results merged with prompt/flag
            answers).

    Returns:
        A JSON-serializable dict in the reference packages' field order.
    """
    meta: dict = {
        "format_version": 1,
        "model_file": "model.onnx",
        "name": options.name,
    }
    for key in ("version", "description", "author", "license"):
        value = getattr(options, key)
        if value:
            meta[key] = value

    preprocessing: dict = {
        "color_order": options.color_order,
        "normalize_scale": options.normalize_scale,
    }
    if options.normalize_mean:
        preprocessing["normalize_mean"] = options.normalize_mean
    if options.normalize_std:
        preprocessing["normalize_std"] = options.normalize_std
    meta["preprocessing"] = preprocessing

    if options.kind == "sot":
        sot: dict = {"output_format": options.output_format}
        defaults = dict(SOT_DEFAULTS.get(options.output_format, SOT_DEFAULTS["ostrack"]))
        if options.template_size is not None:
            defaults["template_size"] = options.template_size
        if options.search_size is not None:
            defaults["search_size"] = options.search_size
        defaults.update(options.sot_overrides)
        sot.update(defaults)
        meta["sot"] = sot
    else:
        detection: dict = {"output_format": options.output_format}
        if options.output_format == "yolov2_grid":
            detection["anchors"] = options.anchors
            detection["grid_size"] = options.grid_size
        detection["num_classes"] = options.num_classes
        if options.input_width is not None and options.input_height is not None:
            detection["input_width"] = options.input_width
            detection["input_height"] = options.input_height
        detection["confidence_threshold"] = options.confidence_threshold
        detection["nms_threshold"] = options.nms_threshold
        if options.one_indexed_classes:
            detection["one_indexed_classes"] = True
        meta["detection"] = detection

        if options.tracking_algorithm not in ("", "none"):
            tracking: dict = {"algorithm": options.tracking_algorithm}
            tracking.update(TRACKING_DEFAULTS.get(options.tracking_algorithm, {}))
            tracking.update(options.tracking_overrides)
            meta["tracking"] = tracking

    if options.runtime_device is not None or options.runtime_hz is not None:
        runtime: dict = {"enabled": options.runtime_enabled}
        if options.runtime_device is not None:
            runtime["device"] = options.runtime_device
        if options.runtime_hz is not None:
            runtime["hz"] = options.runtime_hz
        meta["runtime"] = runtime

    # Only SOT packages default their label; a detection model without labels must
    # fail validation loudly instead of shipping a nonsense ["tracked"] label list.
    if options.labels:
        meta["labels"] = options.labels
    elif options.kind == "sot":
        meta["labels"] = ["tracked"]
    else:
        meta["labels"] = []
    return meta

default_preprocessing

default_preprocessing(
    output_format: str,
) -> tuple[float, list[float], list[float]]

Return (normalize_scale, normalize_mean, normalize_std) defaults per format.

Matches every reference package: YOLOv2 and TF-SSD models take raw 0-255 pixels, DETR and the SOT trackers use ImageNet normalization, everything else scales to [0, 1].

Source code in blueye/sdk/cli/commands/bundle_model/meta.py

def default_preprocessing(output_format: str) -> tuple[float, list[float], list[float]]:
    """Return (normalize_scale, normalize_mean, normalize_std) defaults per format.

    Matches every reference package: YOLOv2 and TF-SSD models take raw 0-255 pixels,
    DETR and the SOT trackers use ImageNet normalization, everything else scales to
    [0, 1].
    """
    if output_format in ("yolov2_grid", "ssd_multi"):
        return 1.0, [], []
    if output_format in ("detr", "ostrack", "mixformerv2"):
        return SCALE_1_OVER_255, list(IMAGENET_MEAN), list(IMAGENET_STD)
    return SCALE_1_OVER_255, [], []

validate_meta

validate_meta(meta: dict) -> list[str]

Check a model_meta dict against the BlueyeCV parser's validation rules.

Parameters:

• meta (dict) –

  The dict produced by :func:build_meta (or hand-assembled).

Returns:

• list[str] –

  A list of human-readable problems; empty when the metadata is valid.

Source code in blueye/sdk/cli/commands/bundle_model/meta.py

def validate_meta(meta: dict) -> list[str]:
    """Check a model_meta dict against the BlueyeCV parser's validation rules.

    Args:
        meta: The dict produced by :func:`build_meta` (or hand-assembled).

    Returns:
        A list of human-readable problems; empty when the metadata is valid.
    """
    errors: list[str] = []
    if meta.get("format_version") != 1:
        errors.append("format_version must be 1")
    if not meta.get("model_file"):
        errors.append("model_file must be non-empty")
    for key in ("name", "version", "description", "author", "license"):
        if key in meta and not isinstance(meta[key], str):
            errors.append(f"{key} must be a string")

    detection = meta.get("detection")
    sot = meta.get("sot")
    if detection is None and sot is None:
        errors.append("at least one of 'detection' or 'sot' must be present")

    if detection is not None:
        output_format = detection.get("output_format", "")
        if not output_format:
            errors.append("detection.output_format must be non-empty")
        num_classes = detection.get("num_classes", 0)
        if not isinstance(num_classes, int) or num_classes <= 0:
            errors.append("detection.num_classes must be a positive integer")
        labels = meta.get("labels", [])
        if isinstance(num_classes, int) and num_classes > 0 and len(labels) != num_classes:
            errors.append(
                f"labels has {len(labels)} entries but detection.num_classes is {num_classes}"
            )
        if output_format == "yolov2_grid":
            anchors = detection.get("anchors", [])
            if not anchors:
                errors.append("yolov2_grid requires a non-empty detection.anchors list")
            elif any(len(pair) != 2 for pair in anchors):
                errors.append("detection.anchors entries must be [width, height] pairs")
            if not detection.get("grid_size"):
                errors.append("yolov2_grid requires detection.grid_size > 0")
        if output_format in FORMATS_REQUIRING_INPUT_SIZE:
            if not detection.get("input_width") or not detection.get("input_height"):
                errors.append(
                    f"{output_format} requires detection.input_width and " "detection.input_height"
                )

    if sot is not None:
        if not sot.get("output_format"):
            errors.append("sot.output_format must be non-empty")
        if not isinstance(sot.get("template_size"), int) or sot.get("template_size", 0) <= 0:
            errors.append("sot.template_size must be > 0")
        if not isinstance(sot.get("search_size"), int) or sot.get("search_size", 0) <= 0:
            errors.append("sot.search_size must be > 0")

    return errors

heuristics

Model-type inference for the blueye bundle-model CLI.

Pure functions over :class:~blueye.sdk.cli.introspect.ModelInfo-shaped data: no onnx import, no I/O, no terminal — everything here is unit-testable with hand-built dataclasses. The rules mirror what the BlueyeCV output decoders expect (see BlueyeCV/docs/model-meta-spec.md).

Classes:

• DlaAssessment –

  Whether the model is a good fit for the Jetson DLA cores, and why.

• InferredConfig –

  What could be derived from the model, plus how sure we are.

• UnsupportedModelError –

  The model is clearly not usable by BlueyeCV; the message explains why.

Functions:

• assess_dla_fitness –

  Judge whether the model is a good candidate for the Jetson DLA cores.

• infer –

  Infer the BlueyeCV model configuration from the ONNX graph and metadata.

• parse_ultralytics_metadata –

  Parse the metadata_props Ultralytics embeds in its ONNX exports.

DlaAssessment dataclass

DlaAssessment(good_fit: bool, reason: str)

Whether the model is a good fit for the Jetson DLA cores, and why.

InferredConfig dataclass

InferredConfig(
    kind: Literal[
        "detection", "sot", "unknown"
    ] = "unknown",
    output_format: str | None = None,
    confidence: Literal["high", "low"] = "low",
    num_classes: int | None = None,
    input_width: int | None = None,
    input_height: int | None = None,
    grid_size: int | None = None,
    template_size: int | None = None,
    search_size: int | None = None,
    labels: list[str] | None = None,
    suggested_name: str | None = None,
    notes: list[str] = list(),
)

What could be derived from the model, plus how sure we are.

Attributes:

• kind (Literal['detection', 'sot', 'unknown']) –

  "detection", "sot", or "unknown" (format must be chosen manually).

• output_format (str | None) –

  The inferred model_meta output_format, or None.

• confidence (Literal['high', 'low']) –

  "high" when the shape/metadata evidence is unambiguous; "low" when the format should be confirmed with the user.

• num_classes (int | None) –

  Class count derived from the output shape or metadata.

• input_width (int | None) –

  Model input width in pixels (None if dynamic/unknown).

• input_height (int | None) –

  Model input height in pixels (None if dynamic/unknown).

• grid_size (int | None) –

  Feature-map grid size (yolov2_grid only).

• template_size (int | None) –

  SOT template crop size (from the template input).

• search_size (int | None) –

  SOT search crop size (from the search input).

• labels (list[str] | None) –

  Class labels from embedded metadata, index order preserved.

• suggested_name (str | None) –

  Human-readable name suggestion (metadata description or None).

• notes (list[str]) –

  Human-readable reasoning, shown in the inference summary.

UnsupportedModelError

The model is clearly not usable by BlueyeCV; the message explains why.

assess_dla_fitness

assess_dla_fitness(info: ModelInfo) -> DlaAssessment

Judge whether the model is a good candidate for the Jetson DLA cores.

The DLA natively executes convolution-style networks; NMS/TopK and attention/normalization layers fall back to the GPU, negating the benefit. This is a heuristic over the op histogram — TensorRT has the final word when the engine is built on the drone.

Parameters:

• info (ModelInfo) –

  The introspected model.

Returns:

• DlaAssessment –

  A DlaAssessment with the verdict and a one-line reason.

Source code in blueye/sdk/cli/commands/bundle_model/heuristics.py

def assess_dla_fitness(info: ModelInfo) -> DlaAssessment:
    """Judge whether the model is a good candidate for the Jetson DLA cores.

    The DLA natively executes convolution-style networks; NMS/TopK and
    attention/normalization layers fall back to the GPU, negating the benefit. This is
    a heuristic over the op histogram — TensorRT has the final word when the engine is
    built on the drone.

    Args:
        info: The introspected model.

    Returns:
        A DlaAssessment with the verdict and a one-line reason.
    """
    histogram = info.op_histogram
    unfriendly = sorted(op for op in histogram if op in _DLA_UNFRIENDLY_OPS)
    matmul_count = histogram.get("MatMul", 0) + histogram.get("Gemm", 0)
    conv_count = sum(count for op, count in histogram.items() if op in _DLA_FRIENDLY_OPS)

    if unfriendly:
        return DlaAssessment(
            good_fit=False,
            reason=f"contains {', '.join(unfriendly)} layers that fall back to the GPU",
        )
    if matmul_count > max(4, histogram.get("Conv", 0)):
        return DlaAssessment(
            good_fit=False,
            reason=f"MatMul-heavy graph ({matmul_count} MatMul/Gemm nodes) — likely a "
            "transformer, which the DLA cannot accelerate",
        )
    if histogram.get("Conv", 0) == 0:
        return DlaAssessment(
            good_fit=False, reason="no convolution layers found — not a CNN-style model"
        )
    return DlaAssessment(
        good_fit=True,
        reason=f"convolution-dominant graph ({histogram.get('Conv', 0)} Conv, "
        f"{conv_count} DLA-native nodes) with no GPU-fallback layers",
    )

infer

infer(info: ModelInfo) -> InferredConfig

Infer the BlueyeCV model configuration from the ONNX graph and metadata.

Parameters:

• info (ModelInfo) –

  The introspected model.

Returns:

• InferredConfig –

  An InferredConfig; kind == "unknown" means the format must be chosen manually.

Raises:

• UnsupportedModelError –

  When the model clearly cannot run in BlueyeCV (classifier, no image input, float16 input, too many inputs).

Source code in blueye/sdk/cli/commands/bundle_model/heuristics.py

def infer(info: ModelInfo) -> InferredConfig:
    """Infer the BlueyeCV model configuration from the ONNX graph and metadata.

    Args:
        info: The introspected model.

    Returns:
        An InferredConfig; ``kind == "unknown"`` means the format must be chosen
        manually.

    Raises:
        UnsupportedModelError: When the model clearly cannot run in BlueyeCV
            (classifier, no image input, float16 input, too many inputs).
    """
    config = InferredConfig()
    image_inputs = [spec for spec in info.inputs if _is_image_input(spec)]
    _reject_unsupported(info, image_inputs)

    # Input geometry from the (primary) image input.
    height, width = _image_hw(image_inputs[0])
    config.input_height = height
    config.input_width = width

    # Ultralytics metadata beats shape guessing where present.
    ultralytics = parse_ultralytics_metadata(info.metadata)
    if "labels" in ultralytics:
        config.labels = list(ultralytics["labels"])  # type: ignore[arg-type]
        config.notes.append(f"{len(config.labels)} labels from embedded Ultralytics metadata")
    if "imgsz" in ultralytics and (height is None or width is None):
        config.input_height, config.input_width = ultralytics["imgsz"]  # type: ignore[misc]
        config.notes.append("input size from embedded 'imgsz' metadata")
    if "description" in ultralytics:
        config.suggested_name = str(ultralytics["description"])

    if len(image_inputs) >= 2:
        _infer_sot(info, image_inputs, config)
        return config

    outputs = info.outputs
    if len(outputs) == 2:
        _infer_detection_two_outputs(outputs, config)
    elif len(outputs) == 1:
        _infer_detection_single_output(outputs[0], config)
    elif len(outputs) == 4:
        config.kind = "detection"
        config.output_format = "ssd_multi"
        config.confidence = "low"
        config.notes.append("4 outputs (boxes/classes/scores/count) -> SSD-style multi-output")

    # Cross-check the class count against embedded labels.
    if config.labels is not None and config.num_classes is None:
        config.num_classes = len(config.labels)
    if (
        config.labels is not None
        and config.num_classes is not None
        and len(config.labels) != config.num_classes
        and config.kind == "detection"
    ):
        config.notes.append(
            f"warning: {len(config.labels)} embedded labels but the output shape implies "
            f"{config.num_classes} classes"
        )

    if config.output_format is None:
        config.kind = "unknown"
        config.notes.append("output shapes did not match any known decoder format")
    return config

parse_ultralytics_metadata

parse_ultralytics_metadata(
    metadata: dict[str, str],
) -> dict[str, object]

Parse the metadata_props Ultralytics embeds in its ONNX exports.

Parameters:

• metadata (dict[str, str]) –

  The raw metadata_props key/value dict.

Returns:

• dict[str, object] –

  A dict possibly containing "labels" (list[str]), "imgsz" ((height, width)), "task" (str), and "description" (str). Keys are absent when not parseable.

Source code in blueye/sdk/cli/commands/bundle_model/heuristics.py

def parse_ultralytics_metadata(metadata: dict[str, str]) -> dict[str, object]:
    """Parse the metadata_props Ultralytics embeds in its ONNX exports.

    Args:
        metadata: The raw metadata_props key/value dict.

    Returns:
        A dict possibly containing "labels" (list[str]), "imgsz" ((height, width)),
        "task" (str), and "description" (str). Keys are absent when not parseable.
    """
    parsed: dict[str, object] = {}
    names_repr = metadata.get("names")
    if names_repr:
        try:
            names = ast.literal_eval(names_repr)
            if isinstance(names, dict):
                parsed["labels"] = [str(names[key]) for key in sorted(names)]
            elif isinstance(names, (list, tuple)):
                parsed["labels"] = [str(name) for name in names]
        except (ValueError, SyntaxError):
            logger.debug("Could not parse metadata 'names': %r", names_repr)
    imgsz_repr = metadata.get("imgsz")
    if imgsz_repr:
        try:
            imgsz = ast.literal_eval(imgsz_repr)
            if isinstance(imgsz, (list, tuple)) and len(imgsz) == 2:
                parsed["imgsz"] = (int(imgsz[0]), int(imgsz[1]))
        except (ValueError, SyntaxError):
            logger.debug("Could not parse metadata 'imgsz': %r", imgsz_repr)
    if metadata.get("task"):
        parsed["task"] = metadata["task"]
    if metadata.get("description"):
        parsed["description"] = metadata["description"]
    return parsed

introspect

ONNX model introspection for the blueye bundle-model CLI.

This is the only CLI module that imports the onnx package. It is imported lazily by the subcommand, after the dependency gate in main has verified the [cli] extra is installed.

Classes:

• IntrospectionError –

  The file could not be read or is not a valid ONNX model.

• ModelInfo –

  Everything the bundler needs to know about an ONNX model.

• TensorSpec –

  Shape and type of one graph input or output.

Functions:

• check_model –

  Run the strict onnx checker on the model file.

• load_model_info –

  Load an ONNX file and extract the information the bundler needs.

IntrospectionError

The file could not be read or is not a valid ONNX model.

ModelInfo dataclass

ModelInfo(
    path: Path,
    inputs: tuple[TensorSpec, ...] = (),
    outputs: tuple[TensorSpec, ...] = (),
    metadata: dict[str, str] = dict(),
    external_data_files: tuple[str, ...] = (),
    op_histogram: dict[str, int] = dict(),
)

Everything the bundler needs to know about an ONNX model.

Attributes:

• path (Path) –

  Path to the .onnx file.

• inputs (tuple[TensorSpec, ...]) –

  Graph inputs (initializers excluded).

• outputs (tuple[TensorSpec, ...]) –

  Graph outputs.

• metadata (dict[str, str]) –

  The model's metadata_props as a plain dict (Ultralytics exports embed "names", "imgsz", "task", "stride", ... here).

• external_data_files (tuple[str, ...]) –

  Unique file names referenced by tensors stored as external data. These files must live next to the .onnx and travel with it.

• op_histogram (dict[str, int]) –

  Node op_type -> count over the whole graph (used for the DLA fitness assessment).

TensorSpec dataclass

TensorSpec(
    name: str,
    dtype: int,
    dims: tuple[int | str | None, ...],
)

Shape and type of one graph input or output.

Attributes:

• name (str) –

  Tensor name in the graph.

• dtype (int) –

  onnx.TensorProto element type as a plain int (see FLOAT32/FLOAT16).

• dims (tuple[int | str | None, ...]) –

  One entry per dimension: a fixed int, a symbolic name (str, e.g. "batch"), or None when the dimension is completely unknown.

dtype_name property

dtype_name: str

Human-readable element type name (e.g. "FLOAT", "FLOAT16").

shape_str property

shape_str: str

Shape rendered like [1, 3, 640, 640] with ? for unknown dims.

check_model

check_model(path: Path) -> str | None

Run the strict onnx checker on the model file.

Parameters:

• path (Path) –

  Path to the .onnx file. The path form is used so external data is found.

Returns:

• str | None –

  None when the model passes, otherwise the checker's error message. Some perfectly deployable models fail strict checking, so the caller decides whether this is fatal.

Source code in blueye/sdk/cli/commands/bundle_model/introspect.py

def check_model(path: Path) -> str | None:
    """Run the strict onnx checker on the model file.

    Args:
        path: Path to the .onnx file. The path form is used so external data is found.

    Returns:
        None when the model passes, otherwise the checker's error message. Some
        perfectly deployable models fail strict checking, so the caller decides
        whether this is fatal.
    """
    try:
        onnx.checker.check_model(str(path))
    except Exception as error:
        return str(error)
    return None

load_model_info

load_model_info(path: Path) -> ModelInfo

Load an ONNX file and extract the information the bundler needs.

External tensor data is not loaded into memory (the file may be gigabytes); only the referenced file names are recorded.

Parameters:

• path (Path) –

  Path to the .onnx file.

Returns:

• ModelInfo –

  The extracted ModelInfo.

Raises:

• IntrospectionError –

  If the file does not exist or is not a parseable ONNX model.

Source code in blueye/sdk/cli/commands/bundle_model/introspect.py

def load_model_info(path: Path) -> ModelInfo:
    """Load an ONNX file and extract the information the bundler needs.

    External tensor data is not loaded into memory (the file may be gigabytes); only
    the referenced file names are recorded.

    Args:
        path: Path to the .onnx file.

    Returns:
        The extracted ModelInfo.

    Raises:
        IntrospectionError: If the file does not exist or is not a parseable ONNX model.
    """
    if not path.is_file():
        raise IntrospectionError(f"No such file: {path}")
    try:
        model = onnx.load(str(path), load_external_data=False)
    except Exception as error:  # onnx raises DecodeError and various ValueErrors.
        raise IntrospectionError(f"Not a valid ONNX model: {path} ({error})") from error

    initializer_names = {initializer.name for initializer in model.graph.initializer}
    inputs = tuple(
        _tensor_spec(value_info)
        for value_info in model.graph.input
        if value_info.name not in initializer_names
    )
    outputs = tuple(_tensor_spec(value_info) for value_info in model.graph.output)

    metadata = {prop.key: prop.value for prop in model.metadata_props}

    op_histogram: dict[str, int] = {}
    for node in model.graph.node:
        op_histogram[node.op_type] = op_histogram.get(node.op_type, 0) + 1

    return ModelInfo(
        path=path,
        inputs=inputs,
        outputs=outputs,
        metadata=metadata,
        external_data_files=_external_data_files(model),
        op_histogram=op_histogram,
    )

bundle

Zip writing for the blueye bundle-model CLI.

Stdlib-only. The bundle is a flat zip: model.onnx, any external weight files under their exact embedded names, and model_meta.json — all at the archive root, matching how BlueyeCV packages are unpacked onto the drone (unzip -d <package_dir>).

Classes:

• BundleError –

  A user-facing bundling problem (missing files, name collisions, ...).

Functions:

• bundle_size –

  Total input size in bytes (for progress reporting).

• write_bundle –

  Write the model package zip.

BundleError

A user-facing bundling problem (missing files, name collisions, ...).

bundle_size

bundle_size(
    onnx_path: Path, external_files: list[str]
) -> int

Total input size in bytes (for progress reporting).

Parameters:

• onnx_path (Path) –

  Path to the .onnx file.

• external_files (list[str]) –

  External-data file names living next to the .onnx.

Raises:

• BundleError –

  When an external file is missing or otherwise unusable — sizing runs before :func:write_bundle, so it validates too instead of leaking a FileNotFoundError.

Source code in blueye/sdk/cli/commands/bundle_model/bundle.py

def bundle_size(onnx_path: Path, external_files: list[str]) -> int:
    """Total input size in bytes (for progress reporting).

    Args:
        onnx_path: Path to the .onnx file.
        external_files: External-data file names living next to the .onnx.

    Raises:
        BundleError: When an external file is missing or otherwise unusable — sizing
            runs before :func:`write_bundle`, so it validates too instead of leaking a
            FileNotFoundError.
    """
    _validate_external_files(onnx_path, external_files)
    total = onnx_path.stat().st_size
    for name in external_files:
        total += (onnx_path.parent / name).stat().st_size
    return total

write_bundle

write_bundle(
    meta: dict,
    onnx_path: Path,
    external_files: list[str],
    output_path: Path,
    progress: Callable[[int], None] | None = None,
) -> None

Write the model package zip.

The archive is written to <output>.part and atomically renamed on success, so an interrupted run never leaves a truncated zip behind.

Parameters:

• meta (dict) –

  The validated model_meta dict.

• onnx_path (Path) –

  The source .onnx file (stored as model.onnx).

• external_files (list[str]) –

  External-data file names (must exist beside the .onnx; stored under their exact names because the .onnx references them by name).

• output_path (Path) –

  Destination zip path.

• progress (Callable[[int], None] | None, default: None ) –

  Optional callback receiving the number of bytes just written.

Raises:

• BundleError –

  When an external file is missing or collides with a reserved name.

Source code in blueye/sdk/cli/commands/bundle_model/bundle.py

def write_bundle(
    meta: dict,
    onnx_path: Path,
    external_files: list[str],
    output_path: Path,
    progress: Callable[[int], None] | None = None,
) -> None:
    """Write the model package zip.

    The archive is written to ``<output>.part`` and atomically renamed on success, so
    an interrupted run never leaves a truncated zip behind.

    Args:
        meta: The validated model_meta dict.
        onnx_path: The source .onnx file (stored as ``model.onnx``).
        external_files: External-data file names (must exist beside the .onnx; stored
            under their exact names because the .onnx references them by name).
        output_path: Destination zip path.
        progress: Optional callback receiving the number of bytes just written.

    Raises:
        BundleError: When an external file is missing or collides with a reserved name.
    """
    progress = progress or (lambda _byte_count: None)

    _validate_external_files(onnx_path, external_files)

    output_path.parent.mkdir(parents=True, exist_ok=True)
    partial_path = output_path.with_suffix(output_path.suffix + ".part")
    try:
        with zipfile.ZipFile(partial_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
            _copy_into_zip(archive, onnx_path, MODEL_FILE_NAME, progress)
            for name in external_files:
                _copy_into_zip(archive, onnx_path.parent / name, name, progress)
            archive.writestr(META_FILE_NAME, json.dumps(meta, indent=2, ensure_ascii=False) + "\n")
        partial_path.replace(output_path)
    finally:
        partial_path.unlink(missing_ok=True)

metadata

PEP 723 inline-metadata parsing for third-party blueye tools.

A tool script declares itself with the standard PEP 723 block, extended with a [tool.blueye] table::

# /// script
# requires-python = ">=3.10"
# dependencies = ["pandas"]
#
# [tool.blueye]
# name = "export-logs"
# description = "Export dive logs to CSV"
# min-sdk-version = "2.7.0"
# ///

TOML parsing is tiered: :mod:tomllib (Python 3.11+), then :mod:tomli when installed (shipped in the [cli] extra for Python 3.10), then a minimal regex fallback so a stdlib-only Python 3.10 environment can still discover tools. The fallback only understands single-line double-quoted string values inside [tool.blueye] (arrays and multi-line strings need tomli) — enough for the keys the CLI reads.

Classes:

• MetadataError –

  The script's inline metadata is missing or invalid; the message says why.

• ToolMetadata –

  The metadata the CLI reads from a tool script.

Functions:

• extract_script_block –

  Extract the un-commented TOML text of the PEP 723 script block.

• parse_tool_metadata –

  Parse a tool script's source into its ToolMetadata.

MetadataError

The script's inline metadata is missing or invalid; the message says why.

ToolMetadata dataclass

ToolMetadata(
    name: str,
    description: str,
    min_sdk_version: str | None = None,
    has_dependencies: bool = False,
    parsed_with_fallback: bool = False,
)

The metadata the CLI reads from a tool script.

Attributes:

• name (str) –

  The subcommand name the tool is invoked as (blueye <name>).

• description (str) –

  One-line description shown in listings and blueye --help.

• min_sdk_version (str | None) –

  Optional minimum blueye.sdk version; mismatches warn at dispatch time but never block execution.

• has_dependencies (bool) –

  True when the PEP 723 block declares dependencies — execution then prefers uv run so the script gets an isolated environment.

• parsed_with_fallback (bool) –

  True when the regex fallback (not a real TOML parser) produced this metadata.

extract_script_block

extract_script_block(source: str) -> str | None

Extract the un-commented TOML text of the PEP 723 script block.

Parameters:

• source (str) –

  The tool script's full source text.

Returns:

• str | None –

  The TOML text, or None when no script block is present.

Raises:

• MetadataError –

  When more than one script block exists (invalid per PEP 723).

Source code in blueye/sdk/cli/external/metadata.py

def extract_script_block(source: str) -> str | None:
    """Extract the un-commented TOML text of the PEP 723 ``script`` block.

    Args:
        source: The tool script's full source text.

    Returns:
        The TOML text, or None when no ``script`` block is present.

    Raises:
        MetadataError: When more than one ``script`` block exists (invalid per
            PEP 723).
    """
    blocks = [match for match in _PEP723_BLOCK.finditer(source) if match.group("type") == "script"]
    if not blocks:
        return None
    if len(blocks) > 1:
        raise MetadataError("multiple '# /// script' blocks (PEP 723 allows exactly one)")
    content = blocks[0].group("content")
    lines = []
    for line in content.splitlines():
        lines.append(line[2:] if line.startswith("# ") else line[1:])
    return "\n".join(lines) + "\n"

parse_tool_metadata

parse_tool_metadata(source: str) -> ToolMetadata

Parse a tool script's source into its ToolMetadata.

Parameters:

• source (str) –

  The tool script's full source text.

Returns:

• ToolMetadata –

  The parsed metadata.

Raises:

• MetadataError –

  When the block is absent, unparseable, or missing/violating the required [tool.blueye] keys.

Source code in blueye/sdk/cli/external/metadata.py

def parse_tool_metadata(source: str) -> ToolMetadata:
    """Parse a tool script's source into its ToolMetadata.

    Args:
        source: The tool script's full source text.

    Returns:
        The parsed metadata.

    Raises:
        MetadataError: When the block is absent, unparseable, or missing/violating
            the required ``[tool.blueye]`` keys.
    """
    block = extract_script_block(source)
    if block is None:
        raise MetadataError("no '# /// script' metadata block found")

    toml_module = _load_toml()
    if toml_module is not None:
        data = _parse_with_toml(toml_module, block)
        table = data.get("tool", {}).get("blueye", {})
        if not isinstance(table, dict):
            table = {}
        has_dependencies = "dependencies" in data
        used_fallback = False
    else:
        table, has_dependencies = _parse_with_fallback(block)
        used_fallback = True

    if not table:
        raise MetadataError("no [tool.blueye] table in the script block")
    name = table.get("name")
    description = table.get("description")
    if not name or not isinstance(name, str):
        raise MetadataError("[tool.blueye] is missing the required 'name' key")
    if not NAME_PATTERN.fullmatch(name):
        raise MetadataError(
            f"tool name '{name}' is invalid (lowercase letters, digits, and hyphens; "
            "must start with a letter; at most 32 characters)"
        )
    if not description or not isinstance(description, str):
        raise MetadataError("[tool.blueye] is missing the required 'description' key")

    min_sdk_version = table.get("min-sdk-version")
    if min_sdk_version is not None and not isinstance(min_sdk_version, str):
        raise MetadataError("[tool.blueye] 'min-sdk-version' must be a string")

    return ToolMetadata(
        name=name,
        description=description,
        min_sdk_version=min_sdk_version,
        has_dependencies=has_dependencies,
        parsed_with_fallback=used_fallback,
    )

discovery

Tools-directory resolution and third-party tool discovery.

Classes:

• DiscoveredTool –

  One script found in the tools directory.

Functions:

• discover_tools –

  Return the runnable tools, keyed by name (valid metadata, not shadowed).

• format_tools_epilog –

  Build the blueye --help section listing discovered tools (None when empty).

• scan_tools_dir –

  Scan the tools directory and parse every candidate script's metadata.

• tools_dir –

  Resolve the third-party tools directory.

• tools_dir_source –

  Describe where the resolved tools directory came from (for display).

DiscoveredTool dataclass

DiscoveredTool(
    path: Path,
    metadata: ToolMetadata | None = None,
    error: str | None = None,
    shadowed_by: str | None = None,
)

One script found in the tools directory.

Attributes:

• path (Path) –

  The script file.

• metadata (ToolMetadata | None) –

  The parsed metadata, or None when parsing failed.

• error (str | None) –

  The parse-failure reason when metadata is None.

• shadowed_by (str | None) –

  Set when the tool's name is unusable: "built-in command" or the file name of an earlier tool that claimed the same name.

discover_tools

discover_tools(
    builtin_names: frozenset[str] = frozenset(),
) -> dict[str, DiscoveredTool]

Return the runnable tools, keyed by name (valid metadata, not shadowed).

Source code in blueye/sdk/cli/external/discovery.py

def discover_tools(builtin_names: frozenset[str] = frozenset()) -> dict[str, DiscoveredTool]:
    """Return the runnable tools, keyed by name (valid metadata, not shadowed)."""
    return {
        tool.metadata.name: tool
        for tool in scan_tools_dir(builtin_names)
        if tool.metadata is not None and tool.shadowed_by is None
    }

format_tools_epilog

format_tools_epilog(
    tools: dict[str, DiscoveredTool],
) -> str | None

Build the blueye --help section listing discovered tools (None when empty).

Source code in blueye/sdk/cli/external/discovery.py

def format_tools_epilog(tools: dict[str, DiscoveredTool]) -> str | None:
    """Build the `blueye --help` section listing discovered tools (None when empty)."""
    if not tools:
        return None
    width = max(len(name) for name in tools)
    lines = [f"external tools (from {tools_dir()}):"]
    for name in sorted(tools):
        lines.append(f"  {name.ljust(width)}  {tools[name].metadata.description}")
    lines.append("")
    lines.append("Run `blueye tools --help` to manage external tools.")
    return "\n".join(lines)

scan_tools_dir

scan_tools_dir(
    builtin_names: frozenset[str] = frozenset(),
) -> list[DiscoveredTool]

Scan the tools directory and parse every candidate script's metadata.

Returns every *.py file (non-recursive, sorted by file name) as a DiscoveredTool, including invalid and shadowed entries — blueye tools list shows them all. Never raises for a missing or empty directory.

Parameters:

• builtin_names (frozenset[str], default: frozenset() ) –

  First-party command names; tools with a colliding name are marked shadowed (built-ins always win).

Source code in blueye/sdk/cli/external/discovery.py

def scan_tools_dir(builtin_names: frozenset[str] = frozenset()) -> list[DiscoveredTool]:
    """Scan the tools directory and parse every candidate script's metadata.

    Returns every ``*.py`` file (non-recursive, sorted by file name) as a
    DiscoveredTool, including invalid and shadowed entries — `blueye tools list` shows
    them all. Never raises for a missing or empty directory.

    Args:
        builtin_names: First-party command names; tools with a colliding name are
            marked shadowed (built-ins always win).
    """
    directory = tools_dir()
    if not directory.is_dir():
        return []

    tools: list[DiscoveredTool] = []
    claimed: dict[str, str] = {}
    for path in sorted(directory.glob("*.py")):
        if not path.is_file():
            continue
        try:
            parsed = parse_tool_metadata(path.read_text(encoding="utf-8"))
        except (MetadataError, OSError, UnicodeDecodeError) as error:
            logger.debug("Skipping tool %s: %s", path, error)
            tools.append(DiscoveredTool(path=path, error=str(error)))
            continue

        shadowed_by = None
        if parsed.name in builtin_names:
            shadowed_by = "built-in command"
        elif parsed.name in claimed:
            shadowed_by = claimed[parsed.name]
        else:
            claimed[parsed.name] = path.name
        tools.append(DiscoveredTool(path=path, metadata=parsed, shadowed_by=shadowed_by))
    return tools

tools_dir

tools_dir() -> Path

Resolve the third-party tools directory.

Precedence: the :data:TOOLS_DIR_ENV environment variable, then the platform's conventional per-user data location. The directory is not created here — only blueye tools install creates it.

Source code in blueye/sdk/cli/external/discovery.py

def tools_dir() -> Path:
    """Resolve the third-party tools directory.

    Precedence: the :data:`TOOLS_DIR_ENV` environment variable, then the platform's
    conventional per-user data location. The directory is not created here — only
    ``blueye tools install`` creates it.
    """
    override = os.environ.get(TOOLS_DIR_ENV)
    if override:
        return Path(override).expanduser()
    if sys.platform == "darwin":
        base = Path.home() / "Library" / "Application Support"
    elif sys.platform.startswith("win"):
        appdata = os.environ.get("APPDATA")
        base = Path(appdata) if appdata else Path.home() / "AppData" / "Roaming"
    else:
        xdg = os.environ.get("XDG_DATA_HOME")
        base = Path(xdg).expanduser() if xdg else Path.home() / ".local" / "share"
    return base / "blueye" / "cli-tools"

tools_dir_source

tools_dir_source() -> str

Describe where the resolved tools directory came from (for display).

Source code in blueye/sdk/cli/external/discovery.py

def tools_dir_source() -> str:
    """Describe where the resolved tools directory came from (for display)."""
    if os.environ.get(TOOLS_DIR_ENV):
        return f"from {TOOLS_DIR_ENV}"
    return "platform default"

execution

Subprocess execution of third-party tools.

Functions:

• run_tool –

  Run a discovered tool as a subprocess and return its exit code.

run_tool

run_tool(tool: DiscoveredTool, args: list[str]) -> int

Run a discovered tool as a subprocess and return its exit code.

Scripts that declare PEP 723 dependencies are run with uv run when uv is available, giving them an isolated environment with those dependencies; otherwise the current interpreter runs the script directly (its dependencies may already be importable here).

Parameters:

• tool (DiscoveredTool) –

  The tool to run (must have valid metadata).

• args (list[str]) –

  Arguments passed through to the script verbatim.

Source code in blueye/sdk/cli/external/execution.py

def run_tool(tool: DiscoveredTool, args: list[str]) -> int:
    """Run a discovered tool as a subprocess and return its exit code.

    Scripts that declare PEP 723 `dependencies` are run with ``uv run`` when uv is
    available, giving them an isolated environment with those dependencies; otherwise
    the current interpreter runs the script directly (its dependencies may already be
    importable here).

    Args:
        tool: The tool to run (must have valid metadata).
        args: Arguments passed through to the script verbatim.
    """
    _warn_min_sdk_version(tool)

    if tool.metadata.has_dependencies and shutil.which("uv") is not None:
        command = ["uv", "run", str(tool.path), *args]
    else:
        if tool.metadata.has_dependencies:
            logger.debug(
                "Tool %s declares dependencies but uv is not available; running with "
                "the current interpreter.",
                tool.metadata.name,
            )
        command = [sys.executable, str(tool.path), *args]

    logger.debug("Running external tool: %s", command)
    return subprocess.run(command, check=False).returncode