Skip to content

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.

No auth in M0.5 (design doc explicitly defers it). When auth lands we'll add an api_key arg here and inject it as a header.

Source code in client/src/armnet_client/client.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class OrchestratorClient:
    """Synchronous client for the orchestrator HTTP API.

    No auth in M0.5 (design doc explicitly defers it). When auth lands
    we'll add an ``api_key`` arg here and inject it as a header.
    """

    def __init__(
        self,
        base_url: str = "http://localhost:8000",
        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.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) -> list[Job]:
        """GET /jobs — the current user's most-recent jobs (newest first)."""
        resp = self._client.get("/jobs", params={"limit": limit})
        _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 
69
70
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 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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 
 96
 97
 98
 99
100
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) -> list[Job]

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

Source code in client/src/armnet_client/client.py 
102
103
104
105
106
def list_jobs(self, *, limit: int = 10) -> list[Job]:
    """GET /jobs — the current user's most-recent jobs (newest first)."""
    resp = self._client.get("/jobs", params={"limit": limit})
    _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 
108
109
110
111
112
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 
114
115
116
117
118
119
120
121
122
123
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 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
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 
140
141
142
143
144
145
146
147
148
149
150
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 
152
153
154
155
156
157
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 
159
160
161
162
163
164
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 
166
167
168
169
170
171
172
173
174
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 
176
177
178
179
180
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 
182
183
184
185
186
187
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 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
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 
34
35
36
37
38
39
40
41
42
43
44
45
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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@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 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 Docker's default platform for the build host.
        :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 = platform or os.environ.get("ARMNET_DOCKER_PLATFORM")
        build_hash = _compute_build_hash(
            dockerfile_path,
            context_path,
            platform=platform,
            build_contexts=resolved_build_contexts,
        )
        tag = f"{name}:armnet-{build_hash[:12]}"

        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,
                )

        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 
 98
 99
100
@classmethod
def from_ref(cls, ref: str) -> "Image":
    return cls(ref=ref)

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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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 Docker's default platform for the build host. :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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@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 Docker's default platform for the build host.
    :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 = platform or os.environ.get("ARMNET_DOCKER_PLATFORM")
    build_hash = _compute_build_hash(
        dockerfile_path,
        context_path,
        platform=platform,
        build_contexts=resolved_build_contexts,
    )
    tag = f"{name}:armnet-{build_hash[:12]}"

    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,
            )

    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 
64
65
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 
68
69
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 
72
73
74
75
76
77
78
79
80
81
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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@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

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

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 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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.
    """

    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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
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}",
            "-e", f"{runtime_env.HF_HOME}={cache_home / 'huggingface'}",
            "-v", f"{cache_home}:{cache_home}",
            "-v", f"{volume_home}:{volume_home}",
        ])
        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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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.

Starts a (non-suppressing) global pynput listener mapping Right Arrow → next_episode, Left Arrow → rerecord_episode, Esc → stop_recording; each press enqueues the event string 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.

Returns the started listener (call .stop() when done), or None when no keyboard is available (e.g. headless Linux without $DISPLAY — the same pynput limitation LeRobot's own recorder has).

Source code in client/src/armnet_client/teleop_stream.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def start_recording_key_listener(events: "queue.Queue[str]"):
    """Bind LeRobot's dataset-recording keyboard shortcuts to ``events``.

    Starts a (non-suppressing) global pynput listener mapping Right Arrow →
    ``next_episode``, Left Arrow → ``rerecord_episode``, Esc →
    ``stop_recording``; each press enqueues the event string 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.

    Returns the started listener (call ``.stop()`` when done), or None when no
    keyboard is available (e.g. headless Linux without ``$DISPLAY`` — the same
    pynput limitation LeRobot's own recorder has).
    """

    try:
        from pynput import keyboard
    except Exception as exc:  # noqa: BLE001 - optional capability, not a hard failure
        logger.warning("recording keyboard controls unavailable (pynput): %r", exc)
        return None

    def on_press(key) -> None:  # noqa: ANN001 - pynput key type
        if key == keyboard.Key.right:
            event = TELEOP_EVENT_NEXT_EPISODE
        elif key == keyboard.Key.left:
            event = TELEOP_EVENT_RERECORD_EPISODE
        elif key == keyboard.Key.esc:
            event = TELEOP_EVENT_STOP_RECORDING
        else:
            return
        logger.info("recording control keypress -> %s", event)
        events.put(event)

    try:
        listener = keyboard.Listener(on_press=on_press)
        listener.start()
    except Exception as exc:  # noqa: BLE001 - e.g. no X display
        logger.warning("failed to start recording keyboard listener: %r", exc)
        return None
    logger.info(
        "recording keyboard controls active: Right Arrow = save episode + reset, "
        "Left Arrow = re-record episode, Esc = stop session"
    )
    return listener

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 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
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 _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
    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)
                # 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
                    message = TeleopAction(
                        job_id=job_id,
                        timestamp=time.time(),
                        action={str(k): float(v) for k, v in raw_action.items()},
                    )
                    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__)

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): if no matching cell is free when the job is submitted, it is accepted and queued — this call then returns immediately with a non-terminal JobResult (status submitted) rather than blocking, since there may be no cell to run it for a while. 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 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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): if no matching cell is free when the job is
    submitted, it is accepted and queued — this call then returns immediately
    with a non-terminal ``JobResult`` (status ``submitted``) rather than blocking,
    since there may be no cell to run it for a while. 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 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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 
64
65
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 
68
69
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 
72
73
74
75
76
77
78
79
80
81
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.
    """

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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@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 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 Docker's default platform for the build host.
        :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 = platform or os.environ.get("ARMNET_DOCKER_PLATFORM")
        build_hash = _compute_build_hash(
            dockerfile_path,
            context_path,
            platform=platform,
            build_contexts=resolved_build_contexts,
        )
        tag = f"{name}:armnet-{build_hash[:12]}"

        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,
                )

        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 
 98
 99
