armnet-client API Reference

armnet_client

armnet client SDK — submit jobs to the platform.

Top-level surface:

• :func:execute — submit one job, block until done.
• :func:execute_many — submit many in parallel, stream results.
• :func:execute_local — run a @main script in the current Python process; bypasses Docker, NATS, and the orchestrator.
• :class:Image — build a Docker image with content-hash caching.
• :class:OrchestratorClient — lower-level HTTP wrapper.

Wire types are re-exported from :mod:armnet_core for ergonomics.

OrchestratorClient

Synchronous client for the orchestrator HTTP API.

Defaults to Armnet's production API and the ARMNET_API_KEY environment variable. Pass base_url or api_key explicitly for development and custom integrations.

Source code in client/src/armnet_client/client.py

class OrchestratorClient:
    """Synchronous client for the orchestrator HTTP API.

    Defaults to Armnet's production API and the ``ARMNET_API_KEY`` environment
    variable. Pass ``base_url`` or ``api_key`` explicitly for development and
    custom integrations.
    """

    def __init__(
        self,
        base_url: Optional[str] = None,
        timeout: float = 30.0,
        api_key: Optional[str] = None,
    ) -> None:
        key = api_key if api_key is not None else default_api_key()
        headers = {API_KEY_HEADER: key} if key else None
        self._client = httpx.Client(
            base_url=(base_url or orchestrator_url()).rstrip("/"),
            timeout=timeout,
            headers=headers,
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "OrchestratorClient":
        return self

    def __exit__(self, *_exc) -> None:
        self.close()

    def submit(self, spec: JobSpec, *, allow_queue: bool = True) -> Job:
        """POST /jobs — create a new job.

        The returned :class:`Job` carries ``dispatched``: True if it was sent to
        a cell immediately, False if it was accepted but queued.

        ``allow_queue`` (default True): when no matching cell is free, queue the
        job (True) or have the orchestrator refuse it with HTTP 409 (False, for
        callers that only make sense running immediately).
        """
        resp = self._client.post(
            "/jobs",
            json=spec.model_dump(mode="json"),
            params={"allow_queue": "true" if allow_queue else "false"},
        )
        _raise_for_status(resp)
        return Job.model_validate(resp.json())

    def get(self, job_id: str) -> Job:
        """GET /jobs/{id} \u2014 fetch the current state of a job."""
        resp = self._client.get(f"/jobs/{job_id}")
        _raise_for_status(resp)
        return Job.model_validate(resp.json())

    def list_jobs(
        self, *, limit: int = 10, queued_only: bool = False
    ) -> list[Job]:
        """GET /jobs — the current user's most-recent jobs (newest first)."""
        resp = self._client.get(
            "/jobs", params={"limit": limit, "queued_only": queued_only}
        )
        _raise_for_status(resp)
        return [Job.model_validate(item) for item in resp.json()]

    def get_status_transitions(self, job_id: str) -> list[JobStatusTransition]:
        """GET /jobs/{id}/status-transitions — a job's status-change audit trail."""
        resp = self._client.get(f"/jobs/{job_id}/status-transitions")
        _raise_for_status(resp)
        return [JobStatusTransition.model_validate(item) for item in resp.json()]

    def stop_job(self, job_id: str, *, reason: str = "client requested stop") -> Job:
        """POST /jobs/{id}/stop \u2014 request an immediate stop of a running job.

        Used when the operator intentionally aborts (e.g. Ctrl-C): this cancels
        the job now rather than waiting out the orchestrator's log-disconnect
        grace window.
        """
        resp = self._client.post(f"/jobs/{job_id}/stop", params={"reason": reason})
        _raise_for_status(resp)
        return Job.model_validate(resp.json())

    def get_cell_status(
        self,
        *,
        embodiment: str = DEFAULT_EMBODIMENT,
        task: Optional[str] = None,
    ) -> CellStatusSummary:
        """GET /cells/status \u2014 fetch current availability for a task."""

        params = {"embodiment": embodiment}
        if task is not None:
            params["task"] = task
        resp = self._client.get("/cells/status", params=params)
        _raise_for_status(resp)
        return CellStatusSummary.model_validate(resp.json())

    def get_registry_credentials(self) -> RegistryCredentials:
        """POST /registry/credentials \u2014 fetch fresh image-registry credentials.

        Used by :meth:`armnet_client.Image.push` to get a short-lived
        OAuth2 token for pushing customer images to the platform's
        Artifact Registry repo. The orchestrator returns 503 if it isn't
        configured for credential issuance (typical for local dev).
        """
        resp = self._client.post("/registry/credentials")
        _raise_for_status(resp)
        return RegistryCredentials.model_validate(resp.json())

    def whoami(self) -> WhoAmI:
        """GET /whoami — fetch the username for the current API key."""

        resp = self._client.get("/whoami")
        _raise_for_status(resp)
        return WhoAmI.model_validate(resp.json())

    def list_secrets(self) -> SecretList:
        """GET /secrets — list secret names for the current user."""

        resp = self._client.get("/secrets")
        _raise_for_status(resp)
        return SecretList.model_validate(resp.json())

    def create_secret(self, name: str, value: str) -> SecretInfo:
        """POST /secrets — create or replace one secret value."""

        resp = self._client.post(
            "/secrets",
            json=SecretCreateRequest(name=name, value=value).model_dump(mode="json"),
        )
        _raise_for_status(resp)
        return SecretInfo.model_validate(resp.json())

    def delete_secret(self, name: str) -> None:
        """DELETE /secrets/{name} — delete one user secret if it exists."""

        resp = self._client.delete(f"/secrets/{name}")
        _raise_for_status(resp)

    def get_volume_credentials(self) -> VolumeCredentials:
        """POST /volume/credentials — fetch short-lived GCS volume credentials."""

        resp = self._client.post("/volume/credentials")
        _raise_for_status(resp)
        return VolumeCredentials.model_validate(resp.json())

    def wait(
        self,
        job_id: str,
        poll_interval_seconds: float = 1.0,
        deadline_seconds: Optional[float] = None,
    ) -> Job:
        """Poll ``GET /jobs/{id}`` until the job reaches a terminal state.

        Returns the final :class:`Job`. Raises :class:`TimeoutError` if
        ``deadline_seconds`` elapses first.
        """
        start = time.monotonic()
        try:
            while True:
                job = self.get(job_id)
                if job.status in TerminalStatus:
                    return job
                if (
                    deadline_seconds is not None
                    and (time.monotonic() - start) >= deadline_seconds
                ):
                    raise TimeoutError(
                        f"Job {job_id} did not reach a terminal state within "
                        f"{deadline_seconds}s (last status: {job.status})."
                    )
                time.sleep(poll_interval_seconds)
        except KeyboardInterrupt:
            # Operator intentionally aborted: stop the job now instead of leaving
            # it to the orchestrator's log-disconnect grace window. Best-effort —
            # always re-raise so the interrupt still propagates.
            try:
                self.stop_job(job_id, reason="client interrupted (KeyboardInterrupt)")
            except Exception:  # noqa: BLE001
                pass
            raise

close

close() -> None

Source code in client/src/armnet_client/client.py

def close(self) -> None:
    self._client.close()

submit

submit(spec: JobSpec, *, allow_queue: bool = True) -> Job

POST /jobs — create a new job.

The returned :class:Job carries dispatched: True if it was sent to a cell immediately, False if it was accepted but queued.

allow_queue (default True): when no matching cell is free, queue the job (True) or have the orchestrator refuse it with HTTP 409 (False, for callers that only make sense running immediately).

Source code in client/src/armnet_client/client.py

def submit(self, spec: JobSpec, *, allow_queue: bool = True) -> Job:
    """POST /jobs — create a new job.

    The returned :class:`Job` carries ``dispatched``: True if it was sent to
    a cell immediately, False if it was accepted but queued.

    ``allow_queue`` (default True): when no matching cell is free, queue the
    job (True) or have the orchestrator refuse it with HTTP 409 (False, for
    callers that only make sense running immediately).
    """
    resp = self._client.post(
        "/jobs",
        json=spec.model_dump(mode="json"),
        params={"allow_queue": "true" if allow_queue else "false"},
    )
    _raise_for_status(resp)
    return Job.model_validate(resp.json())

get

get(job_id: str) -> Job

GET /jobs/{id} — fetch the current state of a job.

Source code in client/src/armnet_client/client.py

def get(self, job_id: str) -> Job:
    """GET /jobs/{id} \u2014 fetch the current state of a job."""
    resp = self._client.get(f"/jobs/{job_id}")
    _raise_for_status(resp)
    return Job.model_validate(resp.json())

list_jobs

list_jobs(*, limit: int = 10, queued_only: bool = False) -> list[Job]

GET /jobs — the current user's most-recent jobs (newest first).

Source code in client/src/armnet_client/client.py

def list_jobs(
    self, *, limit: int = 10, queued_only: bool = False
) -> list[Job]:
    """GET /jobs — the current user's most-recent jobs (newest first)."""
    resp = self._client.get(
        "/jobs", params={"limit": limit, "queued_only": queued_only}
    )
    _raise_for_status(resp)
    return [Job.model_validate(item) for item in resp.json()]

get_status_transitions

get_status_transitions(job_id: str) -> list[JobStatusTransition]

GET /jobs/{id}/status-transitions — a job's status-change audit trail.

Source code in client/src/armnet_client/client.py

def get_status_transitions(self, job_id: str) -> list[JobStatusTransition]:
    """GET /jobs/{id}/status-transitions — a job's status-change audit trail."""
    resp = self._client.get(f"/jobs/{job_id}/status-transitions")
    _raise_for_status(resp)
    return [JobStatusTransition.model_validate(item) for item in resp.json()]

stop_job

stop_job(job_id: str, *, reason: str = 'client requested stop') -> Job

POST /jobs/{id}/stop — request an immediate stop of a running job.

Used when the operator intentionally aborts (e.g. Ctrl-C): this cancels the job now rather than waiting out the orchestrator's log-disconnect grace window.

Source code in client/src/armnet_client/client.py

def stop_job(self, job_id: str, *, reason: str = "client requested stop") -> Job:
    """POST /jobs/{id}/stop \u2014 request an immediate stop of a running job.

    Used when the operator intentionally aborts (e.g. Ctrl-C): this cancels
    the job now rather than waiting out the orchestrator's log-disconnect
    grace window.
    """
    resp = self._client.post(f"/jobs/{job_id}/stop", params={"reason": reason})
    _raise_for_status(resp)
    return Job.model_validate(resp.json())

get_cell_status

get_cell_status(*, embodiment: str = DEFAULT_EMBODIMENT, task: Optional[str] = None) -> CellStatusSummary

GET /cells/status — fetch current availability for a task.

Source code in client/src/armnet_client/client.py

def get_cell_status(
    self,
    *,
    embodiment: str = DEFAULT_EMBODIMENT,
    task: Optional[str] = None,
) -> CellStatusSummary:
    """GET /cells/status \u2014 fetch current availability for a task."""

    params = {"embodiment": embodiment}
    if task is not None:
        params["task"] = task
    resp = self._client.get("/cells/status", params=params)
    _raise_for_status(resp)
    return CellStatusSummary.model_validate(resp.json())

get_registry_credentials

get_registry_credentials() -> RegistryCredentials

POST /registry/credentials — fetch fresh image-registry credentials.

Used by :meth:armnet_client.Image.push to get a short-lived OAuth2 token for pushing customer images to the platform's Artifact Registry repo. The orchestrator returns 503 if it isn't configured for credential issuance (typical for local dev).

Source code in client/src/armnet_client/client.py

def get_registry_credentials(self) -> RegistryCredentials:
    """POST /registry/credentials \u2014 fetch fresh image-registry credentials.

    Used by :meth:`armnet_client.Image.push` to get a short-lived
    OAuth2 token for pushing customer images to the platform's
    Artifact Registry repo. The orchestrator returns 503 if it isn't
    configured for credential issuance (typical for local dev).
    """
    resp = self._client.post("/registry/credentials")
    _raise_for_status(resp)
    return RegistryCredentials.model_validate(resp.json())

whoami

whoami() -> WhoAmI

GET /whoami — fetch the username for the current API key.

Source code in client/src/armnet_client/client.py

def whoami(self) -> WhoAmI:
    """GET /whoami — fetch the username for the current API key."""

    resp = self._client.get("/whoami")
    _raise_for_status(resp)
    return WhoAmI.model_validate(resp.json())

list_secrets

list_secrets() -> SecretList

GET /secrets — list secret names for the current user.

Source code in client/src/armnet_client/client.py

def list_secrets(self) -> SecretList:
    """GET /secrets — list secret names for the current user."""

    resp = self._client.get("/secrets")
    _raise_for_status(resp)
    return SecretList.model_validate(resp.json())

create_secret

create_secret(name: str, value: str) -> SecretInfo

POST /secrets — create or replace one secret value.

Source code in client/src/armnet_client/client.py

def create_secret(self, name: str, value: str) -> SecretInfo:
    """POST /secrets — create or replace one secret value."""

    resp = self._client.post(
        "/secrets",
        json=SecretCreateRequest(name=name, value=value).model_dump(mode="json"),
    )
    _raise_for_status(resp)
    return SecretInfo.model_validate(resp.json())

delete_secret

delete_secret(name: str) -> None

DELETE /secrets/{name} — delete one user secret if it exists.

Source code in client/src/armnet_client/client.py

def delete_secret(self, name: str) -> None:
    """DELETE /secrets/{name} — delete one user secret if it exists."""

    resp = self._client.delete(f"/secrets/{name}")
    _raise_for_status(resp)

get_volume_credentials

get_volume_credentials() -> VolumeCredentials

POST /volume/credentials — fetch short-lived GCS volume credentials.

Source code in client/src/armnet_client/client.py

def get_volume_credentials(self) -> VolumeCredentials:
    """POST /volume/credentials — fetch short-lived GCS volume credentials."""

    resp = self._client.post("/volume/credentials")
    _raise_for_status(resp)
    return VolumeCredentials.model_validate(resp.json())

wait

wait(job_id: str, poll_interval_seconds: float = 1.0, deadline_seconds: Optional[float] = None) -> Job

Poll GET /jobs/{id} until the job reaches a terminal state.

Returns the final :class:Job. Raises :class:TimeoutError if deadline_seconds elapses first.

Source code in client/src/armnet_client/client.py

def wait(
    self,
    job_id: str,
    poll_interval_seconds: float = 1.0,
    deadline_seconds: Optional[float] = None,
) -> Job:
    """Poll ``GET /jobs/{id}`` until the job reaches a terminal state.

    Returns the final :class:`Job`. Raises :class:`TimeoutError` if
    ``deadline_seconds`` elapses first.
    """
    start = time.monotonic()
    try:
        while True:
            job = self.get(job_id)
            if job.status in TerminalStatus:
                return job
            if (
                deadline_seconds is not None
                and (time.monotonic() - start) >= deadline_seconds
            ):
                raise TimeoutError(
                    f"Job {job_id} did not reach a terminal state within "
                    f"{deadline_seconds}s (last status: {job.status})."
                )
            time.sleep(poll_interval_seconds)
    except KeyboardInterrupt:
        # Operator intentionally aborted: stop the job now instead of leaving
        # it to the orchestrator's log-disconnect grace window. Best-effort —
        # always re-raise so the interrupt still propagates.
        try:
            self.stop_job(job_id, reason="client interrupted (KeyboardInterrupt)")
        except Exception:  # noqa: BLE001
            pass
        raise

OrchestratorError

Bases: RuntimeError

Raised when the orchestrator returns a non-2xx response.

status_code is the HTTP status returned by the orchestrator; None if the request never got a response (e.g. connection refused). Callers can branch on it to distinguish e.g. "endpoint not implemented" (404) or "configured-but-unavailable" (503) from real failures.

Source code in client/src/armnet_client/client.py

class OrchestratorError(RuntimeError):
    """Raised when the orchestrator returns a non-2xx response.

    ``status_code`` is the HTTP status returned by the orchestrator;
    ``None`` if the request never got a response (e.g. connection refused).
    Callers can branch on it to distinguish e.g. "endpoint not implemented"
    (404) or "configured-but-unavailable" (503) from real failures.
    """

    def __init__(self, message: str, *, status_code: Optional[int] = None) -> None:
        super().__init__(message)
        self.status_code = status_code

status_code instance-attribute

status_code = status_code

Image dataclass

A reference to a Docker image, either pre-existing or built locally.

Most customer code constructs one with :meth:build. To wrap an existing registry-pushed image, use :meth:from_ref or just pass the image string directly to :func:execute.

Source code in client/src/armnet_client/image.py

@dataclass(frozen=True)
class Image:
    """A reference to a Docker image, either pre-existing or built locally.

    Most customer code constructs one with :meth:`build`. To wrap an
    existing registry-pushed image, use :meth:`from_ref` or just pass the
    image string directly to :func:`execute`.
    """

    ref: str

    def __str__(self) -> str:
        return self.ref

    @classmethod
    def from_ref(cls, ref: str) -> "Image":
        return cls(ref=ref)

    def validate_for_current_user(self, *, if_possible: bool = False) -> "Image":
        """Fast client-side ownership check; the orchestrator remains authoritative.

        In pure-local development the registry endpoint may be unavailable.
        ``if_possible=True`` preserves that workflow while production callers
        still get immediate feedback for another user's or external image.
        """

        try:
            creds = _get_or_fetch_credentials()
        except RegistryCredentialsUnavailableError:
            if if_possible:
                logger.warning(
                    "cannot validate prebuilt image ownership without registry "
                    "credentials; deferring to orchestrator"
                )
                return self
            raise
        namespace_prefix, separator, username = creds.namespace.rpartition("/")
        if not separator:
            raise ImageOwnershipError(
                f"orchestrator returned malformed registry namespace {creds.namespace!r}"
            )
        first = self.ref.split("/", 1)[0]
        explicit_registry = "/" in self.ref and (
            "." in first or ":" in first or first == "localhost"
        )
        if not explicit_registry:
            try:
                return Image(
                    canonical_customer_image(
                        self.ref,
                        registry=creds.registry,
                        namespace_prefix=namespace_prefix,
                        username=username,
                    )
                )
            except ImageReferenceError as exc:
                raise ImageOwnershipError(str(exc)) from exc
        try:
            require_owned_customer_image(
                self.ref,
                registry=creds.registry,
                namespace_prefix=namespace_prefix,
                username=username,
            )
        except ImageReferenceError as exc:
            raise ImageOwnershipError(str(exc)) from exc
        return self

    @classmethod
    def planned_ref(
        cls,
        dockerfile: Union[str, Path],
        name: str,
        context_dir: Union[str, Path, None] = None,
        *,
        platform: Optional[str] = None,
        build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
    ) -> str:
        """The registry ref :meth:`build` then :meth:`push` would produce.

        Worked out from the source alone: the tag is a hash of the build
        context, so the answer needs no Docker daemon and no build. It does
        need the registry credentials, since only they say which namespace the
        image lands in.

        This is what lets a caller ask "is this already pushed?" before
        spending twenty minutes producing a byte-identical image.
        """

        dockerfile_path = Path(dockerfile).resolve()
        if not dockerfile_path.is_file():
            raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")
        context_path = (
            Path(context_dir).resolve() if context_dir is not None else dockerfile_path.parent
        )
        if not context_path.is_dir():
            raise FileNotFoundError(f"build context not found: {context_path}")
        resolved_build_contexts = {
            ctx_name: Path(ctx_path).resolve()
            for ctx_name, ctx_path in (build_contexts or {}).items()
        }
        build_hash = _compute_build_hash(
            dockerfile_path,
            context_path,
            platform=resolve_docker_platform(platform),
            build_contexts=resolved_build_contexts,
        )
        creds = _get_or_fetch_credentials()
        return f"{creds.registry}/{creds.namespace}/{_tag_for(name, build_hash)}"

    @staticmethod
    def exists_in_registry(ref: str) -> bool:
        """Whether ``ref`` is already in the registry.

        Asks the registry's manifest endpoint directly with the same
        short-lived token a push would use, so it costs one request and needs
        no Docker.
        """

        registry, _, repository_and_tag = ref.partition("/")
        repository, _, tag = repository_and_tag.rpartition(":")
        if not registry or not repository or not tag:
            raise ValueError(f"not a registry ref: {ref!r}")
        creds = _get_or_fetch_credentials()
        response = httpx.head(
            f"https://{registry}/v2/{repository}/manifests/{tag}",
            headers={
                "Authorization": f"Bearer {creds.password}",
                "Accept": ", ".join(_MANIFEST_MEDIA_TYPES),
            },
            timeout=_REGISTRY_TIMEOUT_S,
            follow_redirects=True,
        )
        if response.status_code == 404:
            return False
        response.raise_for_status()
        return True

    def push(self, *, if_possible: bool = False) -> "Image":
        """Push this image to the platform's registry; return a new Image
        whose ref points at the registry.

        Flow:

          1. Fetch short-lived OAuth2 credentials from the orchestrator's
             ``POST /registry/credentials`` endpoint (cached on disk;
             refreshed when within 5 minutes of expiry).
          2. ``docker login <registry> -u oauth2accesstoken --password-stdin``.
          3. ``docker tag <local> <registry>/<namespace>/<local>``.
          4. ``docker push`` the new tag.
          5. Return ``Image(ref=<full registry ref>)``.

        Customers don't need any GCP credentials; the orchestrator does
        the IAM dance on their behalf via service-account impersonation.

        For pure-local M0.5 dev where the orchestrator + cell share a
        docker daemon and pushing isn't necessary, pass ``if_possible=True``
        \u2014 the method then logs a warning and returns ``self`` unchanged
        if the orchestrator can't issue credentials, instead of raising.
        Real ``docker push`` failures (network, auth, etc.) always raise.
        """

        _ensure_docker_daemon_available(operation="push images")
        try:
            creds = _get_or_fetch_credentials()
        except RegistryCredentialsUnavailableError as exc:
            if if_possible:
                logger.warning(
                    "Image.push: registry credentials unavailable (%s); "
                    "falling back to local image ref %s",
                    exc,
                    self.ref,
                )
                return self
            raise
        target_ref = _docker_login_tag_push(local_ref=self.ref, creds=creds)
        return Image(ref=target_ref)

    @classmethod
    def build(
        cls,
        dockerfile: Union[str, Path],
        name: str,
        context_dir: Union[str, Path, None] = None,
        force_rebuild: bool = False,
        platform: Optional[str] = None,
        build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
        cache_from: Optional[Iterable[str]] = None,
        cache_to: Optional[Iterable[str]] = None,
    ) -> "Image":
        """Build a Docker image and return a content-addressed reference.

        :param dockerfile: Path to the Dockerfile.
        :param name: Image name (the part before the tag); used as a
            human-readable prefix on the resulting tag.
        :param context_dir: Build context directory. Defaults to the
            Dockerfile's parent.
        :param force_rebuild: If True, always invoke ``docker build`` even
            when an image with the computed content-hash tag already
            exists.
        :param platform: Optional Docker platform, e.g. ``linux/arm64/v8`` or
            ``linux/amd64``. Defaults to ``ARMNET_DOCKER_PLATFORM`` when
            set. Otherwise an x86 host uses Docker's native platform, and
            any other host (Apple Silicon included) builds ``linux/amd64``
            so the image can run on a cell.
        :param build_contexts: Optional named build contexts forwarded to
            ``docker build --build-context name=path``. The Dockerfile can
            then reference them via ``COPY --from=name ...``. Useful for
            pulling in source trees that live outside the primary build
            context (e.g. a sibling checkout on the build host). Contents
            participate in the content hash so changes invalidate the cache.
        :param cache_from: Optional BuildKit ``--cache-from`` specs (e.g.
            ``["type=gha,scope=my-image"]``). When any cache is requested the
            build runs via ``docker buildx build --load`` so layers can be
            reused across machines/CI runs. Falls back to a plain
            ``docker build`` (no cache) if buildx or the cache backend is
            unavailable, so enabling caching can never break a build.
        :param cache_to: Optional BuildKit ``--cache-to`` specs (e.g.
            ``["type=gha,scope=my-image,mode=max"]``).
        :returns: An :class:`Image` whose ``ref`` is
            ``<name>:armnet-<short-hash>``.

        As a convenience for CI, setting ``ARMNET_DOCKER_BUILDX_CACHE=gha`` in
        the environment enables GitHub-Actions layer caching (scoped per image
        ``name``) without passing ``cache_from`` / ``cache_to`` explicitly.
        """

        dockerfile_path = Path(dockerfile).resolve()
        if not dockerfile_path.is_file():
            raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")

        context_path = (
            Path(context_dir).resolve()
            if context_dir is not None
            else dockerfile_path.parent
        )
        if not context_path.is_dir():
            raise FileNotFoundError(f"build context not found: {context_path}")

        resolved_build_contexts: dict[str, Path] = {}
        for ctx_name, ctx_path in (build_contexts or {}).items():
            resolved = Path(ctx_path).resolve()
            if not resolved.is_dir():
                raise FileNotFoundError(
                    f"named build context {ctx_name!r} not found: {resolved}"
                )
            resolved_build_contexts[ctx_name] = resolved

        _ensure_docker_daemon_available(operation="build images")
        platform = resolve_docker_platform(platform)
        build_hash = _compute_build_hash(
            dockerfile_path,
            context_path,
            platform=platform,
            build_contexts=resolved_build_contexts,
        )
        tag = _tag_for(name, build_hash)

        resolved_cache_from, resolved_cache_to = _resolve_buildx_cache(
            name, cache_from, cache_to
        )
        if resolved_cache_from or resolved_cache_to:
            # Layer-cached path (CI). Build with buildx + an external cache
            # backend so expensive layers survive across runs. Uses the ambient
            # Docker config (not an isolated one) so the buildx builder created
            # by e.g. setup-buildx-action and the ACTIONS_* cache tokens are
            # visible. On any failure we fall through to the plain build below,
            # so caching can never turn a working build into a failing one.
            if not force_rebuild and _image_exists_locally(tag):
                logger.info("image %s already built; skipping docker build", tag)
                return cls(ref=tag)
            try:
                _docker_buildx_build(
                    tag=tag,
                    dockerfile=dockerfile_path,
                    context_dir=context_path,
                    platform=platform,
                    build_contexts=resolved_build_contexts,
                    cache_from=resolved_cache_from,
                    cache_to=resolved_cache_to,
                )
                return cls(ref=tag)
            except ImageBuildError as exc:
                logger.warning(
                    "cached buildx build failed (%s); retrying with a plain "
                    "`docker build` (no layer cache)",
                    exc,
                )
        elif platform:
            # Cross-platform builds require buildx; the legacy builder cannot
            # reliably track intermediate layers under QEMU emulation.
            if not force_rebuild and _image_exists_locally(tag):
                logger.info("image %s already built; skipping docker build", tag)
                return cls(ref=tag)
            _docker_buildx_build(
                tag=tag,
                dockerfile=dockerfile_path,
                context_dir=context_path,
                platform=platform,
                build_contexts=resolved_build_contexts,
                cache_from=[],
                cache_to=[],
            )
            return cls(ref=tag)

        with _isolated_docker_config() as docker_env:
            if not force_rebuild and _image_exists_locally(tag, env=docker_env):
                logger.info("image %s already built; skipping docker build", tag)
                return cls(ref=tag)

            cmd = [
                "docker", "build",
                "-t", tag,
                "-f", str(dockerfile_path),
            ]
            if platform:
                cmd.extend(["--platform", platform])
            for ctx_name, ctx_path in resolved_build_contexts.items():
                cmd.extend(["--build-context", f"{ctx_name}={ctx_path}"])
            cmd.append(str(context_path))
            logger.info("docker build %s", " ".join(cmd[2:]))
            try:
                subprocess.run(cmd, check=True, env=docker_env)
            except subprocess.CalledProcessError as exc:
                raise ImageBuildError(
                    f"docker build failed (exit {exc.returncode}) for {tag}"
                ) from exc
            except FileNotFoundError as exc:
                # `docker` binary not on PATH.
                raise ImageBuildError(
                    "`docker` binary not found on PATH; install Docker to build images."
                ) from exc

        return cls(ref=tag)

ref instance-attribute

ref: str

from_ref classmethod

from_ref(ref: str) -> 'Image'

Source code in client/src/armnet_client/image.py

@classmethod
def from_ref(cls, ref: str) -> "Image":
    return cls(ref=ref)

validate_for_current_user

validate_for_current_user(*, if_possible: bool = False) -> 'Image'

Fast client-side ownership check; the orchestrator remains authoritative.

In pure-local development the registry endpoint may be unavailable. if_possible=True preserves that workflow while production callers still get immediate feedback for another user's or external image.

Source code in client/src/armnet_client/image.py

def validate_for_current_user(self, *, if_possible: bool = False) -> "Image":
    """Fast client-side ownership check; the orchestrator remains authoritative.

    In pure-local development the registry endpoint may be unavailable.
    ``if_possible=True`` preserves that workflow while production callers
    still get immediate feedback for another user's or external image.
    """

    try:
        creds = _get_or_fetch_credentials()
    except RegistryCredentialsUnavailableError:
        if if_possible:
            logger.warning(
                "cannot validate prebuilt image ownership without registry "
                "credentials; deferring to orchestrator"
            )
            return self
        raise
    namespace_prefix, separator, username = creds.namespace.rpartition("/")
    if not separator:
        raise ImageOwnershipError(
            f"orchestrator returned malformed registry namespace {creds.namespace!r}"
        )
    first = self.ref.split("/", 1)[0]
    explicit_registry = "/" in self.ref and (
        "." in first or ":" in first or first == "localhost"
    )
    if not explicit_registry:
        try:
            return Image(
                canonical_customer_image(
                    self.ref,
                    registry=creds.registry,
                    namespace_prefix=namespace_prefix,
                    username=username,
                )
            )
        except ImageReferenceError as exc:
            raise ImageOwnershipError(str(exc)) from exc
    try:
        require_owned_customer_image(
            self.ref,
            registry=creds.registry,
            namespace_prefix=namespace_prefix,
            username=username,
        )
    except ImageReferenceError as exc:
        raise ImageOwnershipError(str(exc)) from exc
    return self

planned_ref classmethod

planned_ref(dockerfile: Union[str, Path], name: str, context_dir: Union[str, Path, None] = None, *, platform: Optional[str] = None, build_contexts: Optional[Mapping[str, Union[str, Path]]] = None) -> str

The registry ref :meth:build then :meth:push would produce.

Worked out from the source alone: the tag is a hash of the build context, so the answer needs no Docker daemon and no build. It does need the registry credentials, since only they say which namespace the image lands in.

This is what lets a caller ask "is this already pushed?" before spending twenty minutes producing a byte-identical image.

Source code in client/src/armnet_client/image.py

@classmethod
def planned_ref(
    cls,
    dockerfile: Union[str, Path],
    name: str,
    context_dir: Union[str, Path, None] = None,
    *,
    platform: Optional[str] = None,
    build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
) -> str:
    """The registry ref :meth:`build` then :meth:`push` would produce.

    Worked out from the source alone: the tag is a hash of the build
    context, so the answer needs no Docker daemon and no build. It does
    need the registry credentials, since only they say which namespace the
    image lands in.

    This is what lets a caller ask "is this already pushed?" before
    spending twenty minutes producing a byte-identical image.
    """

    dockerfile_path = Path(dockerfile).resolve()
    if not dockerfile_path.is_file():
        raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")
    context_path = (
        Path(context_dir).resolve() if context_dir is not None else dockerfile_path.parent
    )
    if not context_path.is_dir():
        raise FileNotFoundError(f"build context not found: {context_path}")
    resolved_build_contexts = {
        ctx_name: Path(ctx_path).resolve()
        for ctx_name, ctx_path in (build_contexts or {}).items()
    }
    build_hash = _compute_build_hash(
        dockerfile_path,
        context_path,
        platform=resolve_docker_platform(platform),
        build_contexts=resolved_build_contexts,
    )
    creds = _get_or_fetch_credentials()
    return f"{creds.registry}/{creds.namespace}/{_tag_for(name, build_hash)}"

exists_in_registry staticmethod

exists_in_registry(ref: str) -> bool

Whether ref is already in the registry.

Asks the registry's manifest endpoint directly with the same short-lived token a push would use, so it costs one request and needs no Docker.

Source code in client/src/armnet_client/image.py

@staticmethod
def exists_in_registry(ref: str) -> bool:
    """Whether ``ref`` is already in the registry.

    Asks the registry's manifest endpoint directly with the same
    short-lived token a push would use, so it costs one request and needs
    no Docker.
    """

    registry, _, repository_and_tag = ref.partition("/")
    repository, _, tag = repository_and_tag.rpartition(":")
    if not registry or not repository or not tag:
        raise ValueError(f"not a registry ref: {ref!r}")
    creds = _get_or_fetch_credentials()
    response = httpx.head(
        f"https://{registry}/v2/{repository}/manifests/{tag}",
        headers={
            "Authorization": f"Bearer {creds.password}",
            "Accept": ", ".join(_MANIFEST_MEDIA_TYPES),
        },
        timeout=_REGISTRY_TIMEOUT_S,
        follow_redirects=True,
    )
    if response.status_code == 404:
        return False
    response.raise_for_status()
    return True

push

push(*, if_possible: bool = False) -> 'Image'

Push this image to the platform's registry; return a new Image whose ref points at the registry.

Flow:

1. Fetch short-lived OAuth2 credentials from the orchestrator's POST /registry/credentials endpoint (cached on disk; refreshed when within 5 minutes of expiry).
2. docker login <registry> -u oauth2accesstoken --password-stdin.
3. docker tag <local> <registry>/<namespace>/<local>.
4. docker push the new tag.
5. Return Image(ref=<full registry ref>).

Customers don't need any GCP credentials; the orchestrator does the IAM dance on their behalf via service-account impersonation.

For pure-local M0.5 dev where the orchestrator + cell share a docker daemon and pushing isn't necessary, pass if_possible=True — the method then logs a warning and returns self unchanged if the orchestrator can't issue credentials, instead of raising. Real docker push failures (network, auth, etc.) always raise.

Source code in client/src/armnet_client/image.py

def push(self, *, if_possible: bool = False) -> "Image":
    """Push this image to the platform's registry; return a new Image
    whose ref points at the registry.

    Flow:

      1. Fetch short-lived OAuth2 credentials from the orchestrator's
         ``POST /registry/credentials`` endpoint (cached on disk;
         refreshed when within 5 minutes of expiry).
      2. ``docker login <registry> -u oauth2accesstoken --password-stdin``.
      3. ``docker tag <local> <registry>/<namespace>/<local>``.
      4. ``docker push`` the new tag.
      5. Return ``Image(ref=<full registry ref>)``.

    Customers don't need any GCP credentials; the orchestrator does
    the IAM dance on their behalf via service-account impersonation.

    For pure-local M0.5 dev where the orchestrator + cell share a
    docker daemon and pushing isn't necessary, pass ``if_possible=True``
    \u2014 the method then logs a warning and returns ``self`` unchanged
    if the orchestrator can't issue credentials, instead of raising.
    Real ``docker push`` failures (network, auth, etc.) always raise.
    """

    _ensure_docker_daemon_available(operation="push images")
    try:
        creds = _get_or_fetch_credentials()
    except RegistryCredentialsUnavailableError as exc:
        if if_possible:
            logger.warning(
                "Image.push: registry credentials unavailable (%s); "
                "falling back to local image ref %s",
                exc,
                self.ref,
            )
            return self
        raise
    target_ref = _docker_login_tag_push(local_ref=self.ref, creds=creds)
    return Image(ref=target_ref)

build classmethod

build(dockerfile: Union[str, Path], name: str, context_dir: Union[str, Path, None] = None, force_rebuild: bool = False, platform: Optional[str] = None, build_contexts: Optional[Mapping[str, Union[str, Path]]] = None, cache_from: Optional[Iterable[str]] = None, cache_to: Optional[Iterable[str]] = None) -> 'Image'

Build a Docker image and return a content-addressed reference.

:param dockerfile: Path to the Dockerfile. :param name: Image name (the part before the tag); used as a human-readable prefix on the resulting tag. :param context_dir: Build context directory. Defaults to the Dockerfile's parent. :param force_rebuild: If True, always invoke docker build even when an image with the computed content-hash tag already exists. :param platform: Optional Docker platform, e.g. linux/arm64/v8 or linux/amd64. Defaults to ARMNET_DOCKER_PLATFORM when set. Otherwise an x86 host uses Docker's native platform, and any other host (Apple Silicon included) builds linux/amd64 so the image can run on a cell. :param build_contexts: Optional named build contexts forwarded to docker build --build-context name=path. The Dockerfile can then reference them via COPY --from=name .... Useful for pulling in source trees that live outside the primary build context (e.g. a sibling checkout on the build host). Contents participate in the content hash so changes invalidate the cache. :param cache_from: Optional BuildKit --cache-from specs (e.g. ["type=gha,scope=my-image"]). When any cache is requested the build runs via docker buildx build --load so layers can be reused across machines/CI runs. Falls back to a plain docker build (no cache) if buildx or the cache backend is unavailable, so enabling caching can never break a build. :param cache_to: Optional BuildKit --cache-to specs (e.g. ["type=gha,scope=my-image,mode=max"]). :returns: An :class:Image whose ref is <name>:armnet-<short-hash>.

As a convenience for CI, setting ARMNET_DOCKER_BUILDX_CACHE=gha in the environment enables GitHub-Actions layer caching (scoped per image name) without passing cache_from / cache_to explicitly.

Source code in client/src/armnet_client/image.py

@classmethod
def build(
    cls,
    dockerfile: Union[str, Path],
    name: str,
    context_dir: Union[str, Path, None] = None,
    force_rebuild: bool = False,
    platform: Optional[str] = None,
    build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
    cache_from: Optional[Iterable[str]] = None,
    cache_to: Optional[Iterable[str]] = None,
) -> "Image":
    """Build a Docker image and return a content-addressed reference.

    :param dockerfile: Path to the Dockerfile.
    :param name: Image name (the part before the tag); used as a
        human-readable prefix on the resulting tag.
    :param context_dir: Build context directory. Defaults to the
        Dockerfile's parent.
    :param force_rebuild: If True, always invoke ``docker build`` even
        when an image with the computed content-hash tag already
        exists.
    :param platform: Optional Docker platform, e.g. ``linux/arm64/v8`` or
        ``linux/amd64``. Defaults to ``ARMNET_DOCKER_PLATFORM`` when
        set. Otherwise an x86 host uses Docker's native platform, and
        any other host (Apple Silicon included) builds ``linux/amd64``
        so the image can run on a cell.
    :param build_contexts: Optional named build contexts forwarded to
        ``docker build --build-context name=path``. The Dockerfile can
        then reference them via ``COPY --from=name ...``. Useful for
        pulling in source trees that live outside the primary build
        context (e.g. a sibling checkout on the build host). Contents
        participate in the content hash so changes invalidate the cache.
    :param cache_from: Optional BuildKit ``--cache-from`` specs (e.g.
        ``["type=gha,scope=my-image"]``). When any cache is requested the
        build runs via ``docker buildx build --load`` so layers can be
        reused across machines/CI runs. Falls back to a plain
        ``docker build`` (no cache) if buildx or the cache backend is
        unavailable, so enabling caching can never break a build.
    :param cache_to: Optional BuildKit ``--cache-to`` specs (e.g.
        ``["type=gha,scope=my-image,mode=max"]``).
    :returns: An :class:`Image` whose ``ref`` is
        ``<name>:armnet-<short-hash>``.

    As a convenience for CI, setting ``ARMNET_DOCKER_BUILDX_CACHE=gha`` in
    the environment enables GitHub-Actions layer caching (scoped per image
    ``name``) without passing ``cache_from`` / ``cache_to`` explicitly.
    """

    dockerfile_path = Path(dockerfile).resolve()
    if not dockerfile_path.is_file():
        raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")

    context_path = (
        Path(context_dir).resolve()
        if context_dir is not None
        else dockerfile_path.parent
    )
    if not context_path.is_dir():
        raise FileNotFoundError(f"build context not found: {context_path}")

    resolved_build_contexts: dict[str, Path] = {}
    for ctx_name, ctx_path in (build_contexts or {}).items():
        resolved = Path(ctx_path).resolve()
        if not resolved.is_dir():
            raise FileNotFoundError(
                f"named build context {ctx_name!r} not found: {resolved}"
            )
        resolved_build_contexts[ctx_name] = resolved

    _ensure_docker_daemon_available(operation="build images")
    platform = resolve_docker_platform(platform)
    build_hash = _compute_build_hash(
        dockerfile_path,
        context_path,
        platform=platform,
        build_contexts=resolved_build_contexts,
    )
    tag = _tag_for(name, build_hash)

    resolved_cache_from, resolved_cache_to = _resolve_buildx_cache(
        name, cache_from, cache_to
    )
    if resolved_cache_from or resolved_cache_to:
        # Layer-cached path (CI). Build with buildx + an external cache
        # backend so expensive layers survive across runs. Uses the ambient
        # Docker config (not an isolated one) so the buildx builder created
        # by e.g. setup-buildx-action and the ACTIONS_* cache tokens are
        # visible. On any failure we fall through to the plain build below,
        # so caching can never turn a working build into a failing one.
        if not force_rebuild and _image_exists_locally(tag):
            logger.info("image %s already built; skipping docker build", tag)
            return cls(ref=tag)
        try:
            _docker_buildx_build(
                tag=tag,
                dockerfile=dockerfile_path,
                context_dir=context_path,
                platform=platform,
                build_contexts=resolved_build_contexts,
                cache_from=resolved_cache_from,
                cache_to=resolved_cache_to,
            )
            return cls(ref=tag)
        except ImageBuildError as exc:
            logger.warning(
                "cached buildx build failed (%s); retrying with a plain "
                "`docker build` (no layer cache)",
                exc,
            )
    elif platform:
        # Cross-platform builds require buildx; the legacy builder cannot
        # reliably track intermediate layers under QEMU emulation.
        if not force_rebuild and _image_exists_locally(tag):
            logger.info("image %s already built; skipping docker build", tag)
            return cls(ref=tag)
        _docker_buildx_build(
            tag=tag,
            dockerfile=dockerfile_path,
            context_dir=context_path,
            platform=platform,
            build_contexts=resolved_build_contexts,
            cache_from=[],
            cache_to=[],
        )
        return cls(ref=tag)

    with _isolated_docker_config() as docker_env:
        if not force_rebuild and _image_exists_locally(tag, env=docker_env):
            logger.info("image %s already built; skipping docker build", tag)
            return cls(ref=tag)

        cmd = [
            "docker", "build",
            "-t", tag,
            "-f", str(dockerfile_path),
        ]
        if platform:
            cmd.extend(["--platform", platform])
        for ctx_name, ctx_path in resolved_build_contexts.items():
            cmd.extend(["--build-context", f"{ctx_name}={ctx_path}"])
        cmd.append(str(context_path))
        logger.info("docker build %s", " ".join(cmd[2:]))
        try:
            subprocess.run(cmd, check=True, env=docker_env)
        except subprocess.CalledProcessError as exc:
            raise ImageBuildError(
                f"docker build failed (exit {exc.returncode}) for {tag}"
            ) from exc
        except FileNotFoundError as exc:
            # `docker` binary not on PATH.
            raise ImageBuildError(
                "`docker` binary not found on PATH; install Docker to build images."
            ) from exc

    return cls(ref=tag)

ImageBuildError

Bases: RuntimeError

Raised when docker build fails.

Source code in client/src/armnet_client/image.py

class ImageBuildError(RuntimeError):
    """Raised when ``docker build`` fails."""

ImageOwnershipError

Bases: ValueError

A prebuilt image is outside the authenticated customer's namespace.

Source code in client/src/armnet_client/image.py

class ImageOwnershipError(ValueError):
    """A prebuilt image is outside the authenticated customer's namespace."""

ImagePushError

Bases: RuntimeError

Raised when docker login / docker tag / docker push fails.

Source code in client/src/armnet_client/image.py

class ImagePushError(RuntimeError):
    """Raised when ``docker login`` / ``docker tag`` / ``docker push`` fails."""

RegistryCredentialsUnavailableError

Bases: RuntimeError

Raised when the orchestrator can't issue image-registry credentials.

This is the recoverable shape of an :meth:Image.push failure — the orchestrator either doesn't have the endpoint (404), isn't configured for credential issuance (503), or isn't reachable at all (connection refused, typical for pure-local M0.5 dev). Callers can pass if_possible=True to :meth:Image.push to log + fall back to the local image ref instead of propagating this.

Source code in client/src/armnet_client/image.py

class RegistryCredentialsUnavailableError(RuntimeError):
    """Raised when the orchestrator can't issue image-registry credentials.

    This is the recoverable shape of an :meth:`Image.push` failure \u2014 the
    orchestrator either doesn't have the endpoint (404), isn't configured
    for credential issuance (503), or isn't reachable at all (connection
    refused, typical for pure-local M0.5 dev). Callers can pass
    ``if_possible=True`` to :meth:`Image.push` to log + fall back to the
    local image ref instead of propagating this.
    """

LocalContext dataclass

Developer-facing context for local Docker execution.

This is intentionally smaller than RobotCellConfig. It describes the runtime values a containerized program needs, without exposing operator-only cell settings such as NATS, cell IDs, or completion model configuration.

Source code in client/src/armnet_client/local_container.py

@dataclass
class LocalContext:
    """Developer-facing context for local Docker execution.

    This is intentionally smaller than ``RobotCellConfig``. It describes the
    runtime values a containerized program needs, without exposing operator-only
    cell settings such as NATS, cell IDs, or completion model configuration.
    """

    robot_port: str
    camera_configs: Mapping[str, Any] = field(default_factory=dict)
    robot_id: Optional[str] = None
    calibration_dir: Optional[Union[str, Path]] = None
    calibration_file_path: Optional[Union[str, Path]] = None
    language_instruction: Optional[str] = None
    safety_limit: Optional[float] = None
    docker_gpus: Optional[str] = None

robot_port instance-attribute

robot_port: str

camera_configs class-attribute instance-attribute

camera_configs: Mapping[str, Any] = field(default_factory=dict)

robot_id class-attribute instance-attribute

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute

calibration_dir: Optional[Union[str, Path]] = None

calibration_file_path class-attribute instance-attribute

calibration_file_path: Optional[Union[str, Path]] = None

language_instruction class-attribute instance-attribute

language_instruction: Optional[str] = None

safety_limit class-attribute instance-attribute

safety_limit: Optional[float] = None

docker_gpus class-attribute instance-attribute

docker_gpus: Optional[str] = None

TeleoperatorConfig dataclass

How to connect to the local leader arm being used to teleoperate.

Mirrors what LeRobot needs to construct a teleoperator, plus the sample rate. id is the LeRobot teleoperator id used to find calibration.

Source code in client/src/armnet_client/teleop_stream.py

@dataclass
class TeleoperatorConfig:
    """How to connect to the local leader arm being used to teleoperate.

    Mirrors what LeRobot needs to construct a teleoperator, plus the sample
    rate. ``id`` is the LeRobot teleoperator id used to find calibration.
    """

    port: Optional[str]
    id: Optional[str] = None
    left_port: Optional[str] = None
    right_port: Optional[str] = None
    left_id: Optional[str] = None
    right_id: Optional[str] = None
    embodiment: str = "lerobot/so-101"
    use_degrees: bool = True
    calibration_dir: Optional[str] = None
    # Re-run interactive calibration on connect. Off by default: a leader used
    # for teleop is normally already calibrated.
    calibrate: bool = False
    # Sample rate (Hz). Intentionally higher than the robot loop for snappiness.
    fps: float = _DEFAULT_TELEOP_FPS
    # Named plugin profile supplies the identified hardware defaults. None on
    # an individual field means "inherit from the profile"; explicit values
    # remain available for new GELLO bring-up and diagnostics.
    yam_gravity_profile: str = "trlc_dk1_v1"
    yam_left_gravity_profile: Optional[str] = None
    yam_right_gravity_profile: Optional[str] = None
    yam_gravity_assist: Optional[bool] = None
    yam_gravity_assist_gain: Optional[float] = None
    yam_gravity_current_limit_ma: Optional[int] = None
    yam_gravity_joint_signs: Optional[tuple[int, ...]] = None
    yam_gravity_joint_offsets: Optional[tuple[float, ...]] = None
    # +1 if the GELLO URDF's z axis points physically up when mounted, -1 if
    # down. The bundled URDF is z-down; the wrong value inverts every torque.
    yam_gravity_z_sign: Optional[int] = None
    # Per-link mass override in kg, base to tip (7 values), for leaders built
    # differently from the bundled URDF (e.g. TRLC-DK1). Empty = URDF masses.
    yam_gravity_link_masses: Optional[tuple[float, ...]] = None
    # Restrict assist to these arm joints (empty = all six). Lets validated
    # joints go live while unvalidated ones stay passive.
    yam_gravity_joints: Optional[tuple[str, ...]] = None
    # Per-joint physical ranges in radians, 12 values (lo/hi per joint), for
    # leaders whose joints travel less than the YAM software range (e.g. the
    # DK1 lift). Empty = plugin defaults. Offsets are only valid together with
    # the ranges they were captured with.
    yam_gravity_joint_ranges: Optional[tuple[float, ...]] = None
    # Log computed assist currents without applying torque (safe tuning mode).
    yam_gravity_dry_run: bool = False
    yam_gripper_return: Optional[bool] = None
    yam_gripper_return_current_ma: Optional[int] = None
    yam_assist_side: str = "both"

port instance-attribute

port: Optional[str]

id class-attribute instance-attribute

id: Optional[str] = None

left_port class-attribute instance-attribute

left_port: Optional[str] = None

right_port class-attribute instance-attribute

right_port: Optional[str] = None

left_id class-attribute instance-attribute

left_id: Optional[str] = None

right_id class-attribute instance-attribute

right_id: Optional[str] = None

embodiment class-attribute instance-attribute

embodiment: str = 'lerobot/so-101'

use_degrees class-attribute instance-attribute

use_degrees: bool = True

calibration_dir class-attribute instance-attribute

calibration_dir: Optional[str] = None

calibrate class-attribute instance-attribute

calibrate: bool = False

fps class-attribute instance-attribute

fps: float = _DEFAULT_TELEOP_FPS

yam_gravity_profile class-attribute instance-attribute

yam_gravity_profile: str = 'trlc_dk1_v1'

yam_left_gravity_profile class-attribute instance-attribute

yam_left_gravity_profile: Optional[str] = None

yam_right_gravity_profile class-attribute instance-attribute

yam_right_gravity_profile: Optional[str] = None

yam_gravity_assist class-attribute instance-attribute

yam_gravity_assist: Optional[bool] = None

yam_gravity_assist_gain class-attribute instance-attribute

yam_gravity_assist_gain: Optional[float] = None

yam_gravity_current_limit_ma class-attribute instance-attribute

yam_gravity_current_limit_ma: Optional[int] = None

yam_gravity_joint_signs class-attribute instance-attribute

yam_gravity_joint_signs: Optional[tuple[int, ...]] = None

yam_gravity_joint_offsets class-attribute instance-attribute

yam_gravity_joint_offsets: Optional[tuple[float, ...]] = None

yam_gravity_z_sign class-attribute instance-attribute

yam_gravity_z_sign: Optional[int] = None

yam_gravity_link_masses class-attribute instance-attribute

yam_gravity_link_masses: Optional[tuple[float, ...]] = None

yam_gravity_joints class-attribute instance-attribute

yam_gravity_joints: Optional[tuple[str, ...]] = None

yam_gravity_joint_ranges class-attribute instance-attribute

yam_gravity_joint_ranges: Optional[tuple[float, ...]] = None

yam_gravity_dry_run class-attribute instance-attribute

yam_gravity_dry_run: bool = False

yam_gripper_return class-attribute instance-attribute

yam_gripper_return: Optional[bool] = None

yam_gripper_return_current_ma class-attribute instance-attribute

yam_gripper_return_current_ma: Optional[int] = None

yam_assist_side class-attribute instance-attribute

yam_assist_side: str = 'both'

execute_many

execute_many(specs: Sequence[Mapping[str, Any]], *, max_parallel: int = 4) -> Iterator[JobResult]

Submit many jobs in parallel; yield :class:JobResult as each completes.

specs is a sequence of kwarg-dicts, each accepted by :func:execute (e.g. [{"image": ..., "embodiment": ..., "task": ..., "args": {...}}, ...]). Results are yielded in completion order, not submission order — match them up via your spec's own identifying field if you need to.

max_parallel caps the number of concurrent submissions. Backed by a thread pool; safe to use from synchronous code.

Source code in client/src/armnet_client/execute.py

def execute_many(
    specs: Sequence[Mapping[str, Any]],
    *,
    max_parallel: int = 4,
) -> Iterator[JobResult]:
    """Submit many jobs in parallel; yield :class:`JobResult` as each completes.

    ``specs`` is a sequence of kwarg-dicts, each accepted by
    :func:`execute` (e.g. ``[{"image": ..., "embodiment": ..., "task": ...,
    "args": {...}}, ...]``). Results are yielded in completion order, not
    submission order \u2014 match them up via your spec's own identifying
    field if you need to.

    ``max_parallel`` caps the number of concurrent submissions. Backed by
    a thread pool; safe to use from synchronous code.
    """

    if not specs:
        return

    with ThreadPoolExecutor(max_workers=max_parallel) as pool:
        futures = [pool.submit(execute, **spec) for spec in specs]
        for fut in as_completed(futures):
            yield fut.result()

execute_local

execute_local(main_module: Union[str, Path], *, args: Optional[Mapping[str, Any]] = None, port: Optional[str] = None, robot_id: Optional[str] = None, calibration_dir: Optional[Union[str, Path]] = None, calibration_file_path: Optional[Union[str, Path]] = None, camera_configs: Optional[Mapping[str, Any]] = None, embodiment: str = DEFAULT_EMBODIMENT, task: str = DEFAULT_TASK, timeout_seconds: int = 120, job_id: Optional[str] = None) -> JobResult

Run a @main-decorated script in the current Python process.

:param main_module: Path to the Python file containing exactly one @main-decorated function. :param args: Mapping passed through as ctx.args. :param port: Value placed at ctx.cell.robot_port — typically the path to a real local serial device, e.g. "/dev/ttyUSB0". :param robot_id: Value placed at ctx.cell.robot_id — typically the name of the robot, e.g. "praveen_so101". :param embodiment: Recorded on ctx.embodiment (defaults to the only embodiment supported in M0.5). :param task: Recorded on ctx.task (defaults to the only task supported in M0.5). :param timeout_seconds: Recorded on ctx.timeout_seconds. Not enforced by execute_local (see module docstring). :param job_id: Optional explicit job id for ctx.job_id. Defaults to a stable synthetic id derived from the script path. :returns: A :class:JobResult with return_value set to whatever @main returned, or error populated with a traceback if it raised.

Source code in client/src/armnet_client/local.py

def execute_local(
    main_module: Union[str, Path],
    *,
    args: Optional[Mapping[str, Any]] = None,
    port: Optional[str] = None,
    robot_id: Optional[str] = None,
    calibration_dir: Optional[Union[str, Path]] = None,
    calibration_file_path: Optional[Union[str, Path]] = None,
    camera_configs: Optional[Mapping[str, Any]] = None,
    embodiment: str = DEFAULT_EMBODIMENT,
    task: str = DEFAULT_TASK,
    timeout_seconds: int = 120,
    job_id: Optional[str] = None,
) -> JobResult:
    """Run a ``@main``-decorated script in the current Python process.

    :param main_module: Path to the Python file containing exactly one
        ``@main``-decorated function.
    :param args: Mapping passed through as ``ctx.args``.
    :param port: Value placed at ``ctx.cell.robot_port`` \u2014 typically the path
        to a real local serial device, e.g. ``"/dev/ttyUSB0"``.
    :param robot_id: Value placed at ``ctx.cell.robot_id`` \u2014 typically the
        name of the robot, e.g. ``"praveen_so101"``.
    :param embodiment: Recorded on ``ctx.embodiment`` (defaults to the
        only embodiment supported in M0.5).
    :param task: Recorded on ``ctx.task`` (defaults to the only task
        supported in M0.5).
    :param timeout_seconds: Recorded on ``ctx.timeout_seconds``. Not
        enforced by ``execute_local`` (see module docstring).
    :param job_id: Optional explicit job id for ``ctx.job_id``. Defaults
        to a stable synthetic id derived from the script path.
    :returns: A :class:`JobResult` with ``return_value`` set to whatever
        ``@main`` returned, or ``error`` populated with a traceback if it
        raised.
    """

    Cell, Context, _build_camera_configs, _reset_registry, registered_main = _import_runtime()

    script_path = Path(main_module).resolve()
    if not script_path.is_file():
        raise FileNotFoundError(f"main_module not found: {script_path}")

    embodiment_e = embodiment
    task_e = task
    args_dict = dict(args) if args is not None else {}
    job_id_ = job_id or f"job_local_{script_path.stem}"

    started_at = _utcnow()

    _reset_registry()  # allow successive execute_local calls in one process
    try:
        _load_script(script_path)
    except Exception:
        # Import failure \u2014 there's a Python traceback (think syntax error in
        # the user's script). Mirror what the cell does: stash the traceback
        # in the dedicated field so result.raise_for_status() etc. behave
        # consistently with the remote `execute()` path.
        return JobResult(
            status=JobStatus.FAILED,
            error="failed to import main_module",
            traceback=traceback.format_exc(),
            started_at=started_at,
            finished_at=_utcnow(),
        )

    fn = registered_main()
    if fn is None:
        # Contract violation, not a user-code exception \u2014 no traceback to
        # surface; this stays in `error`.
        return JobResult(
            status=JobStatus.FAILED,
            error=(
                f"no @main-decorated function found in {script_path}. "
                "Decorate exactly one function with `@main` to mark it as the entry point."
            ),
            started_at=started_at,
            finished_at=_utcnow(),
        )

    ctx = Context(
        job_id=job_id_,
        embodiment=embodiment_e,
        task=task_e,
        args=args_dict,
        cell=Cell(
            robot_port=port,
            robot_id=robot_id,
            calibration_dir=Path(calibration_dir) if calibration_dir else None,
            calibration_file_path=Path(calibration_file_path) if calibration_file_path else None,
        ),
        camera_configs=_build_camera_configs(dict(camera_configs or {})),
        timeout_seconds=timeout_seconds,
    )

    try:
        return_value = fn(ctx)
    except Exception:
        return JobResult(
            status=JobStatus.FAILED,
            traceback=traceback.format_exc(),
            started_at=started_at,
            finished_at=_utcnow(),
        )

    return JobResult(
        status=JobStatus.SUCCEEDED,
        exit_code=0,
        return_value=return_value,
        started_at=started_at,
        finished_at=_utcnow(),
    )

execute_local_container

execute_local_container(*, image: Union[str, Image], local_context: Optional[LocalContext] = None, args: Optional[Mapping[str, Any]] = None, embodiment: str = DEFAULT_EMBODIMENT, task: str = DEFAULT_TASK, timeout_seconds: int = 120, cell_socket: Optional[str] = None, cell_endpoint: Optional[str] = None, robot_id: Optional[str] = None, calibration_dir: Optional[Union[str, Path]] = None, calibration_file_path: Optional[Union[str, Path]] = None, cell_config: Optional[RobotCellConfig] = None, cell_config_path: Optional[Union[str, Path]] = None, mount_calibration: bool = True, username: str = 'remoterobo-test', job_id: str = 'job_local_container', stream_output: bool = True, docker_gpus: Optional[str] = None, docker_runtime: Optional[str] = None, gpus: Optional[str] = None) -> JobResult

Run a runtime image locally through Docker.

This is the M2 local-dev path for robot examples: it validates the same container + armnet-runtime + Unix-socket route as remote execution, without going through NATS or the cloud orchestrator.

Source code in client/src/armnet_client/local_container.py

def execute_local_container(
    *,
    image: Union[str, Image],
    local_context: Optional[LocalContext] = None,
    args: Optional[Mapping[str, Any]] = None,
    embodiment: str = DEFAULT_EMBODIMENT,
    task: str = DEFAULT_TASK,
    timeout_seconds: int = 120,
    cell_socket: Optional[str] = None,
    cell_endpoint: Optional[str] = None,
    robot_id: Optional[str] = None,
    calibration_dir: Optional[Union[str, Path]] = None,
    calibration_file_path: Optional[Union[str, Path]] = None,
    cell_config: Optional[RobotCellConfig] = None,
    cell_config_path: Optional[Union[str, Path]] = None,
    mount_calibration: bool = True,
    username: str = "remoterobo-test",
    job_id: str = "job_local_container",
    stream_output: bool = True,
    docker_gpus: Optional[str] = None,
    docker_runtime: Optional[str] = None,
    gpus: Optional[str] = None,
) -> JobResult:
    """Run a runtime image locally through Docker.

    This is the M2 local-dev path for robot examples: it validates the same
    container + armnet-runtime + Unix-socket route as remote execution,
    without going through NATS or the cloud orchestrator.
    """

    started_at = _utcnow()
    embodiment_e = embodiment
    task_e = task
    image_ref = str(image) if isinstance(image, Image) else image
    loaded_cell_config = (
        _cell_config_from_local_context(local_context, embodiment=embodiment_e, task=task_e)
        if local_context is not None
        else _build_cell_config(
            cell_config=cell_config,
            cell_config_path=cell_config_path,
            cell_endpoint=cell_endpoint,
            cell_socket=cell_socket,
            robot_id=robot_id,
            calibration_dir=calibration_dir,
            calibration_file_path=calibration_file_path,
            embodiment=embodiment_e,
            task=task_e,
        )
    )
    docker_gpus = docker_gpus if docker_gpus is not None else (
        local_context.docker_gpus if local_context is not None else None
    )

    with _LocalControlServer(
        robot_port=None if local_context is not None else loaded_cell_config.robot_connector_endpoint,
        use_host_network=local_context is not None,
    ) as control:
        cmd = [
            "docker", "run", "--rm",
            "-e", f"{runtime_env.JOB_ID}={job_id}",
            "-e", f"{runtime_env.EMBODIMENT}={embodiment_e}",
            "-e", f"{runtime_env.TASK}={task_e}",
            "-e", f"{runtime_env.TIMEOUT_SECONDS}={timeout_seconds}",
            "-e", f"{runtime_env.ARGS}={json.dumps(dict(args or {}))}",
            "-e", f"{runtime_env.LOCAL_CONTROL_ENDPOINT}={control.container_endpoint}",
            "-e", f"{runtime_env.LOCAL_CONTAINER}=1",
        ]
        if local_context is not None:
            cmd.extend(["--network", "host"])
            cmd.extend(_direct_hardware_device_args(local_context))
        else:
            cmd.extend(["--add-host", "host.docker.internal:host-gateway"])
        if docker_gpus:
            cmd.extend(_gpu_args(docker_gpus))
        elif docker_runtime:
            cmd.extend(["--runtime", docker_runtime])
            if gpus:
                cmd.extend(["--gpus", gpus])
        cell_config_json = loaded_cell_config.model_dump_json(exclude_none=True)
        cmd.extend(["-e", f"{runtime_env.CELL_CONFIG}={cell_config_json}"])
        endpoint = None if local_context is not None else loaded_cell_config.robot_connector_endpoint
        if endpoint:
            cmd.extend(["-e", f"{runtime_env.CELL_SOCKET}={endpoint}"])
            if not endpoint.startswith("tcp://"):
                socket_path = Path(endpoint)
                socket_dir = str(socket_path.parent)
                cmd.extend([
                    # Mount the parent directory, not the socket file itself.
                    "-v", f"{socket_dir}:{socket_dir}",
                ])
        if loaded_cell_config.robot_id:
            cmd.extend(["-e", f"{runtime_env.ROBOT_ID}={loaded_cell_config.robot_id}"])
        if loaded_cell_config.calibration_dir:
            cmd.extend(["-e", f"{runtime_env.CALIBRATION_DIR}={loaded_cell_config.calibration_dir}"])
        if loaded_cell_config.calibration_file_path:
            cmd.extend([
                "-e",
                f"{runtime_env.CALIBRATION_FILE_PATH}={loaded_cell_config.calibration_file_path}",
            ])
        calibration_mount_path = _cell_calibration_mount_root(loaded_cell_config)
        if mount_calibration and calibration_mount_path:
            cmd.extend(["-v", f"{calibration_mount_path}:{calibration_mount_path}:ro"])
        cache_home = _user_cache_home(username)
        volume_home = _user_volume_home(username)
        cmd.extend([
            "-e", f"{runtime_env.CACHE_HOME}={cache_home}",
            "-e", f"{runtime_env.VOLUME_HOME}={volume_home}",
            # Also set the pre-rename spellings for containers pinned to an
            # older armnet-runtime that still reads them.
            "-e", f"{runtime_env.LEGACY_CACHE_HOME}={cache_home}",
            "-e", f"{runtime_env.LEGACY_VOLUME_HOME}={volume_home}",
            "-e", f"{runtime_env.HF_HOME}={cache_home / 'huggingface'}",
            "-v", f"{cache_home}:{cache_home}",
            "-v", f"{volume_home}:{volume_home}",
        ])
        # Remote-call profiling is off unless the operator asks for it, and the
        # container inherits nothing from this shell, so forward it explicitly.
        call_stats_interval = os.environ.get("ARMNET_REMOTE_CALL_STATS_INTERVAL_S")
        if call_stats_interval:
            cmd.extend([
                "-e",
                f"ARMNET_REMOTE_CALL_STATS_INTERVAL_S={call_stats_interval}",
            ])
        cmd.append(image_ref)

        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
        )
        assert proc.stdout is not None
        assert proc.stderr is not None

        stdout_q: Queue[str] = Queue()
        stderr_q: Queue[str] = Queue()
        stdout_thread = threading.Thread(
            target=_pump_stream,
            args=(proc.stdout, stdout_q, sys.stdout if stream_output else None),
            daemon=True,
        )
        stderr_thread = threading.Thread(
            target=_pump_stream,
            args=(proc.stderr, stderr_q, sys.stderr if stream_output else None),
            daemon=True,
        )
        stdout_thread.start()
        stderr_thread.start()

        try:
            proc.wait(timeout=timeout_seconds)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()
            stdout_thread.join(timeout=1)
            stderr_thread.join(timeout=1)
            stdout = _queue_to_text(stdout_q)
            stderr = _queue_to_text(stderr_q)
            return JobResult(
                status=JobStatus.TIMEOUT,
                error=f"local container exceeded timeout_seconds={timeout_seconds}",
                stdout=stdout,
                stderr=stderr,
                started_at=started_at,
                finished_at=_utcnow(),
            )

        stdout_thread.join(timeout=1)
        stderr_thread.join(timeout=1)
        stdout = _queue_to_text(stdout_q)
        stderr = _queue_to_text(stderr_q)
        return JobResult(
            status=JobStatus.SUCCEEDED if proc.returncode == 0 else JobStatus.FAILED,
            exit_code=proc.returncode,
            stdout=stdout,
            stderr=stderr,
            return_value=_extract_marker(stdout, RESULT_MARKER_JSON),
            traceback=_extract_marker(stdout, TRACEBACK_MARKER_JSON),
            started_at=started_at,
            finished_at=_utcnow(),
        )