100
@classmethod
def from_ref(cls, ref: str) -> "Image":
    return cls(ref=ref)

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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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 Docker's default platform for the build host. :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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@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 Docker's default platform for the build host.
    :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 = platform or os.environ.get("ARMNET_DOCKER_PLATFORM")
    build_hash = _compute_build_hash(
        dockerfile_path,
        context_path,
        platform=platform,
        build_contexts=resolved_build_contexts,
    )
    tag = f"{name}:armnet-{build_hash[:12]}"

    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,
            )

    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)

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 
34
35
36
37
38
39
40
41
42
43
44
45
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.

No auth in M0.5 (design doc explicitly defers it). When auth lands we'll add an api_key arg here and inject it as a header.

Source code in client/src/armnet_client/client.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class OrchestratorClient:
    """Synchronous client for the orchestrator HTTP API.

    No auth in M0.5 (design doc explicitly defers it). When auth lands
    we'll add an ``api_key`` arg here and inject it as a header.
    """

    def __init__(
        self,
        base_url: str = "http://localhost:8000",
        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.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) -> list[Job]:
        """GET /jobs — the current user's most-recent jobs (newest first)."""
        resp = self._client.get("/jobs", params={"limit": limit})
        _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 
69
70
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 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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 
 96
 97
 98
 99
100
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) -> list[Job]

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

Source code in client/src/armnet_client/client.py 
102
103
104
105
106
def list_jobs(self, *, limit: int = 10) -> list[Job]:
    """GET /jobs — the current user's most-recent jobs (newest first)."""
    resp = self._client.get("/jobs", params={"limit": limit})
    _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 
108
109
110
111
112
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 
114
115
116
117
118
119
120
121
122
123
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 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
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 
140
141
142
143
144
145
146
147
148
149
150
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 
152
153
154
155
156
157
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 
159
160
161
162
163
164
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 
166
167
168
169
170
171
172
173
174
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 
176
177
178
179
180
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 
182
183
184
185
186
187
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 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
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

Local armnet volume helpers.

upload_to_local_volume 

upload_to_local_volume(*, username: str, local_path: str | Path, volume_path: str | Path, overwrite: bool = False) -> Path
Source code in client/src/armnet_client/volume.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def upload_to_local_volume(
    *,
    username: str,
    local_path: str | Path,
    volume_path: str | Path,
    overwrite: bool = False,
) -> Path:
    src = Path(local_path).expanduser().resolve()
    if not src.exists():
        raise FileNotFoundError(f"local path not found: {src}")

    dest = _volume_root(username) / _safe_relative_path(volume_path)
    if dest.exists():
        if not overwrite:
            return dest
        if dest.is_dir():
            shutil.rmtree(dest)
        else:
            dest.unlink()

    dest.parent.mkdir(parents=True, exist_ok=True)
    if src.is_dir():
        shutil.copytree(src, dest)
    else:
        shutil.copy2(src, dest)
    return dest

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 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def upload_to_cloud_volume(
    *,
    credentials: VolumeCredentials,
    username: str,
    local_path: str | Path,
    volume_path: str | Path,
    overwrite: bool = False,
) -> list[str]:
    src = Path(local_path).expanduser().resolve()
    root_dest = _volume_root(username) / _safe_relative_path(volume_path)
    uploaded: list[str] = []
    if src.is_dir():
        for child in sorted(p for p in src.rglob("*") if p.is_file()):
            rel = child.relative_to(src)
            dest = root_dest / rel
            object_path = _safe_relative_path(volume_path) / rel
            if _upload_file(credentials, child, object_path, overwrite=overwrite):
                uploaded.append(str(object_path))
            _copy_file_local(child, dest, overwrite=overwrite)
    else:
        object_path = _safe_relative_path(volume_path)
        if _upload_file(credentials, src, object_path, overwrite=overwrite):
            uploaded.append(str(object_path))
        _copy_file_local(src, root_dest, overwrite=overwrite)
    return uploaded

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 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def download_from_cloud_volume(
    *,
    credentials: VolumeCredentials,
    username: str,
    volume_path: str | Path,
    local_path: str | Path,
) -> Path:
    rel = _safe_relative_path(volume_path)
    dest = Path(local_path).expanduser().resolve()
    object_name = credentials.prefix + rel.as_posix()
    meta = _get_object_metadata(credentials, object_name)
    if meta is not None:
        if dest.exists() and _local_md5_b64(dest) == meta.get("md5Hash"):
            return dest
        dest.parent.mkdir(parents=True, exist_ok=True)
        _download_object(credentials, object_name, dest)
        _copy_file_local(dest, _volume_root(username) / rel, overwrite=True)
        return dest

    # Directory download.
    if dest.exists() and dest.is_file():
        raise ValueError(f"local_path is a file but cloud path is a directory prefix: {dest}")
    for item in _list_objects(credentials, object_name.rstrip("/") + "/"):
        item_name = item["name"]
        suffix = item_name[len(credentials.prefix + rel.as_posix().rstrip("/") + "/"):]
        file_dest = dest / suffix
        file_dest.parent.mkdir(parents=True, exist_ok=True)
        _download_object(credentials, item_name, file_dest)
        _copy_file_local(file_dest, _volume_root(username) / rel / suffix, overwrite=True)
    return dest