init_rerun

init_rerun(application_id: str = 'armnet', *, spawn: bool = True, serve_web: bool = False, web_port: Optional[int] = None, grpc_port: Optional[int] = None, open_browser: bool = True, flush_tick_seconds: Optional[float] = None) -> Any

Initialize Rerun and start a viewer.

Thin convenience wrapper so orchestrate scripts don't need to import the Rerun SDK directly. Raises a clear error if rerun-sdk isn't installed (pip install 'armnet-client[viz]').

Viewer selection:

• serve_web=True always hosts the web viewer over HTTP (open the printed URL in a browser; forward the port first if you're on a remote host).
• spawn=True (default) opens a native desktop window — but only when a display is available. On a headless machine (no DISPLAY / Wayland, e.g. SSH or a GUI-less workstation) it automatically falls back to the web viewer instead of failing to open a window.

flush_tick_seconds sets Rerun's micro-batching flush interval (the RERUN_FLUSH_TICK_SECS knob). A very small value (e.g. 0.002 = 2ms) minimizes how long logged data waits before being sent to the viewer — noticeably snappier for teleoperation. Left unset, Rerun's default applies.

Source code in client/src/armnet_client/rerun_stream.py

def init_rerun(
    application_id: str = "armnet",
    *,
    spawn: bool = True,
    serve_web: bool = False,
    web_port: Optional[int] = None,
    grpc_port: Optional[int] = None,
    open_browser: bool = True,
    flush_tick_seconds: Optional[float] = None,
) -> Any:
    """Initialize Rerun and start a viewer.

    Thin convenience wrapper so orchestrate scripts don't need to import the
    Rerun SDK directly. Raises a clear error if ``rerun-sdk`` isn't installed
    (``pip install 'armnet-client[viz]'``).

    Viewer selection:

    - ``serve_web=True`` always hosts the **web viewer** over HTTP (open the
      printed URL in a browser; forward the port first if you're on a remote
      host).
    - ``spawn=True`` (default) opens a native desktop window — but only when a
      display is available. On a **headless** machine (no ``DISPLAY`` /
      Wayland, e.g. SSH or a GUI-less workstation) it automatically falls back
      to the web viewer instead of failing to open a window.

    ``flush_tick_seconds`` sets Rerun's micro-batching flush interval (the
    ``RERUN_FLUSH_TICK_SECS`` knob). A very small value (e.g. ``0.002`` = 2ms)
    minimizes how long logged data waits before being sent to the viewer —
    noticeably snappier for teleoperation. Left unset, Rerun's default applies.
    """

    try:
        import rerun as rr
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError(
            "rerun-sdk is required for visualization; install with "
            "`pip install 'armnet-client[viz]'`"
        ) from exc

    if flush_tick_seconds is not None:
        from datetime import timedelta

        # make_default=True routes the module-level rr.log/serve/spawn calls
        # (and the replay in log_packet) through this recording, so the small
        # flush tick actually applies to the streamed data.
        rr.RecordingStream(
            application_id,
            make_default=True,
            batcher_config=rr.ChunkBatcherConfig(
                flush_tick=timedelta(seconds=flush_tick_seconds)
            ),
        )
    else:
        rr.init(application_id)

    use_web = serve_web or (spawn and not _has_display())
    if use_web:
        if spawn and not serve_web:
            logger.warning(
                "no display detected; serving the Rerun web viewer instead of "
                "spawning a native window"
            )
        web = web_port or _DEFAULT_WEB_VIEWER_PORT
        # serve_grpc starts the data server (the log sink + buffer) and returns
        # its URI; serve_web_viewer hosts the HTML viewer that connects to it.
        # (rr.serve_web bundles both but is deprecated, and the viewer/data live
        # on separate ports either way.)
        server_uri = rr.serve_grpc(grpc_port=grpc_port)
        rr.serve_web_viewer(
            web_port=web,
            open_browser=open_browser and _has_display(),
            connect_to=server_uri,
        )
        grpc = _port_from_uri(server_uri) or grpc_port or _DEFAULT_GRPC_PORT
        encoded_uri = server_uri.replace("+", "%2B")
        viewer_url = f"http://localhost:{web}/?url={encoded_uri}"
        print(
            "[armnet] Rerun viewer ready:\n"
            f"  web viewer:  {viewer_url}\n"
            f"  or native:   rerun --connect {server_uri}\n"
            "  on a remote host, forward BOTH ports first (the browser viewer "
            "loads from the web port but streams data from the gRPC port):\n"
            f"      ssh -L {web}:localhost:{web} -L {grpc}:localhost:{grpc} <user>@<host>",
            flush=True,
        )
    elif spawn:
        rr.spawn()
    return rr

start_recording_key_listener

start_recording_key_listener(events: 'queue.Queue[str]')

Bind LeRobot's dataset-recording keyboard shortcuts to events.

Right Arrow → next_episode, Left Arrow → rerecord_episode, Esc → stop_recording. Each press is enqueued for :func:stream_job_teleop to forward to the cell. Esc does not tear anything down client-side: the runtime ends the session on receipt and the job result unblocks the orchestrate script, which then stops its streams.

Linux reads the kernel input device via evdev, because pynput only hears keys through X11 and a Wayland session never delivers them. macOS keeps pynput. Returns the started listener (call .stop() when done), or None when no keyboard is available.

Source code in client/src/armnet_client/teleop_stream.py

def start_recording_key_listener(events: "queue.Queue[str]"):
    """Bind LeRobot's dataset-recording keyboard shortcuts to ``events``.

    Right Arrow → ``next_episode``, Left Arrow → ``rerecord_episode``, Esc →
    ``stop_recording``. Each press is enqueued for :func:`stream_job_teleop`
    to forward to the cell. Esc does not tear anything down client-side: the
    runtime ends the session on receipt and the job result unblocks the
    orchestrate script, which then stops its streams.

    Linux reads the kernel input device via ``evdev``, because ``pynput`` only
    hears keys through X11 and a Wayland session never delivers them. macOS
    keeps ``pynput``. Returns the started listener (call ``.stop()`` when
    done), or None when no keyboard is available.
    """

    if sys.platform == "linux":
        return _start_evdev_key_listener(events)
    return _start_pynput_key_listener(events)

stream_job_teleop

stream_job_teleop(job_id: str, stop: Event, config: TeleoperatorConfig, events: 'queue.Queue[str] | None' = None) -> None

Sample the local leader and push actions to job_id until stop is set.

Intended to run on a background thread (alongside the rerun stream). Returns quietly if no API key is configured, LeRobot is missing, or the leader/ websocket can't be reached, so teleop never crashes the orchestrate script.

events, when given, is a queue of recording-control event strings (see :func:start_recording_key_listener); each is sent on the same websocket as a :class:TeleopEvent frame. Unlike action frames — which are freshest-wins and freely dropped — a queued event survives websocket reconnects: it is re-queued on a failed send and retried.

Source code in client/src/armnet_client/teleop_stream.py

def stream_job_teleop(
    job_id: str,
    stop: threading.Event,
    config: TeleoperatorConfig,
    events: "queue.Queue[str] | None" = None,
) -> None:
    """Sample the local leader and push actions to ``job_id`` until ``stop`` is set.

    Intended to run on a background thread (alongside the rerun stream). Returns
    quietly if no API key is configured, LeRobot is missing, or the leader/
    websocket can't be reached, so teleop never crashes the orchestrate script.

    ``events``, when given, is a queue of recording-control event strings (see
    :func:`start_recording_key_listener`); each is sent on the same websocket
    as a :class:`TeleopEvent` frame. Unlike action frames — which are
    freshest-wins and freely dropped — a queued event survives websocket
    reconnects: it is re-queued on a failed send and retried.
    """

    key = api_key()
    if not key:
        logger.warning("no API key configured; remote teleop disabled")
        return

    try:
        leader = _make_leader(config)
    except ImportError:
        logger.error(
            "remote teleop requires LeRobot; install with `pip install 'armnet-client[teleop]'`"
        )
        return
    except Exception as exc:  # noqa: BLE001
        logger.error("failed to construct teleoperator: %r", exc)
        return

    # Imported here so _ws_url's import cost is only paid when teleoperating.
    from armnet_client.execute import _websocket_sslopt, _ws_url

    ws_url = _ws_url(orchestrator_url(), f"/jobs/{job_id}/teleop")

    try:
        leader.connect(calibrate=config.calibrate)
    except Exception as exc:  # noqa: BLE001
        logger.error("failed to connect leader arm on %s: %r", config.port, exc)
        return

    period_s = 1.0 / config.fps if config.fps > 0 else 0.0
    previous_action: dict[str, float] = {}
    try:
        # Reconnect loop: a dropped teleop websocket self-heals and never sets
        # the shared stop, so it can't take down the (separate) rerun stream.
        while not stop.is_set():
            try:
                ws = websocket.create_connection(
                    ws_url,
                    header=[f"{API_KEY_HEADER}: {key}"],
                    timeout=5,
                    sslopt=_websocket_sslopt(),
                )
                # Short timeout: send returns fast, and the periodic recv used to
                # service keepalives/pings stays near-non-blocking.
                ws.settimeout(0.01)
            except Exception as exc:  # noqa: BLE001
                logger.warning("teleop websocket connect failed (retrying): %r", exc)
                if stop.wait(1.0):
                    break
                continue

            logger.info("remote teleop streaming to job %s at %.0f Hz", job_id, config.fps)
            last_drain = time.perf_counter()
            try:
                while not stop.is_set():
                    loop_start = time.perf_counter()
                    try:
                        raw_action = leader.get_action()
                    except Exception:  # noqa: BLE001 - transient read; keep going
                        logger.exception("leader get_action failed")
                        continue
                    action = _finite_action(raw_action, previous_action)
                    previous_action.update(action)
                    message = TeleopAction(
                        job_id=job_id,
                        timestamp=time.time(),
                        action=action,
                    )
                    try:
                        ws.send(message.model_dump_json())
                    except websocket.WebSocketTimeoutException:
                        pass  # transient send-buffer backpressure; drop this sample
                    except Exception:  # noqa: BLE001 - dropped; reconnect
                        logger.warning("teleop websocket send failed; reconnecting", exc_info=True)
                        break
                    if events is not None and not _send_pending_events(ws, job_id, events):
                        break  # ws dropped; unsent events were re-queued
                    # Periodically read so the client services server keepalives
                    # and answers WebSocket pings (a send-only client never would,
                    # and the server would eventually close the connection).
                    now = time.perf_counter()
                    if now - last_drain >= 0.5:
                        last_drain = now
                        try:
                            ws.recv()
                        except websocket.WebSocketTimeoutException:
                            pass
                        except Exception:  # noqa: BLE001 - dropped; reconnect
                            logger.warning("teleop websocket recv failed; reconnecting", exc_info=True)
                            break
                    remaining = period_s - (time.perf_counter() - loop_start)
                    if remaining > 0:
                        time.sleep(remaining)
            finally:
                try:
                    ws.close()
                except Exception:  # noqa: BLE001
                    pass
    finally:
        try:
            leader.disconnect()
        except Exception:  # noqa: BLE001
            pass

armnet_client.execute

Top-level job-submission API: execute and execute_many.

Both target the platform's remote orchestrator + cell infrastructure. In M0.5 the orchestrator and cell happen to run on localhost, but they are still "remote" from the SDK's perspective — the SDK speaks to them only through the orchestrator HTTP API.

The orchestrator URL is hard-coded into the SDK build (see :mod:armnet_client._config); customers do not pass it. Set DEBUG_ARMNET_ORCHESTRATOR_URI to override during development.

logger module-attribute

logger = logging.getLogger(__name__)

LogStreamResult dataclass

How a log stream ended.

error is unauthorized, forbidden, not_found, or another server reason. Those are permanent: the stream does not reconnect. terminal_status is set when the job finished. saw_log is false when the job was already over before this connection, which is when there is nothing to show because logs are not stored.

Source code in client/src/armnet_client/execute.py

@dataclass
class LogStreamResult:
    """How a log stream ended.

    ``error`` is ``unauthorized``, ``forbidden``, ``not_found``, or another
    server reason. Those are permanent: the stream does not reconnect.
    ``terminal_status`` is set when the job finished. ``saw_log`` is false when
    the job was already over before this connection, which is when there is
    nothing to show because logs are not stored.
    """

    terminal_status: str | None = None
    error: str | None = None
    saw_log: bool = False

terminal_status class-attribute instance-attribute

terminal_status: str | None = None

error class-attribute instance-attribute

error: str | None = None

saw_log class-attribute instance-attribute

saw_log: bool = False

execute

execute(*, image: Union[str, Image], embodiment: str, task: Optional[str] = None, args: Optional[Mapping[str, Any]] = None, secrets: Optional[Mapping[str, str]] = None, detach: bool = False, stream_logs: bool = True, use_rerun: bool = False, timeout_seconds: int = 120, allow_queue: bool = True) -> JobResult

Submit a single job to the platform and block until it terminates.

Returns the final :class:JobResult. Use result.return_value to access whatever the customer's @main-decorated function returned; use result.status/result.error for outcome.

The SDK polls the orchestrator until the job reaches a terminal state or until timeout_seconds + 30 seconds have elapsed (the cell enforces the actual timeout_seconds cap on the container; the +30 here is just SDK-side polling slack).

Omit task to submit a task-less job: any cell of embodiment may run it, regardless of the task it is configured for.

allow_queue (default True): the job is durably queued in JetStream. If no cell claims it during the submission response window, this call returns immediately with a non-terminal JobResult (status queued) rather than blocking. Set allow_queue=False for jobs that only make sense running right now (e.g. interactive teleop/rerun): the orchestrator refuses to create the job (raising :class:~armnet_client.client.OrchestratorError, HTTP 409) if it can't be dispatched immediately.

Source code in client/src/armnet_client/execute.py

def execute(
    *,
    image: Union[str, Image],
    embodiment: str,
    task: Optional[str] = None,
    args: Optional[Mapping[str, Any]] = None,
    secrets: Optional[Mapping[str, str]] = None,
    detach: bool = False,
    stream_logs: bool = True,
    use_rerun: bool = False,
    timeout_seconds: int = 120,
    allow_queue: bool = True,
) -> JobResult:
    """Submit a single job to the platform and block until it terminates.

    Returns the final :class:`JobResult`. Use ``result.return_value`` to
    access whatever the customer's ``@main``-decorated function returned;
    use ``result.status``/``result.error`` for outcome.

    The SDK polls the orchestrator until the job reaches a terminal state
    or until ``timeout_seconds + 30`` seconds have elapsed (the cell
    enforces the actual ``timeout_seconds`` cap on the container; the
    +30 here is just SDK-side polling slack).

    Omit ``task`` to submit a task-less job: any cell of ``embodiment`` may run
    it, regardless of the task it is configured for.

    ``allow_queue`` (default True): the job is durably queued in JetStream. If
    no cell claims it during the submission response window, this call returns
    immediately with a non-terminal ``JobResult`` (status ``queued``) rather
    than blocking. Set ``allow_queue=False``
    for jobs that only make sense running right now (e.g. interactive
    teleop/rerun): the orchestrator refuses to create the job (raising
    :class:`~armnet_client.client.OrchestratorError`, HTTP 409) if it can't
    be dispatched immediately.
    """

    spec = JobSpec(
        image=str(image) if isinstance(image, Image) else image,
        embodiment=embodiment,
        task=task,
        args=dict(args) if args is not None else {},
        secrets=dict(secrets) if secrets is not None else {},
        detach=detach,
        timeout_seconds=timeout_seconds,
    )
    return _execute_spec(
        spec, stream_logs=stream_logs, stream_rerun=use_rerun, allow_queue=allow_queue
    )

execute_many

execute_many(specs: Sequence[Mapping[str, Any]], *, max_parallel: int = 4) -> Iterator[JobResult]

Submit many jobs in parallel; yield :class:JobResult as each completes.

specs is a sequence of kwarg-dicts, each accepted by :func:execute (e.g. [{"image": ..., "embodiment": ..., "task": ..., "args": {...}}, ...]). Results are yielded in completion order, not submission order — match them up via your spec's own identifying field if you need to.

max_parallel caps the number of concurrent submissions. Backed by a thread pool; safe to use from synchronous code.

Source code in client/src/armnet_client/execute.py

def execute_many(
    specs: Sequence[Mapping[str, Any]],
    *,
    max_parallel: int = 4,
) -> Iterator[JobResult]:
    """Submit many jobs in parallel; yield :class:`JobResult` as each completes.

    ``specs`` is a sequence of kwarg-dicts, each accepted by
    :func:`execute` (e.g. ``[{"image": ..., "embodiment": ..., "task": ...,
    "args": {...}}, ...]``). Results are yielded in completion order, not
    submission order \u2014 match them up via your spec's own identifying
    field if you need to.

    ``max_parallel`` caps the number of concurrent submissions. Backed by
    a thread pool; safe to use from synchronous code.
    """

    if not specs:
        return

    with ThreadPoolExecutor(max_workers=max_parallel) as pool:
        futures = [pool.submit(execute, **spec) for spec in specs]
        for fut in as_completed(futures):
            yield fut.result()

armnet_client.image

Programmatic Docker image building with content-hash caching.

The :class:Image class lets customer code declaratively build a Docker image from a Dockerfile and pass the resulting reference to :func:execute. Builds are content-addressed: the resulting image tag is derived from a hash of the Dockerfile + every file in the build context, so identical inputs short-circuit to the cached image without re-invoking docker build.

Example::

from armnet_client import Image, execute

img = Image.build(
    dockerfile="Dockerfile",
    context_dir=".",
    name="my-eval-runner",
)
result = execute(image=img, embodiment="lerobot/so-101",
                 task="assemble_block_tower", args={"seed": 0})

The cache lives in the local Docker daemon (we tag with a content hash). For M0.5 the cell runs against the same daemon so cached images are immediately available; for M1+ we'll add a registry push step.

logger module-attribute

logger = logging.getLogger(__name__)

ImageBuildError

Bases: RuntimeError

Raised when docker build fails.

Source code in client/src/armnet_client/image.py

class ImageBuildError(RuntimeError):
    """Raised when ``docker build`` fails."""

ImagePushError

Bases: RuntimeError

Raised when docker login / docker tag / docker push fails.

Source code in client/src/armnet_client/image.py

class ImagePushError(RuntimeError):
    """Raised when ``docker login`` / ``docker tag`` / ``docker push`` fails."""

RegistryCredentialsUnavailableError

Bases: RuntimeError

Raised when the orchestrator can't issue image-registry credentials.

This is the recoverable shape of an :meth:Image.push failure — the orchestrator either doesn't have the endpoint (404), isn't configured for credential issuance (503), or isn't reachable at all (connection refused, typical for pure-local M0.5 dev). Callers can pass if_possible=True to :meth:Image.push to log + fall back to the local image ref instead of propagating this.

Source code in client/src/armnet_client/image.py

class RegistryCredentialsUnavailableError(RuntimeError):
    """Raised when the orchestrator can't issue image-registry credentials.

    This is the recoverable shape of an :meth:`Image.push` failure \u2014 the
    orchestrator either doesn't have the endpoint (404), isn't configured
    for credential issuance (503), or isn't reachable at all (connection
    refused, typical for pure-local M0.5 dev). Callers can pass
    ``if_possible=True`` to :meth:`Image.push` to log + fall back to the
    local image ref instead of propagating this.
    """

ImageOwnershipError

Bases: ValueError

A prebuilt image is outside the authenticated customer's namespace.

Source code in client/src/armnet_client/image.py

class ImageOwnershipError(ValueError):
    """A prebuilt image is outside the authenticated customer's namespace."""

Image dataclass

A reference to a Docker image, either pre-existing or built locally.

Most customer code constructs one with :meth:build. To wrap an existing registry-pushed image, use :meth:from_ref or just pass the image string directly to :func:execute.

Source code in client/src/armnet_client/image.py

@dataclass(frozen=True)
class Image:
    """A reference to a Docker image, either pre-existing or built locally.

    Most customer code constructs one with :meth:`build`. To wrap an
    existing registry-pushed image, use :meth:`from_ref` or just pass the
    image string directly to :func:`execute`.
    """

    ref: str

    def __str__(self) -> str:
        return self.ref

    @classmethod
    def from_ref(cls, ref: str) -> "Image":
        return cls(ref=ref)

    def validate_for_current_user(self, *, if_possible: bool = False) -> "Image":
        """Fast client-side ownership check; the orchestrator remains authoritative.

        In pure-local development the registry endpoint may be unavailable.
        ``if_possible=True`` preserves that workflow while production callers
        still get immediate feedback for another user's or external image.
        """

        try:
            creds = _get_or_fetch_credentials()
        except RegistryCredentialsUnavailableError:
            if if_possible:
                logger.warning(
                    "cannot validate prebuilt image ownership without registry "
                    "credentials; deferring to orchestrator"
                )
                return self
            raise
        namespace_prefix, separator, username = creds.namespace.rpartition("/")
        if not separator:
            raise ImageOwnershipError(
                f"orchestrator returned malformed registry namespace {creds.namespace!r}"
            )
        first = self.ref.split("/", 1)[0]
        explicit_registry = "/" in self.ref and (
            "." in first or ":" in first or first == "localhost"
        )
        if not explicit_registry:
            try:
                return Image(
                    canonical_customer_image(
                        self.ref,
                        registry=creds.registry,
                        namespace_prefix=namespace_prefix,
                        username=username,
                    )
                )
            except ImageReferenceError as exc:
                raise ImageOwnershipError(str(exc)) from exc
        try:
            require_owned_customer_image(
                self.ref,
                registry=creds.registry,
                namespace_prefix=namespace_prefix,
                username=username,
            )
        except ImageReferenceError as exc:
            raise ImageOwnershipError(str(exc)) from exc
        return self

    @classmethod
    def planned_ref(
        cls,
        dockerfile: Union[str, Path],
        name: str,
        context_dir: Union[str, Path, None] = None,
        *,
        platform: Optional[str] = None,
        build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
    ) -> str:
        """The registry ref :meth:`build` then :meth:`push` would produce.

        Worked out from the source alone: the tag is a hash of the build
        context, so the answer needs no Docker daemon and no build. It does
        need the registry credentials, since only they say which namespace the
        image lands in.

        This is what lets a caller ask "is this already pushed?" before
        spending twenty minutes producing a byte-identical image.
        """

        dockerfile_path = Path(dockerfile).resolve()
        if not dockerfile_path.is_file():
            raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")
        context_path = (
            Path(context_dir).resolve() if context_dir is not None else dockerfile_path.parent
        )
        if not context_path.is_dir():
            raise FileNotFoundError(f"build context not found: {context_path}")
        resolved_build_contexts = {
            ctx_name: Path(ctx_path).resolve()
            for ctx_name, ctx_path in (build_contexts or {}).items()
        }
        build_hash = _compute_build_hash(
            dockerfile_path,
            context_path,
            platform=resolve_docker_platform(platform),
            build_contexts=resolved_build_contexts,
        )
        creds = _get_or_fetch_credentials()
        return f"{creds.registry}/{creds.namespace}/{_tag_for(name, build_hash)}"

    @staticmethod
    def exists_in_registry(ref: str) -> bool:
        """Whether ``ref`` is already in the registry.

        Asks the registry's manifest endpoint directly with the same
        short-lived token a push would use, so it costs one request and needs
        no Docker.
        """

        registry, _, repository_and_tag = ref.partition("/")
        repository, _, tag = repository_and_tag.rpartition(":")
        if not registry or not repository or not tag:
            raise ValueError(f"not a registry ref: {ref!r}")
        creds = _get_or_fetch_credentials()
        response = httpx.head(
            f"https://{registry}/v2/{repository}/manifests/{tag}",
            headers={
                "Authorization": f"Bearer {creds.password}",
                "Accept": ", ".join(_MANIFEST_MEDIA_TYPES),
            },
            timeout=_REGISTRY_TIMEOUT_S,
            follow_redirects=True,
        )
        if response.status_code == 404:
            return False
        response.raise_for_status()
        return True

    def push(self, *, if_possible: bool = False) -> "Image":
        """Push this image to the platform's registry; return a new Image
        whose ref points at the registry.

        Flow:

          1. Fetch short-lived OAuth2 credentials from the orchestrator's
             ``POST /registry/credentials`` endpoint (cached on disk;
             refreshed when within 5 minutes of expiry).
          2. ``docker login <registry> -u oauth2accesstoken --password-stdin``.
          3. ``docker tag <local> <registry>/<namespace>/<local>``.
          4. ``docker push`` the new tag.
          5. Return ``Image(ref=<full registry ref>)``.

        Customers don't need any GCP credentials; the orchestrator does
        the IAM dance on their behalf via service-account impersonation.

        For pure-local M0.5 dev where the orchestrator + cell share a
        docker daemon and pushing isn't necessary, pass ``if_possible=True``
        \u2014 the method then logs a warning and returns ``self`` unchanged
        if the orchestrator can't issue credentials, instead of raising.
        Real ``docker push`` failures (network, auth, etc.) always raise.
        """

        _ensure_docker_daemon_available(operation="push images")
        try:
            creds = _get_or_fetch_credentials()
        except RegistryCredentialsUnavailableError as exc:
            if if_possible:
                logger.warning(
                    "Image.push: registry credentials unavailable (%s); "
                    "falling back to local image ref %s",
                    exc,
                    self.ref,
                )
                return self
            raise
        target_ref = _docker_login_tag_push(local_ref=self.ref, creds=creds)
        return Image(ref=target_ref)

    @classmethod
    def build(
        cls,
        dockerfile: Union[str, Path],
        name: str,
        context_dir: Union[str, Path, None] = None,
        force_rebuild: bool = False,
        platform: Optional[str] = None,
        build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
        cache_from: Optional[Iterable[str]] = None,
        cache_to: Optional[Iterable[str]] = None,
    ) -> "Image":
        """Build a Docker image and return a content-addressed reference.

        :param dockerfile: Path to the Dockerfile.
        :param name: Image name (the part before the tag); used as a
            human-readable prefix on the resulting tag.
        :param context_dir: Build context directory. Defaults to the
            Dockerfile's parent.
        :param force_rebuild: If True, always invoke ``docker build`` even
            when an image with the computed content-hash tag already
            exists.
        :param platform: Optional Docker platform, e.g. ``linux/arm64/v8`` or
            ``linux/amd64``. Defaults to ``ARMNET_DOCKER_PLATFORM`` when
            set. Otherwise an x86 host uses Docker's native platform, and
            any other host (Apple Silicon included) builds ``linux/amd64``
            so the image can run on a cell.
        :param build_contexts: Optional named build contexts forwarded to
            ``docker build --build-context name=path``. The Dockerfile can
            then reference them via ``COPY --from=name ...``. Useful for
            pulling in source trees that live outside the primary build
            context (e.g. a sibling checkout on the build host). Contents
            participate in the content hash so changes invalidate the cache.
        :param cache_from: Optional BuildKit ``--cache-from`` specs (e.g.
            ``["type=gha,scope=my-image"]``). When any cache is requested the
            build runs via ``docker buildx build --load`` so layers can be
            reused across machines/CI runs. Falls back to a plain
            ``docker build`` (no cache) if buildx or the cache backend is
            unavailable, so enabling caching can never break a build.
        :param cache_to: Optional BuildKit ``--cache-to`` specs (e.g.
            ``["type=gha,scope=my-image,mode=max"]``).
        :returns: An :class:`Image` whose ``ref`` is
            ``<name>:armnet-<short-hash>``.

        As a convenience for CI, setting ``ARMNET_DOCKER_BUILDX_CACHE=gha`` in
        the environment enables GitHub-Actions layer caching (scoped per image
        ``name``) without passing ``cache_from`` / ``cache_to`` explicitly.
        """

        dockerfile_path = Path(dockerfile).resolve()
        if not dockerfile_path.is_file():
            raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")

        context_path = (
            Path(context_dir).resolve()
            if context_dir is not None
            else dockerfile_path.parent
        )
        if not context_path.is_dir():
            raise FileNotFoundError(f"build context not found: {context_path}")

        resolved_build_contexts: dict[str, Path] = {}
        for ctx_name, ctx_path in (build_contexts or {}).items():
            resolved = Path(ctx_path).resolve()
            if not resolved.is_dir():
                raise FileNotFoundError(
                    f"named build context {ctx_name!r} not found: {resolved}"
                )
            resolved_build_contexts[ctx_name] = resolved

        _ensure_docker_daemon_available(operation="build images")
        platform = resolve_docker_platform(platform)
        build_hash = _compute_build_hash(
            dockerfile_path,
            context_path,
            platform=platform,
            build_contexts=resolved_build_contexts,
        )
        tag = _tag_for(name, build_hash)

        resolved_cache_from, resolved_cache_to = _resolve_buildx_cache(
            name, cache_from, cache_to
        )
        if resolved_cache_from or resolved_cache_to:
            # Layer-cached path (CI). Build with buildx + an external cache
            # backend so expensive layers survive across runs. Uses the ambient
            # Docker config (not an isolated one) so the buildx builder created
            # by e.g. setup-buildx-action and the ACTIONS_* cache tokens are
            # visible. On any failure we fall through to the plain build below,
            # so caching can never turn a working build into a failing one.
            if not force_rebuild and _image_exists_locally(tag):
                logger.info("image %s already built; skipping docker build", tag)
                return cls(ref=tag)
            try:
                _docker_buildx_build(
                    tag=tag,
                    dockerfile=dockerfile_path,
                    context_dir=context_path,
                    platform=platform,
                    build_contexts=resolved_build_contexts,
                    cache_from=resolved_cache_from,
                    cache_to=resolved_cache_to,
                )
                return cls(ref=tag)
            except ImageBuildError as exc:
                logger.warning(
                    "cached buildx build failed (%s); retrying with a plain "
                    "`docker build` (no layer cache)",
                    exc,
                )
        elif platform:
            # Cross-platform builds require buildx; the legacy builder cannot
            # reliably track intermediate layers under QEMU emulation.
            if not force_rebuild and _image_exists_locally(tag):
                logger.info("image %s already built; skipping docker build", tag)
                return cls(ref=tag)
            _docker_buildx_build(
                tag=tag,
                dockerfile=dockerfile_path,
                context_dir=context_path,
                platform=platform,
                build_contexts=resolved_build_contexts,
                cache_from=[],
                cache_to=[],
            )
            return cls(ref=tag)

        with _isolated_docker_config() as docker_env:
            if not force_rebuild and _image_exists_locally(tag, env=docker_env):
                logger.info("image %s already built; skipping docker build", tag)
                return cls(ref=tag)

            cmd = [
                "docker", "build",
                "-t", tag,
                "-f", str(dockerfile_path),
            ]
            if platform:
                cmd.extend(["--platform", platform])
            for ctx_name, ctx_path in resolved_build_contexts.items():
                cmd.extend(["--build-context", f"{ctx_name}={ctx_path}"])
            cmd.append(str(context_path))
            logger.info("docker build %s", " ".join(cmd[2:]))
            try:
                subprocess.run(cmd, check=True, env=docker_env)
            except subprocess.CalledProcessError as exc:
                raise ImageBuildError(
                    f"docker build failed (exit {exc.returncode}) for {tag}"
                ) from exc
            except FileNotFoundError as exc:
                # `docker` binary not on PATH.
                raise ImageBuildError(
                    "`docker` binary not found on PATH; install Docker to build images."
                ) from exc

        return cls(ref=tag)

ref instance-attribute

ref: str

from_ref classmethod

from_ref(ref: str) -> 'Image'

Source code in client/src/armnet_client/image.py

@classmethod
def from_ref(cls, ref: str) -> "Image":
    return cls(ref=ref)

validate_for_current_user

validate_for_current_user(*, if_possible: bool = False) -> 'Image'

Fast client-side ownership check; the orchestrator remains authoritative.

In pure-local development the registry endpoint may be unavailable. if_possible=True preserves that workflow while production callers still get immediate feedback for another user's or external image.

Source code in client/src/armnet_client/image.py

def validate_for_current_user(self, *, if_possible: bool = False) -> "Image":
    """Fast client-side ownership check; the orchestrator remains authoritative.

    In pure-local development the registry endpoint may be unavailable.
    ``if_possible=True`` preserves that workflow while production callers
    still get immediate feedback for another user's or external image.
    """

    try:
        creds = _get_or_fetch_credentials()
    except RegistryCredentialsUnavailableError:
        if if_possible:
            logger.warning(
                "cannot validate prebuilt image ownership without registry "
                "credentials; deferring to orchestrator"
            )
            return self
        raise
    namespace_prefix, separator, username = creds.namespace.rpartition("/")
    if not separator:
        raise ImageOwnershipError(
            f"orchestrator returned malformed registry namespace {creds.namespace!r}"
        )
    first = self.ref.split("/", 1)[0]
    explicit_registry = "/" in self.ref and (
        "." in first or ":" in first or first == "localhost"
    )
    if not explicit_registry:
        try:
            return Image(
                canonical_customer_image(
                    self.ref,
                    registry=creds.registry,
                    namespace_prefix=namespace_prefix,
                    username=username,
                )
            )
        except ImageReferenceError as exc:
            raise ImageOwnershipError(str(exc)) from exc
    try:
        require_owned_customer_image(
            self.ref,
            registry=creds.registry,
            namespace_prefix=namespace_prefix,
            username=username,
        )
    except ImageReferenceError as exc:
        raise ImageOwnershipError(str(exc)) from exc
    return self

planned_ref classmethod

planned_ref(dockerfile: Union[str, Path], name: str, context_dir: Union[str, Path, None] = None, *, platform: Optional[str] = None, build_contexts: Optional[Mapping[str, Union[str, Path]]] = None) -> str

The registry ref :meth:build then :meth:push would produce.

Worked out from the source alone: the tag is a hash of the build context, so the answer needs no Docker daemon and no build. It does need the registry credentials, since only they say which namespace the image lands in.

This is what lets a caller ask "is this already pushed?" before spending twenty minutes producing a byte-identical image.

Source code in client/src/armnet_client/image.py

@classmethod
def planned_ref(
    cls,
    dockerfile: Union[str, Path],
    name: str,
    context_dir: Union[str, Path, None] = None,
    *,
    platform: Optional[str] = None,
    build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
) -> str:
    """The registry ref :meth:`build` then :meth:`push` would produce.

    Worked out from the source alone: the tag is a hash of the build
    context, so the answer needs no Docker daemon and no build. It does
    need the registry credentials, since only they say which namespace the
    image lands in.

    This is what lets a caller ask "is this already pushed?" before
    spending twenty minutes producing a byte-identical image.
    """

    dockerfile_path = Path(dockerfile).resolve()
    if not dockerfile_path.is_file():
        raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")
    context_path = (
        Path(context_dir).resolve() if context_dir is not None else dockerfile_path.parent
    )
    if not context_path.is_dir():
        raise FileNotFoundError(f"build context not found: {context_path}")
    resolved_build_contexts = {
        ctx_name: Path(ctx_path).resolve()
        for ctx_name, ctx_path in (build_contexts or {}).items()
    }
    build_hash = _compute_build_hash(
        dockerfile_path,
        context_path,
        platform=resolve_docker_platform(platform),
        build_contexts=resolved_build_contexts,
    )
    creds = _get_or_fetch_credentials()
    return f"{creds.registry}/{creds.namespace}/{_tag_for(name, build_hash)}"

exists_in_registry staticmethod

exists_in_registry(ref: str) -> bool

Whether ref is already in the registry.

Asks the registry's manifest endpoint directly with the same short-lived token a push would use, so it costs one request and needs no Docker.

Source code in client/src/armnet_client/image.py

@staticmethod
def exists_in_registry(ref: str) -> bool:
    """Whether ``ref`` is already in the registry.

    Asks the registry's manifest endpoint directly with the same
    short-lived token a push would use, so it costs one request and needs
    no Docker.
    """

    registry, _, repository_and_tag = ref.partition("/")
    repository, _, tag = repository_and_tag.rpartition(":")
    if not registry or not repository or not tag:
        raise ValueError(f"not a registry ref: {ref!r}")
    creds = _get_or_fetch_credentials()
    response = httpx.head(
        f"https://{registry}/v2/{repository}/manifests/{tag}",
        headers={
            "Authorization": f"Bearer {creds.password}",
            "Accept": ", ".join(_MANIFEST_MEDIA_TYPES),
        },
        timeout=_REGISTRY_TIMEOUT_S,
        follow_redirects=True,
    )
    if response.status_code == 404:
        return False
    response.raise_for_status()
    return True

push

push(*, if_possible: bool = False) -> 'Image'

Push this image to the platform's registry; return a new Image whose ref points at the registry.

Flow:

1. Fetch short-lived OAuth2 credentials from the orchestrator's POST /registry/credentials endpoint (cached on disk; refreshed when within 5 minutes of expiry).
2. docker login <registry> -u oauth2accesstoken --password-stdin.
3. docker tag <local> <registry>/<namespace>/<local>.
4. docker push the new tag.
5. Return Image(ref=<full registry ref>).

Customers don't need any GCP credentials; the orchestrator does the IAM dance on their behalf via service-account impersonation.

For pure-local M0.5 dev where the orchestrator + cell share a docker daemon and pushing isn't necessary, pass if_possible=True — the method then logs a warning and returns self unchanged if the orchestrator can't issue credentials, instead of raising. Real docker push failures (network, auth, etc.) always raise.

Source code in client/src/armnet_client/image.py

def push(self, *, if_possible: bool = False) -> "Image":
    """Push this image to the platform's registry; return a new Image
    whose ref points at the registry.

    Flow:

      1. Fetch short-lived OAuth2 credentials from the orchestrator's
         ``POST /registry/credentials`` endpoint (cached on disk;
         refreshed when within 5 minutes of expiry).
      2. ``docker login <registry> -u oauth2accesstoken --password-stdin``.
      3. ``docker tag <local> <registry>/<namespace>/<local>``.
      4. ``docker push`` the new tag.
      5. Return ``Image(ref=<full registry ref>)``.

    Customers don't need any GCP credentials; the orchestrator does
    the IAM dance on their behalf via service-account impersonation.

    For pure-local M0.5 dev where the orchestrator + cell share a
    docker daemon and pushing isn't necessary, pass ``if_possible=True``
    \u2014 the method then logs a warning and returns ``self`` unchanged
    if the orchestrator can't issue credentials, instead of raising.
    Real ``docker push`` failures (network, auth, etc.) always raise.
    """

    _ensure_docker_daemon_available(operation="push images")
    try:
        creds = _get_or_fetch_credentials()
    except RegistryCredentialsUnavailableError as exc:
        if if_possible:
            logger.warning(
                "Image.push: registry credentials unavailable (%s); "
                "falling back to local image ref %s",
                exc,
                self.ref,
            )
            return self
        raise
    target_ref = _docker_login_tag_push(local_ref=self.ref, creds=creds)
    return Image(ref=target_ref)

build classmethod

build(dockerfile: Union[str, Path], name: str, context_dir: Union[str, Path, None] = None, force_rebuild: bool = False, platform: Optional[str] = None, build_contexts: Optional[Mapping[str, Union[str, Path]]] = None, cache_from: Optional[Iterable[str]] = None, cache_to: Optional[Iterable[str]] = None) -> 'Image'

Build a Docker image and return a content-addressed reference.

:param dockerfile: Path to the Dockerfile. :param name: Image name (the part before the tag); used as a human-readable prefix on the resulting tag. :param context_dir: Build context directory. Defaults to the Dockerfile's parent. :param force_rebuild: If True, always invoke docker build even when an image with the computed content-hash tag already exists. :param platform: Optional Docker platform, e.g. linux/arm64/v8 or linux/amd64. Defaults to ARMNET_DOCKER_PLATFORM when set. Otherwise an x86 host uses Docker's native platform, and any other host (Apple Silicon included) builds linux/amd64 so the image can run on a cell. :param build_contexts: Optional named build contexts forwarded to docker build --build-context name=path. The Dockerfile can then reference them via COPY --from=name .... Useful for pulling in source trees that live outside the primary build context (e.g. a sibling checkout on the build host). Contents participate in the content hash so changes invalidate the cache. :param cache_from: Optional BuildKit --cache-from specs (e.g. ["type=gha,scope=my-image"]). When any cache is requested the build runs via docker buildx build --load so layers can be reused across machines/CI runs. Falls back to a plain docker build (no cache) if buildx or the cache backend is unavailable, so enabling caching can never break a build. :param cache_to: Optional BuildKit --cache-to specs (e.g. ["type=gha,scope=my-image,mode=max"]). :returns: An :class:Image whose ref is <name>:armnet-<short-hash>.

As a convenience for CI, setting ARMNET_DOCKER_BUILDX_CACHE=gha in the environment enables GitHub-Actions layer caching (scoped per image name) without passing cache_from / cache_to explicitly.

Source code in client/src/armnet_client/image.py

@classmethod
def build(
    cls,
    dockerfile: Union[str, Path],
    name: str,
    context_dir: Union[str, Path, None] = None,
    force_rebuild: bool = False,
    platform: Optional[str] = None,
    build_contexts: Optional[Mapping[str, Union[str, Path]]] = None,
    cache_from: Optional[Iterable[str]] = None,
    cache_to: Optional[Iterable[str]] = None,
) -> "Image":
    """Build a Docker image and return a content-addressed reference.

    :param dockerfile: Path to the Dockerfile.
    :param name: Image name (the part before the tag); used as a
        human-readable prefix on the resulting tag.
    :param context_dir: Build context directory. Defaults to the
        Dockerfile's parent.
    :param force_rebuild: If True, always invoke ``docker build`` even
        when an image with the computed content-hash tag already
        exists.
    :param platform: Optional Docker platform, e.g. ``linux/arm64/v8`` or
        ``linux/amd64``. Defaults to ``ARMNET_DOCKER_PLATFORM`` when
        set. Otherwise an x86 host uses Docker's native platform, and
        any other host (Apple Silicon included) builds ``linux/amd64``
        so the image can run on a cell.
    :param build_contexts: Optional named build contexts forwarded to
        ``docker build --build-context name=path``. The Dockerfile can
        then reference them via ``COPY --from=name ...``. Useful for
        pulling in source trees that live outside the primary build
        context (e.g. a sibling checkout on the build host). Contents
        participate in the content hash so changes invalidate the cache.
    :param cache_from: Optional BuildKit ``--cache-from`` specs (e.g.
        ``["type=gha,scope=my-image"]``). When any cache is requested the
        build runs via ``docker buildx build --load`` so layers can be
        reused across machines/CI runs. Falls back to a plain
        ``docker build`` (no cache) if buildx or the cache backend is
        unavailable, so enabling caching can never break a build.
    :param cache_to: Optional BuildKit ``--cache-to`` specs (e.g.
        ``["type=gha,scope=my-image,mode=max"]``).
    :returns: An :class:`Image` whose ``ref`` is
        ``<name>:armnet-<short-hash>``.

    As a convenience for CI, setting ``ARMNET_DOCKER_BUILDX_CACHE=gha`` in
    the environment enables GitHub-Actions layer caching (scoped per image
    ``name``) without passing ``cache_from`` / ``cache_to`` explicitly.
    """

    dockerfile_path = Path(dockerfile).resolve()
    if not dockerfile_path.is_file():
        raise FileNotFoundError(f"Dockerfile not found: {dockerfile_path}")

    context_path = (
        Path(context_dir).resolve()
        if context_dir is not None
        else dockerfile_path.parent
    )
    if not context_path.is_dir():
        raise FileNotFoundError(f"build context not found: {context_path}")

    resolved_build_contexts: dict[str, Path] = {}
    for ctx_name, ctx_path in (build_contexts or {}).items():
        resolved = Path(ctx_path).resolve()
        if not resolved.is_dir():
            raise FileNotFoundError(
                f"named build context {ctx_name!r} not found: {resolved}"
            )
        resolved_build_contexts[ctx_name] = resolved

    _ensure_docker_daemon_available(operation="build images")
    platform = resolve_docker_platform(platform)
    build_hash = _compute_build_hash(
        dockerfile_path,
        context_path,
        platform=platform,
        build_contexts=resolved_build_contexts,
    )
    tag = _tag_for(name, build_hash)

    resolved_cache_from, resolved_cache_to = _resolve_buildx_cache(
        name, cache_from, cache_to
    )
    if resolved_cache_from or resolved_cache_to:
        # Layer-cached path (CI). Build with buildx + an external cache
        # backend so expensive layers survive across runs. Uses the ambient
        # Docker config (not an isolated one) so the buildx builder created
        # by e.g. setup-buildx-action and the ACTIONS_* cache tokens are
        # visible. On any failure we fall through to the plain build below,
        # so caching can never turn a working build into a failing one.
        if not force_rebuild and _image_exists_locally(tag):
            logger.info("image %s already built; skipping docker build", tag)
            return cls(ref=tag)
        try:
            _docker_buildx_build(
                tag=tag,
                dockerfile=dockerfile_path,
                context_dir=context_path,
                platform=platform,
                build_contexts=resolved_build_contexts,
                cache_from=resolved_cache_from,
                cache_to=resolved_cache_to,
            )
            return cls(ref=tag)
        except ImageBuildError as exc:
            logger.warning(
                "cached buildx build failed (%s); retrying with a plain "
                "`docker build` (no layer cache)",
                exc,
            )
    elif platform:
        # Cross-platform builds require buildx; the legacy builder cannot
        # reliably track intermediate layers under QEMU emulation.
        if not force_rebuild and _image_exists_locally(tag):
            logger.info("image %s already built; skipping docker build", tag)
            return cls(ref=tag)
        _docker_buildx_build(
            tag=tag,
            dockerfile=dockerfile_path,
            context_dir=context_path,
            platform=platform,
            build_contexts=resolved_build_contexts,
            cache_from=[],
            cache_to=[],
        )
        return cls(ref=tag)

    with _isolated_docker_config() as docker_env:
        if not force_rebuild and _image_exists_locally(tag, env=docker_env):
            logger.info("image %s already built; skipping docker build", tag)
            return cls(ref=tag)

        cmd = [
            "docker", "build",
            "-t", tag,
            "-f", str(dockerfile_path),
        ]
        if platform:
            cmd.extend(["--platform", platform])
        for ctx_name, ctx_path in resolved_build_contexts.items():
            cmd.extend(["--build-context", f"{ctx_name}={ctx_path}"])
        cmd.append(str(context_path))
        logger.info("docker build %s", " ".join(cmd[2:]))
        try:
            subprocess.run(cmd, check=True, env=docker_env)
        except subprocess.CalledProcessError as exc:
            raise ImageBuildError(
                f"docker build failed (exit {exc.returncode}) for {tag}"
            ) from exc
        except FileNotFoundError as exc:
            # `docker` binary not on PATH.
            raise ImageBuildError(
                "`docker` binary not found on PATH; install Docker to build images."
            ) from exc

    return cls(ref=tag)

default_docker_platform

default_docker_platform() -> str | None

Platform to build when the caller did not pass one.

ARMNET_DOCKER_PLATFORM wins. An x86 host keeps Docker's native platform. Any other host, including Apple Silicon, builds linux/amd64 because the cells are x86.

Source code in client/src/armnet_client/image.py

def default_docker_platform() -> str | None:
    """Platform to build when the caller did not pass one.

    ``ARMNET_DOCKER_PLATFORM`` wins. An x86 host keeps Docker's native
    platform. Any other host, including Apple Silicon, builds
    ``linux/amd64`` because the cells are x86.
    """

    explicit = os.environ.get("ARMNET_DOCKER_PLATFORM", "").strip()
    if explicit:
        return explicit
    if host_platform.machine().lower() in _X86_MACHINES:
        return None
    return "linux/amd64"

resolve_docker_platform

resolve_docker_platform(explicit: str | None = None) -> str | None

Caller argument, else :func:default_docker_platform.

Source code in client/src/armnet_client/image.py

def resolve_docker_platform(explicit: str | None = None) -> str | None:
    """Caller argument, else :func:`default_docker_platform`."""

    if explicit:
        return explicit
    return default_docker_platform()

armnet_client.client

HTTP client for the orchestrator.

Thin wrapper around the two M0.5 endpoints. Used internally by :func:armnet_client.execute; exposed publicly for power users that need fine-grained control (custom polling, fire-and-forget submissions, etc.).

OrchestratorError

Bases: RuntimeError

Raised when the orchestrator returns a non-2xx response.

status_code is the HTTP status returned by the orchestrator; None if the request never got a response (e.g. connection refused). Callers can branch on it to distinguish e.g. "endpoint not implemented" (404) or "configured-but-unavailable" (503) from real failures.

Source code in client/src/armnet_client/client.py

class OrchestratorError(RuntimeError):
    """Raised when the orchestrator returns a non-2xx response.

    ``status_code`` is the HTTP status returned by the orchestrator;
    ``None`` if the request never got a response (e.g. connection refused).
    Callers can branch on it to distinguish e.g. "endpoint not implemented"
    (404) or "configured-but-unavailable" (503) from real failures.
    """

    def __init__(self, message: str, *, status_code: Optional[int] = None) -> None:
        super().__init__(message)
        self.status_code = status_code

status_code instance-attribute

status_code = status_code

OrchestratorClient

Synchronous client for the orchestrator HTTP API.

Defaults to Armnet's production API and the ARMNET_API_KEY environment variable. Pass base_url or api_key explicitly for development and custom integrations.

Source code in client/src/armnet_client/client.py

class OrchestratorClient:
    """Synchronous client for the orchestrator HTTP API.

    Defaults to Armnet's production API and the ``ARMNET_API_KEY`` environment
    variable. Pass ``base_url`` or ``api_key`` explicitly for development and
    custom integrations.
    """

    def __init__(
        self,
        base_url: Optional[str] = None,
        timeout: float = 30.0,
        api_key: Optional[str] = None,
    ) -> None:
        key = api_key if api_key is not None else default_api_key()
        headers = {API_KEY_HEADER: key} if key else None
        self._client = httpx.Client(
            base_url=(base_url or orchestrator_url()).rstrip("/"),
            timeout=timeout,
            headers=headers,
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "OrchestratorClient":
        return self

    def __exit__(self, *_exc) -> None:
        self.close()

    def submit(self, spec: JobSpec, *, allow_queue: bool = True) -> Job:
        """POST /jobs — create a new job.

        The returned :class:`Job` carries ``dispatched``: True if it was sent to
        a cell immediately, False if it was accepted but queued.

        ``allow_queue`` (default True): when no matching cell is free, queue the
        job (True) or have the orchestrator refuse it with HTTP 409 (False, for
        callers that only make sense running immediately).
        """
        resp = self._client.post(
            "/jobs",
            json=spec.model_dump(mode="json"),
            params={"allow_queue": "true" if allow_queue else "false"},
        )
        _raise_for_status(resp)
        return Job.model_validate(resp.json())

    def get(self, job_id: str) -> Job:
        """GET /jobs/{id} \u2014 fetch the current state of a job."""
        resp = self._client.get(f"/jobs/{job_id}")
        _raise_for_status(resp)
        return Job.model_validate(resp.json())

    def list_jobs(
        self, *, limit: int = 10, queued_only: bool = False
    ) -> list[Job]:
        """GET /jobs — the current user's most-recent jobs (newest first)."""
        resp = self._client.get(
            "/jobs", params={"limit": limit, "queued_only": queued_only}
        )
        _raise_for_status(resp)
        return [Job.model_validate(item) for item in resp.json()]

    def get_status_transitions(self, job_id: str) -> list[JobStatusTransition]:
        """GET /jobs/{id}/status-transitions — a job's status-change audit trail."""
        resp = self._client.get(f"/jobs/{job_id}/status-transitions")
        _raise_for_status(resp)
        return [JobStatusTransition.model_validate(item) for item in resp.json()]

    def stop_job(self, job_id: str, *, reason: str = "client requested stop") -> Job:
        """POST /jobs/{id}/stop \u2014 request an immediate stop of a running job.

        Used when the operator intentionally aborts (e.g. Ctrl-C): this cancels
        the job now rather than waiting out the orchestrator's log-disconnect
        grace window.
        """
        resp = self._client.post(f"/jobs/{job_id}/stop", params={"reason": reason})
        _raise_for_status(resp)
        return Job.model_validate(resp.json())

    def get_cell_status(
        self,
        *,
        embodiment: str = DEFAULT_EMBODIMENT,
        task: Optional[str] = None,
    ) -> CellStatusSummary:
        """GET /cells/status \u2014 fetch current availability for a task."""

        params = {"embodiment": embodiment}
        if task is not None:
            params["task"] = task
        resp = self._client.get("/cells/status", params=params)
        _raise_for_status(resp)
        return CellStatusSummary.model_validate(resp.json())

    def get_registry_credentials(self) -> RegistryCredentials:
        """POST /registry/credentials \u2014 fetch fresh image-registry credentials.

        Used by :meth:`armnet_client.Image.push` to get a short-lived
        OAuth2 token for pushing customer images to the platform's
        Artifact Registry repo. The orchestrator returns 503 if it isn't
        configured for credential issuance (typical for local dev).
        """
        resp = self._client.post("/registry/credentials")
        _raise_for_status(resp)
        return RegistryCredentials.model_validate(resp.json())

    def whoami(self) -> WhoAmI:
        """GET /whoami — fetch the username for the current API key."""

        resp = self._client.get("/whoami")
        _raise_for_status(resp)
        return WhoAmI.model_validate(resp.json())

    def list_secrets(self) -> SecretList:
        """GET /secrets — list secret names for the current user."""

        resp = self._client.get("/secrets")
        _raise_for_status(resp)
        return SecretList.model_validate(resp.json())

    def create_secret(self, name: str, value: str) -> SecretInfo:
        """POST /secrets — create or replace one secret value."""

        resp = self._client.post(
            "/secrets",
            json=SecretCreateRequest(name=name, value=value).model_dump(mode="json"),
        )
        _raise_for_status(resp)
        return SecretInfo.model_validate(resp.json())

    def delete_secret(self, name: str) -> None:
        """DELETE /secrets/{name} — delete one user secret if it exists."""

        resp = self._client.delete(f"/secrets/{name}")
        _raise_for_status(resp)

    def get_volume_credentials(self) -> VolumeCredentials:
        """POST /volume/credentials — fetch short-lived GCS volume credentials."""

        resp = self._client.post("/volume/credentials")
        _raise_for_status(resp)
        return VolumeCredentials.model_validate(resp.json())

    def wait(
        self,
        job_id: str,
        poll_interval_seconds: float = 1.0,
        deadline_seconds: Optional[float] = None,
    ) -> Job:
        """Poll ``GET /jobs/{id}`` until the job reaches a terminal state.

        Returns the final :class:`Job`. Raises :class:`TimeoutError` if
        ``deadline_seconds`` elapses first.
        """
        start = time.monotonic()
        try:
            while True:
                job = self.get(job_id)
                if job.status in TerminalStatus:
                    return job
                if (
                    deadline_seconds is not None
                    and (time.monotonic() - start) >= deadline_seconds
                ):
                    raise TimeoutError(
                        f"Job {job_id} did not reach a terminal state within "
                        f"{deadline_seconds}s (last status: {job.status})."
                    )
                time.sleep(poll_interval_seconds)
        except KeyboardInterrupt:
            # Operator intentionally aborted: stop the job now instead of leaving
            # it to the orchestrator's log-disconnect grace window. Best-effort —
            # always re-raise so the interrupt still propagates.
            try:
                self.stop_job(job_id, reason="client interrupted (KeyboardInterrupt)")
            except Exception:  # noqa: BLE001
                pass
            raise

close

close() -> None

Source code in client/src/armnet_client/client.py

def close(self) -> None:
    self._client.close()

submit

submit(spec: JobSpec, *, allow_queue: bool = True) -> Job

POST /jobs — create a new job.

The returned :class:Job carries dispatched: True if it was sent to a cell immediately, False if it was accepted but queued.

allow_queue (default True): when no matching cell is free, queue the job (True) or have the orchestrator refuse it with HTTP 409 (False, for callers that only make sense running immediately).

Source code in client/src/armnet_client/client.py

def submit(self, spec: JobSpec, *, allow_queue: bool = True) -> Job:
    """POST /jobs — create a new job.

    The returned :class:`Job` carries ``dispatched``: True if it was sent to
    a cell immediately, False if it was accepted but queued.

    ``allow_queue`` (default True): when no matching cell is free, queue the
    job (True) or have the orchestrator refuse it with HTTP 409 (False, for
    callers that only make sense running immediately).
    """
    resp = self._client.post(
        "/jobs",
        json=spec.model_dump(mode="json"),
        params={"allow_queue": "true" if allow_queue else "false"},
    )
    _raise_for_status(resp)
    return Job.model_validate(resp.json())

get

get(job_id: str) -> Job

GET /jobs/{id} — fetch the current state of a job.

Source code in client/src/armnet_client/client.py

def get(self, job_id: str) -> Job:
    """GET /jobs/{id} \u2014 fetch the current state of a job."""
    resp = self._client.get(f"/jobs/{job_id}")
    _raise_for_status(resp)
    return Job.model_validate(resp.json())

list_jobs

list_jobs(*, limit: int = 10, queued_only: bool = False) -> list[Job]

GET /jobs — the current user's most-recent jobs (newest first).

Source code in client/src/armnet_client/client.py

def list_jobs(
    self, *, limit: int = 10, queued_only: bool = False
) -> list[Job]:
    """GET /jobs — the current user's most-recent jobs (newest first)."""
    resp = self._client.get(
        "/jobs", params={"limit": limit, "queued_only": queued_only}
    )
    _raise_for_status(resp)
    return [Job.model_validate(item) for item in resp.json()]

get_status_transitions

get_status_transitions(job_id: str) -> list[JobStatusTransition]

GET /jobs/{id}/status-transitions — a job's status-change audit trail.

Source code in client/src/armnet_client/client.py

def get_status_transitions(self, job_id: str) -> list[JobStatusTransition]:
    """GET /jobs/{id}/status-transitions — a job's status-change audit trail."""
    resp = self._client.get(f"/jobs/{job_id}/status-transitions")
    _raise_for_status(resp)
    return [JobStatusTransition.model_validate(item) for item in resp.json()]

stop_job

stop_job(job_id: str, *, reason: str = 'client requested stop') -> Job

POST /jobs/{id}/stop — request an immediate stop of a running job.

Used when the operator intentionally aborts (e.g. Ctrl-C): this cancels the job now rather than waiting out the orchestrator's log-disconnect grace window.

Source code in client/src/armnet_client/client.py

def stop_job(self, job_id: str, *, reason: str = "client requested stop") -> Job:
    """POST /jobs/{id}/stop \u2014 request an immediate stop of a running job.

    Used when the operator intentionally aborts (e.g. Ctrl-C): this cancels
    the job now rather than waiting out the orchestrator's log-disconnect
    grace window.
    """
    resp = self._client.post(f"/jobs/{job_id}/stop", params={"reason": reason})
    _raise_for_status(resp)
    return Job.model_validate(resp.json())

get_cell_status

get_cell_status(*, embodiment: str = DEFAULT_EMBODIMENT, task: Optional[str] = None) -> CellStatusSummary

GET /cells/status — fetch current availability for a task.

Source code in client/src/armnet_client/client.py

def get_cell_status(
    self,
    *,
    embodiment: str = DEFAULT_EMBODIMENT,
    task: Optional[str] = None,
) -> CellStatusSummary:
    """GET /cells/status \u2014 fetch current availability for a task."""

    params = {"embodiment": embodiment}
    if task is not None:
        params["task"] = task
    resp = self._client.get("/cells/status", params=params)
    _raise_for_status(resp)
    return CellStatusSummary.model_validate(resp.json())

get_registry_credentials

get_registry_credentials() -> RegistryCredentials

POST /registry/credentials — fetch fresh image-registry credentials.

Used by :meth:armnet_client.Image.push to get a short-lived OAuth2 token for pushing customer images to the platform's Artifact Registry repo. The orchestrator returns 503 if it isn't configured for credential issuance (typical for local dev).

Source code in client/src/armnet_client/client.py

def get_registry_credentials(self) -> RegistryCredentials:
    """POST /registry/credentials \u2014 fetch fresh image-registry credentials.

    Used by :meth:`armnet_client.Image.push` to get a short-lived
    OAuth2 token for pushing customer images to the platform's
    Artifact Registry repo. The orchestrator returns 503 if it isn't
    configured for credential issuance (typical for local dev).
    """
    resp = self._client.post("/registry/credentials")
    _raise_for_status(resp)
    return RegistryCredentials.model_validate(resp.json())

whoami

whoami() -> WhoAmI

GET /whoami — fetch the username for the current API key.

Source code in client/src/armnet_client/client.py

def whoami(self) -> WhoAmI:
    """GET /whoami — fetch the username for the current API key."""

    resp = self._client.get("/whoami")
    _raise_for_status(resp)
    return WhoAmI.model_validate(resp.json())

list_secrets

list_secrets() -> SecretList

GET /secrets — list secret names for the current user.

Source code in client/src/armnet_client/client.py

def list_secrets(self) -> SecretList:
    """GET /secrets — list secret names for the current user."""

    resp = self._client.get("/secrets")
    _raise_for_status(resp)
    return SecretList.model_validate(resp.json())

create_secret

create_secret(name: str, value: str) -> SecretInfo

POST /secrets — create or replace one secret value.

Source code in client/src/armnet_client/client.py

def create_secret(self, name: str, value: str) -> SecretInfo:
    """POST /secrets — create or replace one secret value."""

    resp = self._client.post(
        "/secrets",
        json=SecretCreateRequest(name=name, value=value).model_dump(mode="json"),
    )
    _raise_for_status(resp)
    return SecretInfo.model_validate(resp.json())

delete_secret

delete_secret(name: str) -> None

DELETE /secrets/{name} — delete one user secret if it exists.

Source code in client/src/armnet_client/client.py

def delete_secret(self, name: str) -> None:
    """DELETE /secrets/{name} — delete one user secret if it exists."""

    resp = self._client.delete(f"/secrets/{name}")
    _raise_for_status(resp)

get_volume_credentials

get_volume_credentials() -> VolumeCredentials

POST /volume/credentials — fetch short-lived GCS volume credentials.

Source code in client/src/armnet_client/client.py

def get_volume_credentials(self) -> VolumeCredentials:
    """POST /volume/credentials — fetch short-lived GCS volume credentials."""

    resp = self._client.post("/volume/credentials")
    _raise_for_status(resp)
    return VolumeCredentials.model_validate(resp.json())

wait

wait(job_id: str, poll_interval_seconds: float = 1.0, deadline_seconds: Optional[float] = None) -> Job

Poll GET /jobs/{id} until the job reaches a terminal state.

Returns the final :class:Job. Raises :class:TimeoutError if deadline_seconds elapses first.

Source code in client/src/armnet_client/client.py

def wait(
    self,
    job_id: str,
    poll_interval_seconds: float = 1.0,
    deadline_seconds: Optional[float] = None,
) -> Job:
    """Poll ``GET /jobs/{id}`` until the job reaches a terminal state.

    Returns the final :class:`Job`. Raises :class:`TimeoutError` if
    ``deadline_seconds`` elapses first.
    """
    start = time.monotonic()
    try:
        while True:
            job = self.get(job_id)
            if job.status in TerminalStatus:
                return job
            if (
                deadline_seconds is not None
                and (time.monotonic() - start) >= deadline_seconds
            ):
                raise TimeoutError(
                    f"Job {job_id} did not reach a terminal state within "
                    f"{deadline_seconds}s (last status: {job.status})."
                )
            time.sleep(poll_interval_seconds)
    except KeyboardInterrupt:
        # Operator intentionally aborted: stop the job now instead of leaving
        # it to the orchestrator's log-disconnect grace window. Best-effort —
        # always re-raise so the interrupt still propagates.
        try:
            self.stop_job(job_id, reason="client interrupted (KeyboardInterrupt)")
        except Exception:  # noqa: BLE001
            pass
        raise

armnet_client.volume

armnet volume helpers (client side).

Thin wrappers over :mod:armnet_core.volume_sync, which holds the shared cloud <-> local transfer engine. This keeps the GCS logic in one place so the cell process (which cannot import armnet-client) reuses the exact same mirroring behaviour. The wrappers add tqdm progress bars for interactive CLI use and preserve the historical function signatures.

upload_to_cloud_volume

upload_to_cloud_volume(*, credentials: VolumeCredentials, username: str, local_path: str | Path, volume_path: str | Path, overwrite: bool = False) -> list[str]

Source code in client/src/armnet_client/volume.py

def upload_to_cloud_volume(
    *,
    credentials: VolumeCredentials,
    username: str,
    local_path: str | Path,
    volume_path: str | Path,
    overwrite: bool = False,
) -> list[str]:
    return volume_sync.upload_path(
        credentials=credentials,
        local_path=local_path,
        volume_path=volume_path,
        overwrite=overwrite,
        local_cache_root=_volume_root(username),
        progress=_default_progress(),
    )

download_from_cloud_volume

download_from_cloud_volume(*, credentials: VolumeCredentials, username: str, volume_path: str | Path, local_path: str | Path) -> Path

Source code in client/src/armnet_client/volume.py

def download_from_cloud_volume(
    *,
    credentials: VolumeCredentials,
    username: str,
    volume_path: str | Path,
    local_path: str | Path,
) -> Path:
    return volume_sync.download_path(
        credentials=credentials,
        volume_path=volume_path,
        local_path=local_path,
        local_cache_root=_volume_root(username),
        progress=_default_progress(),
    )

delete_from_cloud_volume

delete_from_cloud_volume(*, credentials: VolumeCredentials, volume_path: str | Path) -> None

Source code in client/src/armnet_client/volume.py

def delete_from_cloud_volume(*, credentials: VolumeCredentials, volume_path: str | Path) -> None:
    volume_sync.delete_path(credentials=credentials, volume_path=volume_path)

list_cloud_volume

list_cloud_volume(*, credentials: VolumeCredentials, volume_path: str | None = None) -> list[tuple[str, int]]

Source code in client/src/armnet_client/volume.py

def list_cloud_volume(
    *, credentials: VolumeCredentials, volume_path: str | None = None
) -> list[tuple[str, int]]:
    return volume_sync.list_objects_tree(credentials=credentials, volume_path=volume_path)

armnet_client.local

execute_local — in-process execution of a @main script.

Faster-iteration counterpart to :func:execute: no Docker, no orchestrator, no NATS. Loads the customer's Python file directly, finds the @main-decorated function, builds a :class:~armnet_runtime.Context with a user-supplied robot_port, and calls it in the current process.

Returns the same :class:~armnet_core.JobResult shape as :func:execute so downstream code (result.return_value, result.status, etc.) is identical.

Notes for M0.5:

• The LeRobot import-system swap that armnet-runtime will perform in M3+ is intentionally not done here — the whole point of execute_local is to talk to a real local robot driver, not a safety-aware connector.
• Stdout / stderr are not captured (they go to the user's terminal); the corresponding :class:JobResult fields are empty strings.
• timeout_seconds is recorded into ctx.timeout_seconds for the function to honour, but is not enforced by execute_local itself. Killing arbitrary in-process Python by deadline is hard to do cleanly without subprocesses; deferred until we need it.

execute_local

execute_local(main_module: Union[str, Path], *, args: Optional[Mapping[str, Any]] = None, port: Optional[str] = None, robot_id: Optional[str] = None, calibration_dir: Optional[Union[str, Path]] = None, calibration_file_path: Optional[Union[str, Path]] = None, camera_configs: Optional[Mapping[str, Any]] = None, embodiment: str = DEFAULT_EMBODIMENT, task: str = DEFAULT_TASK, timeout_seconds: int = 120, job_id: Optional[str] = None) -> JobResult

Run a @main-decorated script in the current Python process.

:param main_module: Path to the Python file containing exactly one @main-decorated function. :param args: Mapping passed through as ctx.args. :param port: Value placed at ctx.cell.robot_port — typically the path to a real local serial device, e.g. "/dev/ttyUSB0". :param robot_id: Value placed at ctx.cell.robot_id — typically the name of the robot, e.g. "praveen_so101". :param embodiment: Recorded on ctx.embodiment (defaults to the only embodiment supported in M0.5). :param task: Recorded on ctx.task (defaults to the only task supported in M0.5). :param timeout_seconds: Recorded on ctx.timeout_seconds. Not enforced by execute_local (see module docstring). :param job_id: Optional explicit job id for ctx.job_id. Defaults to a stable synthetic id derived from the script path. :returns: A :class:JobResult with return_value set to whatever @main returned, or error populated with a traceback if it raised.

Source code in client/src/armnet_client/local.py

def execute_local(
    main_module: Union[str, Path],
    *,
    args: Optional[Mapping[str, Any]] = None,
    port: Optional[str] = None,
    robot_id: Optional[str] = None,
    calibration_dir: Optional[Union[str, Path]] = None,
    calibration_file_path: Optional[Union[str, Path]] = None,
    camera_configs: Optional[Mapping[str, Any]] = None,
    embodiment: str = DEFAULT_EMBODIMENT,
    task: str = DEFAULT_TASK,
    timeout_seconds: int = 120,
    job_id: Optional[str] = None,
) -> JobResult:
    """Run a ``@main``-decorated script in the current Python process.

    :param main_module: Path to the Python file containing exactly one
        ``@main``-decorated function.
    :param args: Mapping passed through as ``ctx.args``.
    :param port: Value placed at ``ctx.cell.robot_port`` \u2014 typically the path
        to a real local serial device, e.g. ``"/dev/ttyUSB0"``.
    :param robot_id: Value placed at ``ctx.cell.robot_id`` \u2014 typically the
        name of the robot, e.g. ``"praveen_so101"``.
    :param embodiment: Recorded on ``ctx.embodiment`` (defaults to the
        only embodiment supported in M0.5).
    :param task: Recorded on ``ctx.task`` (defaults to the only task
        supported in M0.5).
    :param timeout_seconds: Recorded on ``ctx.timeout_seconds``. Not
        enforced by ``execute_local`` (see module docstring).
    :param job_id: Optional explicit job id for ``ctx.job_id``. Defaults
        to a stable synthetic id derived from the script path.
    :returns: A :class:`JobResult` with ``return_value`` set to whatever
        ``@main`` returned, or ``error`` populated with a traceback if it
        raised.
    """

    Cell, Context, _build_camera_configs, _reset_registry, registered_main = _import_runtime()

    script_path = Path(main_module).resolve()
    if not script_path.is_file():
        raise FileNotFoundError(f"main_module not found: {script_path}")

    embodiment_e = embodiment
    task_e = task
    args_dict = dict(args) if args is not None else {}
    job_id_ = job_id or f"job_local_{script_path.stem}"

    started_at = _utcnow()

    _reset_registry()  # allow successive execute_local calls in one process
    try:
        _load_script(script_path)
    except Exception:
        # Import failure \u2014 there's a Python traceback (think syntax error in
        # the user's script). Mirror what the cell does: stash the traceback
        # in the dedicated field so result.raise_for_status() etc. behave
        # consistently with the remote `execute()` path.
        return JobResult(
            status=JobStatus.FAILED,
            error="failed to import main_module",
            traceback=traceback.format_exc(),
            started_at=started_at,
            finished_at=_utcnow(),
        )

    fn = registered_main()
    if fn is None:
        # Contract violation, not a user-code exception \u2014 no traceback to
        # surface; this stays in `error`.
        return JobResult(
            status=JobStatus.FAILED,
            error=(
                f"no @main-decorated function found in {script_path}. "
                "Decorate exactly one function with `@main` to mark it as the entry point."
            ),
            started_at=started_at,
            finished_at=_utcnow(),
        )

    ctx = Context(
        job_id=job_id_,
        embodiment=embodiment_e,
        task=task_e,
        args=args_dict,
        cell=Cell(
            robot_port=port,
            robot_id=robot_id,
            calibration_dir=Path(calibration_dir) if calibration_dir else None,
            calibration_file_path=Path(calibration_file_path) if calibration_file_path else None,
        ),
        camera_configs=_build_camera_configs(dict(camera_configs or {})),
        timeout_seconds=timeout_seconds,
    )

    try:
        return_value = fn(ctx)
    except Exception:
        return JobResult(
            status=JobStatus.FAILED,
            traceback=traceback.format_exc(),
            started_at=started_at,
            finished_at=_utcnow(),
        )

    return JobResult(
        status=JobStatus.SUCCEEDED,
        exit_code=0,
        return_value=return_value,
        started_at=started_at,
        finished_at=_utcnow(),
    )

armnet_client.local_container

Docker-backed local execution for robot cell development.

LocalContext dataclass

Developer-facing context for local Docker execution.

This is intentionally smaller than RobotCellConfig. It describes the runtime values a containerized program needs, without exposing operator-only cell settings such as NATS, cell IDs, or completion model configuration.

Source code in client/src/armnet_client/local_container.py

@dataclass
class LocalContext:
    """Developer-facing context for local Docker execution.

    This is intentionally smaller than ``RobotCellConfig``. It describes the
    runtime values a containerized program needs, without exposing operator-only
    cell settings such as NATS, cell IDs, or completion model configuration.
    """

    robot_port: str
    camera_configs: Mapping[str, Any] = field(default_factory=dict)
    robot_id: Optional[str] = None
    calibration_dir: Optional[Union[str, Path]] = None
    calibration_file_path: Optional[Union[str, Path]] = None
    language_instruction: Optional[str] = None
    safety_limit: Optional[float] = None
    docker_gpus: Optional[str] = None

robot_port instance-attribute

robot_port: str

camera_configs class-attribute instance-attribute

camera_configs: Mapping[str, Any] = field(default_factory=dict)

robot_id class-attribute instance-attribute

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute

calibration_dir: Optional[Union[str, Path]] = None

calibration_file_path class-attribute instance-attribute

calibration_file_path: Optional[Union[str, Path]] = None

language_instruction class-attribute instance-attribute

language_instruction: Optional[str] = None

safety_limit class-attribute instance-attribute

safety_limit: Optional[float] = None

docker_gpus class-attribute instance-attribute

docker_gpus: Optional[str] = None

execute_local_container

execute_local_container(*, image: Union[str, Image], local_context: Optional[LocalContext] = None, args: Optional[Mapping[str, Any]] = None, embodiment: str = DEFAULT_EMBODIMENT, task: str = DEFAULT_TASK, timeout_seconds: int = 120, cell_socket: Optional[str] = None, cell_endpoint: Optional[str] = None, robot_id: Optional[str] = None, calibration_dir: Optional[Union[str, Path]] = None, calibration_file_path: Optional[Union[str, Path]] = None, cell_config: Optional[RobotCellConfig] = None, cell_config_path: Optional[Union[str, Path]] = None, mount_calibration: bool = True, username: str = 'remoterobo-test', job_id: str = 'job_local_container', stream_output: bool = True, docker_gpus: Optional[str] = None, docker_runtime: Optional[str] = None, gpus: Optional[str] = None) -> JobResult

Run a runtime image locally through Docker.

This is the M2 local-dev path for robot examples: it validates the same container + armnet-runtime + Unix-socket route as remote execution, without going through NATS or the cloud orchestrator.

Source code in client/src/armnet_client/local_container.py

def execute_local_container(
    *,
    image: Union[str, Image],
    local_context: Optional[LocalContext] = None,
    args: Optional[Mapping[str, Any]] = None,
    embodiment: str = DEFAULT_EMBODIMENT,
    task: str = DEFAULT_TASK,
    timeout_seconds: int = 120,
    cell_socket: Optional[str] = None,
    cell_endpoint: Optional[str] = None,
    robot_id: Optional[str] = None,
    calibration_dir: Optional[Union[str, Path]] = None,
    calibration_file_path: Optional[Union[str, Path]] = None,
    cell_config: Optional[RobotCellConfig] = None,
    cell_config_path: Optional[Union[str, Path]] = None,
    mount_calibration: bool = True,
    username: str = "remoterobo-test",
    job_id: str = "job_local_container",
    stream_output: bool = True,
    docker_gpus: Optional[str] = None,
    docker_runtime: Optional[str] = None,
    gpus: Optional[str] = None,
) -> JobResult:
    """Run a runtime image locally through Docker.

    This is the M2 local-dev path for robot examples: it validates the same
    container + armnet-runtime + Unix-socket route as remote execution,
    without going through NATS or the cloud orchestrator.
    """

    started_at = _utcnow()
    embodiment_e = embodiment
    task_e = task
    image_ref = str(image) if isinstance(image, Image) else image
    loaded_cell_config = (
        _cell_config_from_local_context(local_context, embodiment=embodiment_e, task=task_e)
        if local_context is not None
        else _build_cell_config(
            cell_config=cell_config,
            cell_config_path=cell_config_path,
            cell_endpoint=cell_endpoint,
            cell_socket=cell_socket,
            robot_id=robot_id,
            calibration_dir=calibration_dir,
            calibration_file_path=calibration_file_path,
            embodiment=embodiment_e,
            task=task_e,
        )
    )
    docker_gpus = docker_gpus if docker_gpus is not None else (
        local_context.docker_gpus if local_context is not None else None
    )

    with _LocalControlServer(
        robot_port=None if local_context is not None else loaded_cell_config.robot_connector_endpoint,
        use_host_network=local_context is not None,
    ) as control:
        cmd = [
            "docker", "run", "--rm",
            "-e", f"{runtime_env.JOB_ID}={job_id}",
            "-e", f"{runtime_env.EMBODIMENT}={embodiment_e}",
            "-e", f"{runtime_env.TASK}={task_e}",
            "-e", f"{runtime_env.TIMEOUT_SECONDS}={timeout_seconds}",
            "-e", f"{runtime_env.ARGS}={json.dumps(dict(args or {}))}",
            "-e", f"{runtime_env.LOCAL_CONTROL_ENDPOINT}={control.container_endpoint}",
            "-e", f"{runtime_env.LOCAL_CONTAINER}=1",
        ]
        if local_context is not None:
            cmd.extend(["--network", "host"])
            cmd.extend(_direct_hardware_device_args(local_context))
        else:
            cmd.extend(["--add-host", "host.docker.internal:host-gateway"])
        if docker_gpus:
            cmd.extend(_gpu_args(docker_gpus))
        elif docker_runtime:
            cmd.extend(["--runtime", docker_runtime])
            if gpus:
                cmd.extend(["--gpus", gpus])
        cell_config_json = loaded_cell_config.model_dump_json(exclude_none=True)
        cmd.extend(["-e", f"{runtime_env.CELL_CONFIG}={cell_config_json}"])
        endpoint = None if local_context is not None else loaded_cell_config.robot_connector_endpoint
        if endpoint:
            cmd.extend(["-e", f"{runtime_env.CELL_SOCKET}={endpoint}"])
            if not endpoint.startswith("tcp://"):
                socket_path = Path(endpoint)
                socket_dir = str(socket_path.parent)
                cmd.extend([
                    # Mount the parent directory, not the socket file itself.
                    "-v", f"{socket_dir}:{socket_dir}",
                ])
        if loaded_cell_config.robot_id:
            cmd.extend(["-e", f"{runtime_env.ROBOT_ID}={loaded_cell_config.robot_id}"])
        if loaded_cell_config.calibration_dir:
            cmd.extend(["-e", f"{runtime_env.CALIBRATION_DIR}={loaded_cell_config.calibration_dir}"])
        if loaded_cell_config.calibration_file_path:
            cmd.extend([
                "-e",
                f"{runtime_env.CALIBRATION_FILE_PATH}={loaded_cell_config.calibration_file_path}",
            ])
        calibration_mount_path = _cell_calibration_mount_root(loaded_cell_config)
        if mount_calibration and calibration_mount_path:
            cmd.extend(["-v", f"{calibration_mount_path}:{calibration_mount_path}:ro"])
        cache_home = _user_cache_home(username)
        volume_home = _user_volume_home(username)
        cmd.extend([
            "-e", f"{runtime_env.CACHE_HOME}={cache_home}",
            "-e", f"{runtime_env.VOLUME_HOME}={volume_home}",
            # Also set the pre-rename spellings for containers pinned to an
            # older armnet-runtime that still reads them.
            "-e", f"{runtime_env.LEGACY_CACHE_HOME}={cache_home}",
            "-e", f"{runtime_env.LEGACY_VOLUME_HOME}={volume_home}",
            "-e", f"{runtime_env.HF_HOME}={cache_home / 'huggingface'}",
            "-v", f"{cache_home}:{cache_home}",
            "-v", f"{volume_home}:{volume_home}",
        ])
        # Remote-call profiling is off unless the operator asks for it, and the
        # container inherits nothing from this shell, so forward it explicitly.
        call_stats_interval = os.environ.get("ARMNET_REMOTE_CALL_STATS_INTERVAL_S")
        if call_stats_interval:
            cmd.extend([
                "-e",
                f"ARMNET_REMOTE_CALL_STATS_INTERVAL_S={call_stats_interval}",
            ])
        cmd.append(image_ref)

        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
        )
        assert proc.stdout is not None
        assert proc.stderr is not None

        stdout_q: Queue[str] = Queue()
        stderr_q: Queue[str] = Queue()
        stdout_thread = threading.Thread(
            target=_pump_stream,
            args=(proc.stdout, stdout_q, sys.stdout if stream_output else None),
            daemon=True,
        )
        stderr_thread = threading.Thread(
            target=_pump_stream,
            args=(proc.stderr, stderr_q, sys.stderr if stream_output else None),
            daemon=True,
        )
        stdout_thread.start()
        stderr_thread.start()

        try:
            proc.wait(timeout=timeout_seconds)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()
            stdout_thread.join(timeout=1)
            stderr_thread.join(timeout=1)
            stdout = _queue_to_text(stdout_q)
            stderr = _queue_to_text(stderr_q)
            return JobResult(
                status=JobStatus.TIMEOUT,
                error=f"local container exceeded timeout_seconds={timeout_seconds}",
                stdout=stdout,
                stderr=stderr,
                started_at=started_at,
                finished_at=_utcnow(),
            )

        stdout_thread.join(timeout=1)
        stderr_thread.join(timeout=1)
        stdout = _queue_to_text(stdout_q)
        stderr = _queue_to_text(stderr_q)
        return JobResult(
            status=JobStatus.SUCCEEDED if proc.returncode == 0 else JobStatus.FAILED,
            exit_code=proc.returncode,
            stdout=stdout,
            stderr=stderr,
            return_value=_extract_marker(stdout, RESULT_MARKER_JSON),
            traceback=_extract_marker(stdout, TRACEBACK_MARKER_JSON),
            started_at=started_at,
            finished_at=_utcnow(),
        )

armnet_client.rerun_stream

Client-side Rerun streaming.

Connects to the orchestrator's /jobs/{job_id}/rerun websocket, decodes the binary RerunPacket frames the cell republished from the container's ctx.log_rerun_data(...) calls, and replays each into the active Rerun recording via rr.log(...).

Call :func:init_rerun (or rerun.init(...) directly) once in your orchestrate script before submitting the job so the packets land in a live viewer.

logger module-attribute

logger = logging.getLogger(__name__)

init_rerun

init_rerun(application_id: str = 'armnet', *, spawn: bool = True, serve_web: bool = False, web_port: Optional[int] = None, grpc_port: Optional[int] = None, open_browser: bool = True, flush_tick_seconds: Optional[float] = None) -> Any

Initialize Rerun and start a viewer.

Thin convenience wrapper so orchestrate scripts don't need to import the Rerun SDK directly. Raises a clear error if rerun-sdk isn't installed (pip install 'armnet-client[viz]').

Viewer selection:

• serve_web=True always hosts the web viewer over HTTP (open the printed URL in a browser; forward the port first if you're on a remote host).
• spawn=True (default) opens a native desktop window — but only when a display is available. On a headless machine (no DISPLAY / Wayland, e.g. SSH or a GUI-less workstation) it automatically falls back to the web viewer instead of failing to open a window.

flush_tick_seconds sets Rerun's micro-batching flush interval (the RERUN_FLUSH_TICK_SECS knob). A very small value (e.g. 0.002 = 2ms) minimizes how long logged data waits before being sent to the viewer — noticeably snappier for teleoperation. Left unset, Rerun's default applies.

Source code in client/src/armnet_client/rerun_stream.py

def init_rerun(
    application_id: str = "armnet",
    *,
    spawn: bool = True,
    serve_web: bool = False,
    web_port: Optional[int] = None,
    grpc_port: Optional[int] = None,
    open_browser: bool = True,
    flush_tick_seconds: Optional[float] = None,
) -> Any:
    """Initialize Rerun and start a viewer.

    Thin convenience wrapper so orchestrate scripts don't need to import the
    Rerun SDK directly. Raises a clear error if ``rerun-sdk`` isn't installed
    (``pip install 'armnet-client[viz]'``).

    Viewer selection:

    - ``serve_web=True`` always hosts the **web viewer** over HTTP (open the
      printed URL in a browser; forward the port first if you're on a remote
      host).
    - ``spawn=True`` (default) opens a native desktop window — but only when a
      display is available. On a **headless** machine (no ``DISPLAY`` /
      Wayland, e.g. SSH or a GUI-less workstation) it automatically falls back
      to the web viewer instead of failing to open a window.

    ``flush_tick_seconds`` sets Rerun's micro-batching flush interval (the
    ``RERUN_FLUSH_TICK_SECS`` knob). A very small value (e.g. ``0.002`` = 2ms)
    minimizes how long logged data waits before being sent to the viewer —
    noticeably snappier for teleoperation. Left unset, Rerun's default applies.
    """

    try:
        import rerun as rr
    except ImportError as exc:  # pragma: no cover
        raise RuntimeError(
            "rerun-sdk is required for visualization; install with "
            "`pip install 'armnet-client[viz]'`"
        ) from exc

    if flush_tick_seconds is not None:
        from datetime import timedelta

        # make_default=True routes the module-level rr.log/serve/spawn calls
        # (and the replay in log_packet) through this recording, so the small
        # flush tick actually applies to the streamed data.
        rr.RecordingStream(
            application_id,
            make_default=True,
            batcher_config=rr.ChunkBatcherConfig(
                flush_tick=timedelta(seconds=flush_tick_seconds)
            ),
        )
    else:
        rr.init(application_id)

    use_web = serve_web or (spawn and not _has_display())
    if use_web:
        if spawn and not serve_web:
            logger.warning(
                "no display detected; serving the Rerun web viewer instead of "
                "spawning a native window"
            )
        web = web_port or _DEFAULT_WEB_VIEWER_PORT
        # serve_grpc starts the data server (the log sink + buffer) and returns
        # its URI; serve_web_viewer hosts the HTML viewer that connects to it.
        # (rr.serve_web bundles both but is deprecated, and the viewer/data live
        # on separate ports either way.)
        server_uri = rr.serve_grpc(grpc_port=grpc_port)
        rr.serve_web_viewer(
            web_port=web,
            open_browser=open_browser and _has_display(),
            connect_to=server_uri,
        )
        grpc = _port_from_uri(server_uri) or grpc_port or _DEFAULT_GRPC_PORT
        encoded_uri = server_uri.replace("+", "%2B")
        viewer_url = f"http://localhost:{web}/?url={encoded_uri}"
        print(
            "[armnet] Rerun viewer ready:\n"
            f"  web viewer:  {viewer_url}\n"
            f"  or native:   rerun --connect {server_uri}\n"
            "  on a remote host, forward BOTH ports first (the browser viewer "
            "loads from the web port but streams data from the gRPC port):\n"
            f"      ssh -L {web}:localhost:{web} -L {grpc}:localhost:{grpc} <user>@<host>",
            flush=True,
        )
    elif spawn:
        rr.spawn()
    return rr

stream_job_rerun

stream_job_rerun(job_id: str, stop: Event) -> None

Consume the rerun websocket for job_id until stop is set.

Intended to run on a background thread. Silently returns if no API key is configured, the websocket can't be reached, or the rerun SDK is missing, so visualization never blocks or fails the job itself.

Source code in client/src/armnet_client/rerun_stream.py

def stream_job_rerun(job_id: str, stop: threading.Event) -> None:
    """Consume the rerun websocket for ``job_id`` until ``stop`` is set.

    Intended to run on a background thread. Silently returns if no API key is
    configured, the websocket can't be reached, or the rerun SDK is missing, so
    visualization never blocks or fails the job itself.
    """

    key = api_key()
    if not key:
        return

    try:
        from armnet_runtime.rerun import decode_packet, log_packet
    except ImportError as exc:  # pragma: no cover
        logger.debug("rerun wire helpers unavailable: %r", exc)
        return

    from armnet_client.execute import _websocket_sslopt, _ws_url

    ws_url = _ws_url(orchestrator_url(), f"/jobs/{job_id}/rerun")
    try:
        ws = websocket.create_connection(
            ws_url,
            header=[f"{API_KEY_HEADER}: {key}"],
            timeout=5,
            sslopt=_websocket_sslopt(),
        )
    except Exception as exc:  # noqa: BLE001
        logger.debug("failed to connect rerun websocket for job %s: %r", job_id, exc)
        return

    try:
        ws.settimeout(0.2)
        while not stop.is_set():
            try:
                frame = ws.recv()
            except (TimeoutError, websocket.WebSocketTimeoutException):
                continue
            except Exception:
                break
            if not frame:
                continue
            if isinstance(frame, bytes):
                try:
                    log_packet(decode_packet(frame))
                    # print(f"Total latency: {time.time() - packet.obs_ts} seconds", flush=True)
                except Exception:  # noqa: BLE001 - never let a bad packet kill the stream
                    logger.exception("failed to replay rerun packet for job %s", job_id)
                continue
            # Text frame: control message (e.g. terminal).
            try:
                payload = json.loads(frame)
            except json.JSONDecodeError:
                continue
            if payload.get("type") == "terminal":
                break
    finally:
        stop.set()
        try:
            ws.close()
        except Exception:
            pass

armnet_client.teleop_stream

Client-side remote teleoperation.

Connects to a locally-attached LeRobot leader arm (e.g. an SO-101 leader), samples its joint positions at a high rate, and pushes each reading to the orchestrator's /jobs/{job_id}/teleop websocket. The orchestrator forwards them to the job's cell, which keeps only the freshest action and serves it to the runtime container (ctx.cell.get_teleop_action()), which drives the remote follower.

Sampling faster than the robot's control rate (default 60Hz vs a ~20Hz robot loop) makes teleop feel snappier: the cell always has a near-current command, so a dropped or late frame doesn't force the follower to wait a full robot tick for the next one.

When recording a dataset (armnet-lerobot-record), discrete recording-control events ride the same websocket: pass an events queue to :func:stream_job_teleop and feed it from :func:start_recording_key_listener, which binds LeRobot's standard dataset recording shortcuts (Right Arrow → save episode and move on, Left Arrow → discard and re-record, Esc → stop the session).

LeRobot is an optional dependency — install with armnet-client[teleop]. It is imported lazily so the rest of the client works without it.

logger module-attribute

logger = logging.getLogger(__name__)

TeleoperatorConfig dataclass

How to connect to the local leader arm being used to teleoperate.

Mirrors what LeRobot needs to construct a teleoperator, plus the sample rate. id is the LeRobot teleoperator id used to find calibration.

Source code in client/src/armnet_client/teleop_stream.py

@dataclass
class TeleoperatorConfig:
    """How to connect to the local leader arm being used to teleoperate.

    Mirrors what LeRobot needs to construct a teleoperator, plus the sample
    rate. ``id`` is the LeRobot teleoperator id used to find calibration.
    """

    port: Optional[str]
    id: Optional[str] = None
    left_port: Optional[str] = None
    right_port: Optional[str] = None
    left_id: Optional[str] = None
    right_id: Optional[str] = None
    embodiment: str = "lerobot/so-101"
    use_degrees: bool = True
    calibration_dir: Optional[str] = None
    # Re-run interactive calibration on connect. Off by default: a leader used
    # for teleop is normally already calibrated.
    calibrate: bool = False
    # Sample rate (Hz). Intentionally higher than the robot loop for snappiness.
    fps: float = _DEFAULT_TELEOP_FPS
    # Named plugin profile supplies the identified hardware defaults. None on
    # an individual field means "inherit from the profile"; explicit values
    # remain available for new GELLO bring-up and diagnostics.
    yam_gravity_profile: str = "trlc_dk1_v1"
    yam_left_gravity_profile: Optional[str] = None
    yam_right_gravity_profile: Optional[str] = None
    yam_gravity_assist: Optional[bool] = None
    yam_gravity_assist_gain: Optional[float] = None
    yam_gravity_current_limit_ma: Optional[int] = None
    yam_gravity_joint_signs: Optional[tuple[int, ...]] = None
    yam_gravity_joint_offsets: Optional[tuple[float, ...]] = None
    # +1 if the GELLO URDF's z axis points physically up when mounted, -1 if
    # down. The bundled URDF is z-down; the wrong value inverts every torque.
    yam_gravity_z_sign: Optional[int] = None
    # Per-link mass override in kg, base to tip (7 values), for leaders built
    # differently from the bundled URDF (e.g. TRLC-DK1). Empty = URDF masses.
    yam_gravity_link_masses: Optional[tuple[float, ...]] = None
    # Restrict assist to these arm joints (empty = all six). Lets validated
    # joints go live while unvalidated ones stay passive.
    yam_gravity_joints: Optional[tuple[str, ...]] = None
    # Per-joint physical ranges in radians, 12 values (lo/hi per joint), for
    # leaders whose joints travel less than the YAM software range (e.g. the
    # DK1 lift). Empty = plugin defaults. Offsets are only valid together with
    # the ranges they were captured with.
    yam_gravity_joint_ranges: Optional[tuple[float, ...]] = None
    # Log computed assist currents without applying torque (safe tuning mode).
    yam_gravity_dry_run: bool = False
    yam_gripper_return: Optional[bool] = None
    yam_gripper_return_current_ma: Optional[int] = None
    yam_assist_side: str = "both"

port instance-attribute

port: Optional[str]

id class-attribute instance-attribute

id: Optional[str] = None

left_port class-attribute instance-attribute

left_port: Optional[str] = None

right_port class-attribute instance-attribute

right_port: Optional[str] = None

left_id class-attribute instance-attribute

left_id: Optional[str] = None

right_id class-attribute instance-attribute

right_id: Optional[str] = None

embodiment class-attribute instance-attribute

embodiment: str = 'lerobot/so-101'

use_degrees class-attribute instance-attribute

use_degrees: bool = True

calibration_dir class-attribute instance-attribute

calibration_dir: Optional[str] = None

calibrate class-attribute instance-attribute

calibrate: bool = False

fps class-attribute instance-attribute

fps: float = _DEFAULT_TELEOP_FPS

yam_gravity_profile class-attribute instance-attribute

yam_gravity_profile: str = 'trlc_dk1_v1'

yam_left_gravity_profile class-attribute instance-attribute

yam_left_gravity_profile: Optional[str] = None

yam_right_gravity_profile class-attribute instance-attribute

yam_right_gravity_profile: Optional[str] = None

yam_gravity_assist class-attribute instance-attribute

yam_gravity_assist: Optional[bool] = None

yam_gravity_assist_gain class-attribute instance-attribute

yam_gravity_assist_gain: Optional[float] = None

yam_gravity_current_limit_ma class-attribute instance-attribute

yam_gravity_current_limit_ma: Optional[int] = None

yam_gravity_joint_signs class-attribute instance-attribute

yam_gravity_joint_signs: Optional[tuple[int, ...]] = None

yam_gravity_joint_offsets class-attribute instance-attribute

yam_gravity_joint_offsets: Optional[tuple[float, ...]] = None

yam_gravity_z_sign class-attribute instance-attribute

yam_gravity_z_sign: Optional[int] = None

yam_gravity_link_masses class-attribute instance-attribute

yam_gravity_link_masses: Optional[tuple[float, ...]] = None

yam_gravity_joints class-attribute instance-attribute

yam_gravity_joints: Optional[tuple[str, ...]] = None

yam_gravity_joint_ranges class-attribute instance-attribute

yam_gravity_joint_ranges: Optional[tuple[float, ...]] = None

yam_gravity_dry_run class-attribute instance-attribute

yam_gravity_dry_run: bool = False

yam_gripper_return class-attribute instance-attribute

yam_gripper_return: Optional[bool] = None

yam_gripper_return_current_ma class-attribute instance-attribute

yam_gripper_return_current_ma: Optional[int] = None

yam_assist_side class-attribute instance-attribute

yam_assist_side: str = 'both'

start_recording_key_listener

start_recording_key_listener(events: 'queue.Queue[str]')

Bind LeRobot's dataset-recording keyboard shortcuts to events.

Right Arrow → next_episode, Left Arrow → rerecord_episode, Esc → stop_recording. Each press is enqueued for :func:stream_job_teleop to forward to the cell. Esc does not tear anything down client-side: the runtime ends the session on receipt and the job result unblocks the orchestrate script, which then stops its streams.

Linux reads the kernel input device via evdev, because pynput only hears keys through X11 and a Wayland session never delivers them. macOS keeps pynput. Returns the started listener (call .stop() when done), or None when no keyboard is available.

Source code in client/src/armnet_client/teleop_stream.py

def start_recording_key_listener(events: "queue.Queue[str]"):
    """Bind LeRobot's dataset-recording keyboard shortcuts to ``events``.

    Right Arrow → ``next_episode``, Left Arrow → ``rerecord_episode``, Esc →
    ``stop_recording``. Each press is enqueued for :func:`stream_job_teleop`
    to forward to the cell. Esc does not tear anything down client-side: the
    runtime ends the session on receipt and the job result unblocks the
    orchestrate script, which then stops its streams.

    Linux reads the kernel input device via ``evdev``, because ``pynput`` only
    hears keys through X11 and a Wayland session never delivers them. macOS
    keeps ``pynput``. Returns the started listener (call ``.stop()`` when
    done), or None when no keyboard is available.
    """

    if sys.platform == "linux":
        return _start_evdev_key_listener(events)
    return _start_pynput_key_listener(events)

stream_job_teleop

stream_job_teleop(job_id: str, stop: Event, config: TeleoperatorConfig, events: 'queue.Queue[str] | None' = None) -> None

Sample the local leader and push actions to job_id until stop is set.

Intended to run on a background thread (alongside the rerun stream). Returns quietly if no API key is configured, LeRobot is missing, or the leader/ websocket can't be reached, so teleop never crashes the orchestrate script.

events, when given, is a queue of recording-control event strings (see :func:start_recording_key_listener); each is sent on the same websocket as a :class:TeleopEvent frame. Unlike action frames — which are freshest-wins and freely dropped — a queued event survives websocket reconnects: it is re-queued on a failed send and retried.

Source code in client/src/armnet_client/teleop_stream.py

def stream_job_teleop(
    job_id: str,
    stop: threading.Event,
    config: TeleoperatorConfig,
    events: "queue.Queue[str] | None" = None,
) -> None:
    """Sample the local leader and push actions to ``job_id`` until ``stop`` is set.

    Intended to run on a background thread (alongside the rerun stream). Returns
    quietly if no API key is configured, LeRobot is missing, or the leader/
    websocket can't be reached, so teleop never crashes the orchestrate script.

    ``events``, when given, is a queue of recording-control event strings (see
    :func:`start_recording_key_listener`); each is sent on the same websocket
    as a :class:`TeleopEvent` frame. Unlike action frames — which are
    freshest-wins and freely dropped — a queued event survives websocket
    reconnects: it is re-queued on a failed send and retried.
    """

    key = api_key()
    if not key:
        logger.warning("no API key configured; remote teleop disabled")
        return

    try:
        leader = _make_leader(config)
    except ImportError:
        logger.error(
            "remote teleop requires LeRobot; install with `pip install 'armnet-client[teleop]'`"
        )
        return
    except Exception as exc:  # noqa: BLE001
        logger.error("failed to construct teleoperator: %r", exc)
        return

    # Imported here so _ws_url's import cost is only paid when teleoperating.
    from armnet_client.execute import _websocket_sslopt, _ws_url

    ws_url = _ws_url(orchestrator_url(), f"/jobs/{job_id}/teleop")

    try:
        leader.connect(calibrate=config.calibrate)
    except Exception as exc:  # noqa: BLE001
        logger.error("failed to connect leader arm on %s: %r", config.port, exc)
        return

    period_s = 1.0 / config.fps if config.fps > 0 else 0.0
    previous_action: dict[str, float] = {}
    try:
        # Reconnect loop: a dropped teleop websocket self-heals and never sets
        # the shared stop, so it can't take down the (separate) rerun stream.
        while not stop.is_set():
            try:
                ws = websocket.create_connection(
                    ws_url,
                    header=[f"{API_KEY_HEADER}: {key}"],
                    timeout=5,
                    sslopt=_websocket_sslopt(),
                )
                # Short timeout: send returns fast, and the periodic recv used to
                # service keepalives/pings stays near-non-blocking.
                ws.settimeout(0.01)
            except Exception as exc:  # noqa: BLE001
                logger.warning("teleop websocket connect failed (retrying): %r", exc)
                if stop.wait(1.0):
                    break
                continue

            logger.info("remote teleop streaming to job %s at %.0f Hz", job_id, config.fps)
            last_drain = time.perf_counter()
            try:
                while not stop.is_set():
                    loop_start = time.perf_counter()
                    try:
                        raw_action = leader.get_action()
                    except Exception:  # noqa: BLE001 - transient read; keep going
                        logger.exception("leader get_action failed")
                        continue
                    action = _finite_action(raw_action, previous_action)
                    previous_action.update(action)
                    message = TeleopAction(
                        job_id=job_id,
                        timestamp=time.time(),
                        action=action,
                    )
                    try:
                        ws.send(message.model_dump_json())
                    except websocket.WebSocketTimeoutException:
                        pass  # transient send-buffer backpressure; drop this sample
                    except Exception:  # noqa: BLE001 - dropped; reconnect
                        logger.warning("teleop websocket send failed; reconnecting", exc_info=True)
                        break
                    if events is not None and not _send_pending_events(ws, job_id, events):
                        break  # ws dropped; unsent events were re-queued
                    # Periodically read so the client services server keepalives
                    # and answers WebSocket pings (a send-only client never would,
                    # and the server would eventually close the connection).
                    now = time.perf_counter()
                    if now - last_drain >= 0.5:
                        last_drain = now
                        try:
                            ws.recv()
                        except websocket.WebSocketTimeoutException:
                            pass
                        except Exception:  # noqa: BLE001 - dropped; reconnect
                            logger.warning("teleop websocket recv failed; reconnecting", exc_info=True)
                            break
                    remaining = period_s - (time.perf_counter() - loop_start)
                    if remaining > 0:
                        time.sleep(remaining)
            finally:
                try:
                    ws.close()
                except Exception:  # noqa: BLE001
                    pass
    finally:
        try:
            leader.disconnect()
        except Exception:  # noqa: BLE001
            pass

armnet_client.current_gate

Current-gate profiles derived from robot-telemetry sidecars.

CURRENT_BIT module-attribute

CURRENT_BIT = 1 << LIVE_GROUP_BITS['current']

DEFAULT_PERCENTILES module-attribute

DEFAULT_PERCENTILES = (95, 99)

DEFAULT_GATE_PERCENTILE module-attribute

DEFAULT_GATE_PERCENTILE = 99

DEFAULT_GATE_PERCENT_ABOVE module-attribute

DEFAULT_GATE_PERCENT_ABOVE = 0.0

DEFAULT_GATE_DURATION_MS module-attribute

DEFAULT_GATE_DURATION_MS = 200.0

TINY_MAX_THRESHOLD module-attribute

TINY_MAX_THRESHOLD = 15

DEFAULT_PROFILE_PATH module-attribute

DEFAULT_PROFILE_PATH = Path(str(resources.files('armnet_client').joinpath('current_gate_profile.json')))

DEFAULT_DATASETS_PATH module-attribute

DEFAULT_DATASETS_PATH = Path(str(resources.files('armnet_client').joinpath('current_gate_datasets.json')))

collect_current_samples

collect_current_samples(table: Any) -> dict[tuple[str, str], list[int]]

Source code in client/src/armnet_client/current_gate.py

def collect_current_samples(table: Any) -> dict[tuple[str, str], list[int]]:
    samples: dict[tuple[str, str], list[int]] = defaultdict(list)
    motors: set[str] = set()
    for column in table.column_names:
        if column.endswith(f".{_CURRENT_SUFFIX}"):
            motors.add(column[: -len(_CURRENT_SUFFIX) - 1])
    for motor in sorted(motors):
        parsed = _parse_motor_name(motor)
        if parsed is None:
            continue
        arm, joint = parsed
        current_col = f"{motor}.{_CURRENT_SUFFIX}"
        valid_col = f"{motor}.{_VALID_SUFFIX}"
        fresh_col = f"{motor}.{_FRESH_SUFFIX}"
        if valid_col not in table.column_names or fresh_col not in table.column_names:
            continue
        currents = table.column(current_col).to_pylist()
        valids = table.column(valid_col).to_pylist()
        freshes = table.column(fresh_col).to_pylist()
        for current, valid, fresh in zip(currents, valids, freshes, strict=True):
            if not isinstance(current, int) or isinstance(current, bool):
                continue
            if not isinstance(valid, int) or not isinstance(fresh, int):
                continue
            if (valid & CURRENT_BIT) == 0 or (fresh & CURRENT_BIT) == 0:
                continue
            samples[(arm, joint)].append(current)
    return dict(samples)

compute_percentile_wire_counts

compute_percentile_wire_counts(samples: Mapping[tuple[str, str], Sequence[int]], percentiles: Sequence[int] = DEFAULT_PERCENTILES) -> dict[int, dict[str, dict[str, int]]]

Source code in client/src/armnet_client/current_gate.py

def compute_percentile_wire_counts(
    samples: Mapping[tuple[str, str], Sequence[int]],
    percentiles: Sequence[int] = DEFAULT_PERCENTILES,
) -> dict[int, dict[str, dict[str, int]]]:
    result: dict[int, dict[str, dict[str, int]]] = {
        int(p): defaultdict(dict) for p in percentiles
    }
    for (arm, joint), values in samples.items():
        if not values:
            continue
        ordered = sorted(int(v) for v in values)
        for percentile in percentiles:
            result[int(percentile)][arm][joint] = _nearest_rank_percentile(
                ordered,
                float(percentile),
            )
    return {
        percentile: {arm: dict(joints) for arm, joints in arms.items()}
        for percentile, arms in result.items()
    }

derive_profile_from_dataset_roots

derive_profile_from_dataset_roots(dataset_roots: Sequence[Path], sources: Sequence[str], percentiles: Sequence[int] = DEFAULT_PERCENTILES) -> dict[str, Any]

Source code in client/src/armnet_client/current_gate.py

def derive_profile_from_dataset_roots(
    dataset_roots: Sequence[Path],
    sources: Sequence[str],
    percentiles: Sequence[int] = DEFAULT_PERCENTILES,
) -> dict[str, Any]:
    if not dataset_roots:
        raise ValueError("at least one dataset root is required")
    if len(dataset_roots) != len(sources):
        raise ValueError("sources must align with dataset_roots")
    merged: dict[tuple[str, str], list[int]] = defaultdict(list)
    for root in dataset_roots:
        table = _sidecar_table(Path(root))
        if table is None:
            raise ValueError(f"no robot telemetry sidecar frames under {root}")
        for key, values in collect_current_samples(table).items():
            merged[key].extend(values)
    if not merged:
        raise ValueError("no valid/fresh current samples found in datasets")
    percentile_tables = compute_percentile_wire_counts(merged, percentiles=percentiles)
    maxima: dict[str, dict[str, int]] = defaultdict(dict)
    for (arm, joint), values in merged.items():
        maxima[arm][joint] = max(int(v) for v in values)
    return {
        "sources": list(sources),
        "sample_counts": {
            f"{arm}.{joint}": len(values)
            for (arm, joint), values in sorted(merged.items())
        },
        "percentiles": {
            str(percentile): arms for percentile, arms in percentile_tables.items()
        },
        "max": {arm: dict(joints) for arm, joints in sorted(maxima.items())},
    }

download_dataset_root

download_dataset_root(repo_id: str, *, cache_dir: Path | None = None) -> Path

Source code in client/src/armnet_client/current_gate.py

def download_dataset_root(repo_id: str, *, cache_dir: Path | None = None) -> Path:
    try:
        from huggingface_hub import snapshot_download
    except ImportError as exc:  # pragma: no cover - env dependent
        raise ImportError(
            "huggingface_hub is required to download datasets for current-gate "
            "derivation; install 'armnet-client[current-gate]'"
        ) from exc
    kwargs: dict[str, Any] = {
        "repo_id": repo_id,
        "repo_type": "dataset",
        "allow_patterns": [
            "candywrapper_plugins/robot_telemetry/frames/chunk-*/*.parquet",
            "candywrapper_plugins/robot_telemetry/**",
        ],
    }
    if cache_dir is not None:
        kwargs["cache_dir"] = str(cache_dir)
    return Path(snapshot_download(**kwargs))

derive_profile_from_repo_ids

derive_profile_from_repo_ids(repo_ids: Sequence[str], *, cache_dir: Path | None = None, percentiles: Sequence[int] = DEFAULT_PERCENTILES) -> dict[str, Any]

Source code in client/src/armnet_client/current_gate.py

def derive_profile_from_repo_ids(
    repo_ids: Sequence[str],
    *,
    cache_dir: Path | None = None,
    percentiles: Sequence[int] = DEFAULT_PERCENTILES,
) -> dict[str, Any]:
    if not repo_ids:
        raise ValueError("datasets JSON must contain at least one repo id")
    roots = [download_dataset_root(repo_id, cache_dir=cache_dir) for repo_id in repo_ids]
    return derive_profile_from_dataset_roots(
        roots,
        sources=list(repo_ids),
        percentiles=percentiles,
    )

load_profile

load_profile(path: Path | None = None) -> dict[str, Any]

Source code in client/src/armnet_client/current_gate.py

def load_profile(path: Path | None = None) -> dict[str, Any]:
    profile_path = Path(path) if path is not None else DEFAULT_PROFILE_PATH
    with profile_path.open(encoding="utf-8") as handle:
        profile = json.load(handle)
    if not isinstance(profile, dict) or "percentiles" not in profile:
        raise ValueError(f"invalid current gate profile: {profile_path}")
    return profile

write_profile

write_profile(path: Path, profile: Mapping[str, Any]) -> None

Source code in client/src/armnet_client/current_gate.py

def write_profile(path: Path, profile: Mapping[str, Any]) -> None:
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as handle:
        json.dump(profile, handle, indent=2, sort_keys=True)
        handle.write("\n")

reference_wire_counts_from_profile

reference_wire_counts_from_profile(profile: Mapping[str, Any], percentile: int) -> dict[str, dict[str, int]]

Source code in client/src/armnet_client/current_gate.py

def reference_wire_counts_from_profile(
    profile: Mapping[str, Any],
    percentile: int,
) -> dict[str, dict[str, int]]:
    percentiles = profile.get("percentiles")
    if not isinstance(percentiles, Mapping):
        raise ValueError("current gate profile missing percentiles")
    key = str(int(percentile))
    if key not in percentiles:
        raise ValueError("current gate percentile must be 95 or 99")
    raw = percentiles[key]
    if not isinstance(raw, Mapping):
        raise ValueError(f"invalid percentile table for p{percentile}")
    return _arm_joint_int_table(raw)

max_wire_counts_from_profile

max_wire_counts_from_profile(profile: Mapping[str, Any]) -> dict[str, dict[str, int]]

Source code in client/src/armnet_client/current_gate.py

def max_wire_counts_from_profile(profile: Mapping[str, Any]) -> dict[str, dict[str, int]]:
    raw = profile.get("max")
    if raw is None:
        return {}
    if not isinstance(raw, Mapping):
        raise ValueError("current gate profile max must be an object")
    return _arm_joint_int_table(raw)

current_gate_config

current_gate_config(percentile: int = DEFAULT_GATE_PERCENTILE, percent_above: float | None = DEFAULT_GATE_PERCENT_ABOVE, duration_ms: float = DEFAULT_GATE_DURATION_MS, *, profile_path: Path | None = None) -> dict | None

Build the job-arg payload for the edge current gate.

Joints whose baseline max is at or below TINY_MAX_THRESHOLD use that max as the trip limit instead. Pass percent_above <= 0 (or None) to omit the optional job-level percentile tightening; the cell-owned edge gate remains active.

Source code in client/src/armnet_client/current_gate.py

def current_gate_config(
    percentile: int = DEFAULT_GATE_PERCENTILE,
    percent_above: float | None = DEFAULT_GATE_PERCENT_ABOVE,
    duration_ms: float = DEFAULT_GATE_DURATION_MS,
    *,
    profile_path: Path | None = None,
) -> dict | None:
    """Build the job-arg payload for the edge current gate.

    Joints whose baseline max is at or below ``TINY_MAX_THRESHOLD`` use that
    max as the trip limit instead. Pass ``percent_above <= 0`` (or ``None``)
    to omit the optional job-level percentile tightening; the cell-owned edge
    gate remains active.
    """
    if percent_above is None or float(percent_above) <= 0.0:
        return None
    profile = load_profile(profile_path)
    percent = float(percent_above)
    duration = float(duration_ms)
    if duration <= 0.0:
        raise ValueError("current gate duration must be greater than zero")
    return {
        "enabled": True,
        "mode": "percentile",
        "percentile": int(percentile),
        "percent_above": percent,
        "duration_ms": duration,
        "tiny_max_threshold": TINY_MAX_THRESHOLD,
        "reference_wire_counts": reference_wire_counts_from_profile(
            profile,
            int(percentile),
        ),
        "max_wire_counts": max_wire_counts_from_profile(profile),
    }

current_gate_job_args

current_gate_job_args(percentile: int = DEFAULT_GATE_PERCENTILE, percent_above: float | None = DEFAULT_GATE_PERCENT_ABOVE, duration_ms: float = DEFAULT_GATE_DURATION_MS, *, profile_path: Path | None = None) -> dict[str, dict]

Source code in client/src/armnet_client/current_gate.py

def current_gate_job_args(
    percentile: int = DEFAULT_GATE_PERCENTILE,
    percent_above: float | None = DEFAULT_GATE_PERCENT_ABOVE,
    duration_ms: float = DEFAULT_GATE_DURATION_MS,
    *,
    profile_path: Path | None = None,
) -> dict[str, dict]:
    cfg = current_gate_config(
        percentile,
        percent_above,
        duration_ms,
        profile_path=profile_path,
    )
    return {"current_gate": cfg} if cfg is not None else {}

main

main(argv: Iterable[str] | None = None) -> int

Source code in client/src/armnet_client/current_gate.py

def main(argv: Iterable[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Derive a current-gate profile JSON from robot-telemetry sidecars "
            "in one or more Hugging Face datasets."
        )
    )
    parser.add_argument(
        "--datasets",
        type=Path,
        default=DEFAULT_DATASETS_PATH,
        help=(
            "JSON file containing a list of Hugging Face dataset repo ids "
            f"(default: {DEFAULT_DATASETS_PATH})"
        ),
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=DEFAULT_PROFILE_PATH,
        help=f"Profile JSON output path (default: {DEFAULT_PROFILE_PATH})",
    )
    parser.add_argument(
        "--cache-dir",
        type=Path,
        default=None,
        help="Optional Hugging Face cache directory",
    )
    args = parser.parse_args(list(argv) if argv is not None else None)
    repo_ids = _load_repo_ids(args.datasets)
    profile = derive_profile_from_repo_ids(repo_ids, cache_dir=args.cache_dir)
    write_profile(args.output, profile)
    print(f"wrote current-gate profile for {len(repo_ids)} dataset(s) to {args.output}")
    return 0