delete_from_cloud_volume 

delete_from_cloud_volume(*, credentials: VolumeCredentials, volume_path: str | Path) -> None
Source code in client/src/armnet_client/volume.py 
104
105
106
107
108
109
110
111
112
def delete_from_cloud_volume(*, credentials: VolumeCredentials, volume_path: str | Path) -> None:
    rel = _safe_relative_path(volume_path)
    object_name = credentials.prefix + rel.as_posix()
    meta = _get_object_metadata(credentials, object_name)
    if meta is not None:
        _delete_object(credentials, object_name)
        return
    for item in _list_objects(credentials, object_name.rstrip("/") + "/"):
        _delete_object(credentials, item["name"])

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 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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.
    """

    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 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
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}",
            "-e", f"{runtime_env.HF_HOME}={cache_home / 'huggingface'}",
            "-v", f"{cache_home}:{cache_home}",
            "-v", f"{volume_home}:{volume_home}",
        ])
        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 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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 _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)
    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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@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

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

start_recording_key_listener 

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

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

Starts a (non-suppressing) global pynput listener mapping Right Arrow → next_episode, Left Arrow → rerecord_episode, Esc → stop_recording; each press enqueues the event string 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.

Returns the started listener (call .stop() when done), or None when no keyboard is available (e.g. headless Linux without $DISPLAY — the same pynput limitation LeRobot's own recorder has).

Source code in client/src/armnet_client/teleop_stream.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def start_recording_key_listener(events: "queue.Queue[str]"):
    """Bind LeRobot's dataset-recording keyboard shortcuts to ``events``.

    Starts a (non-suppressing) global pynput listener mapping Right Arrow →
    ``next_episode``, Left Arrow → ``rerecord_episode``, Esc →
    ``stop_recording``; each press enqueues the event string 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.

    Returns the started listener (call ``.stop()`` when done), or None when no
    keyboard is available (e.g. headless Linux without ``$DISPLAY`` — the same
    pynput limitation LeRobot's own recorder has).
    """

    try:
        from pynput import keyboard
    except Exception as exc:  # noqa: BLE001 - optional capability, not a hard failure
        logger.warning("recording keyboard controls unavailable (pynput): %r", exc)
        return None

    def on_press(key) -> None:  # noqa: ANN001 - pynput key type
        if key == keyboard.Key.right:
            event = TELEOP_EVENT_NEXT_EPISODE
        elif key == keyboard.Key.left:
            event = TELEOP_EVENT_RERECORD_EPISODE
        elif key == keyboard.Key.esc:
            event = TELEOP_EVENT_STOP_RECORDING
        else:
            return
        logger.info("recording control keypress -> %s", event)
        events.put(event)

    try:
        listener = keyboard.Listener(on_press=on_press)
        listener.start()
    except Exception as exc:  # noqa: BLE001 - e.g. no X display
        logger.warning("failed to start recording keyboard listener: %r", exc)
        return None
    logger.info(
        "recording keyboard controls active: Right Arrow = save episode + reset, "
        "Left Arrow = re-record episode, Esc = stop session"
    )
    return listener

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 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
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 _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
    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)
                # 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
                    message = TeleopAction(
                        job_id=job_id,
                        timestamp=time.time(),
                        action={str(k): float(v) for k, v in raw_action.items()},
                    )
                    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