Skip to content

armnet-core API Reference

armnet_core

Shared wire types for the armnet platform.

Single source of truth for everything that crosses a component boundary (orchestrator HTTP API request/response bodies, NATS message envelopes, runtime SDK return values).

API_KEY_ENV module-attribute 

API_KEY_ENV = 'ARMNET_API_KEY'

API_KEY_HEADER module-attribute 

API_KEY_HEADER = 'X-Armnet-Api-Key'

DEFAULT_EMBODIMENT module-attribute 

DEFAULT_EMBODIMENT: Embodiment = 'lerobot/so-101'

DEFAULT_ENVIRONMENT module-attribute 

DEFAULT_ENVIRONMENT: Environment = 'busybox'

DEFAULT_TASK module-attribute 

DEFAULT_TASK: Task = 'assemble_block_tower'

Embodiment module-attribute 

Embodiment = str

EnvironmentName module-attribute 

EnvironmentName = str

Task module-attribute 

Task = str

BOX_LENGTH_SCALES module-attribute 

BOX_LENGTH_SCALES = MappingProxyType({ShirtSize.XS: 0.035, ShirtSize.S: 0.055, ShirtSize.M: 0.085, ShirtSize.L: 0.13, ShirtSize.XL: 0.2})

DEFAULT_LAYOUT_ATTEMPTS module-attribute 

DEFAULT_LAYOUT_ATTEMPTS = 256

EDGE_DISTANCE_SCALES module-attribute 

EDGE_DISTANCE_SCALES = MappingProxyType({ShirtSize.XS: 0.015, ShirtSize.S: 0.035, ShirtSize.M: 0.07, ShirtSize.L: 0.14, ShirtSize.XL: 0.28})

MAX_MANUAL_RESET_ELEMENTS module-attribute 

MAX_MANUAL_RESET_ELEMENTS = 12

POINT_ARROW_LENGTH_SCALE module-attribute 

POINT_ARROW_LENGTH_SCALE = 0.04

POINT_RADIUS_SCALE module-attribute 

POINT_RADIUS_SCALE = 0.006

ManualResetElement module-attribute 

ManualResetElement = Annotated[ManualResetPoint | ManualResetBoundingBox, Field(discriminator='type')]

ManualResetElementGeometry module-attribute 

ManualResetElementGeometry = Annotated[ManualResetPointGeometry | ManualResetBoundingBoxGeometry, Field(discriminator='type')]

BIMANUAL_YAM_EMBODIMENT module-attribute 

BIMANUAL_YAM_EMBODIMENT = 'lerobot/bimanual_yam'

TELEOP_EVENT_NEXT_EPISODE module-attribute 

TELEOP_EVENT_NEXT_EPISODE = 'next_episode'

TELEOP_EVENT_RERECORD_EPISODE module-attribute 

TELEOP_EVENT_RERECORD_EPISODE = 'rerecord_episode'

TELEOP_EVENT_STOP_RECORDING module-attribute 

TELEOP_EVENT_STOP_RECORDING = 'stop_recording'

TELEOP_EVENTS module-attribute 

TELEOP_EVENTS = (TELEOP_EVENT_NEXT_EPISODE, TELEOP_EVENT_RERECORD_EPISODE, TELEOP_EVENT_STOP_RECORDING)

VARIATION_CLIP_SIGMA module-attribute 

VARIATION_CLIP_SIGMA = 2.5

YAM_ARM_JOINTS module-attribute 

YAM_ARM_JOINTS = ('shoulder_pan', 'shoulder_lift', 'elbow_flex', 'wrist_flex', 'wrist_roll', 'wrist_yaw')

YAM_ARM_PORTS module-attribute 

YAM_ARM_PORTS = {'left': 'can_yam_left', 'right': 'can_yam_right'}

YAM_CAMERA_INDEX_OR_PATH module-attribute 

YAM_CAMERA_INDEX_OR_PATH = {'top': '/dev/video_top', 'left_wrist': '/dev/video_left_wrist', 'right_wrist': '/dev/video_right_wrist'}

YAM_CAMERA_NAMES module-attribute 

YAM_CAMERA_NAMES = ('top', 'left_wrist', 'right_wrist')

YAM_JOINTS module-attribute 

YAM_JOINTS = YAM_ARM_JOINTS + ('gripper',)

CurrentGateConfig module-attribute 

CurrentGateConfig = Annotated[UniformCurrentGateConfig | PercentileCurrentGateConfig | OffCurrentGateConfig, Field(discriminator='mode')]

JobId module-attribute 

JobId = Annotated[str, BeforeValidator(canonical_job_id)]

TerminalStatus module-attribute 

TerminalStatus = frozenset({JobStatus.SUCCEEDED, JobStatus.FAILED, JobStatus.TIMEOUT, JobStatus.CANCELLED})

FULL_FRAME_VERTICES module-attribute 

FULL_FRAME_VERTICES: tuple[NormalizedPoint, ...] = ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0))

MIN_NORMALIZED_POLYGON_AREA module-attribute 

MIN_NORMALIZED_POLYGON_AREA = 1e-06

NormalizedPoint module-attribute 

NormalizedPoint = tuple[float, float]

CompletionStatus

Bases: NamedTuple

Whether an episode has finished, how it went, and who decided.

bool(status) is status.complete, so a caller that only cares whether to keep going can use it directly.

scored_by names the judge — an environment's own name when its instrumentation decided, "operator" for a human verdict, "monitor" for the automated completion model. Recorded with results so a success rate can be read knowing what produced it.

Source code in core/src/armnet_core/environment.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class CompletionStatus(NamedTuple):
    """Whether an episode has finished, how it went, and who decided.

    ``bool(status)`` is ``status.complete``, so a caller that only cares
    whether to keep going can use it directly.

    ``scored_by`` names the judge — an environment's own name when its
    instrumentation decided, ``"operator"`` for a human verdict, ``"monitor"``
    for the automated completion model. Recorded with results so a success rate
    can be read knowing what produced it.
    """

    complete: bool
    success: bool
    scored_by: Optional[str] = None

    def __bool__(self) -> bool:
        return self.complete

complete instance-attribute 

complete: bool

success instance-attribute 

success: bool

scored_by class-attribute instance-attribute 

scored_by: Optional[str] = None

Environment

Bases: Protocol

A kind of workcell and the tasks it offers.

Source code in core/src/armnet_core/environment.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@runtime_checkable
class Environment(Protocol):
    """A kind of workcell and the tasks it offers."""

    name: str

    def tasks(self) -> Mapping[str, TaskSpec]:
        """Every task this environment can run, keyed by slug."""

    def connect(
        self,
        config: Mapping[str, Any],
        *,
        task: Optional[str],
        report_progress: Optional[Callable[[str], None]] = None,
    ) -> Optional[InstrumentedCell]:
        """Open a session, or return ``None`` if this cell has no hardware.

        ``config`` is the cell's environment block, passed through untouched
        from its configuration file.
        """

name instance-attribute 

name: str

tasks 

tasks() -> Mapping[str, TaskSpec]

Every task this environment can run, keyed by slug.

Source code in core/src/armnet_core/environment.py 
185
186
def tasks(self) -> Mapping[str, TaskSpec]:
    """Every task this environment can run, keyed by slug."""

connect 

connect(config: Mapping[str, Any], *, task: Optional[str], report_progress: Optional[Callable[[str], None]] = None) -> Optional[InstrumentedCell]

Open a session, or return None if this cell has no hardware.

config is the cell's environment block, passed through untouched from its configuration file.

Source code in core/src/armnet_core/environment.py 
188
189
190
191
192
193
194
195
196
197
198
199
def connect(
    self,
    config: Mapping[str, Any],
    *,
    task: Optional[str],
    report_progress: Optional[Callable[[str], None]] = None,
) -> Optional[InstrumentedCell]:
    """Open a session, or return ``None`` if this cell has no hardware.

    ``config`` is the cell's environment block, passed through untouched
    from its configuration file.
    """

EnvironmentNotFound

Bases: LookupError

Raised when no installed package provides the named environment.

Source code in core/src/armnet_core/environment.py 
202
203
class EnvironmentNotFound(LookupError):
    """Raised when no installed package provides the named environment."""

InstrumentedCell

Bases: Protocol

A live session against one cell's instrumentation, for one job.

Implementations own everything about their kind of workcell: the transport, how a task's goal is expressed, and how the scene is restored between episodes. Job code sees only this interface, through ctx.cell.

No method here may raise because the hardware misbehaved. Instrumentation is an aid to a rollout, never the thing that fails one, so a silent panel reports "nothing known" and lets the operator decide.

Source code in core/src/armnet_core/environment.py 
 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
@runtime_checkable
class InstrumentedCell(Protocol):
    """A live session against one cell's instrumentation, for one job.

    Implementations own everything about their kind of workcell: the transport,
    how a task's goal is expressed, and how the scene is restored between
    episodes. Job code sees only this interface, through ``ctx.cell``.

    No method here may raise because the hardware misbehaved. Instrumentation
    is an aid to a rollout, never the thing that fails one, so a silent panel
    reports "nothing known" and lets the operator decide.
    """

    def instrument(self, robot: Any) -> Any:
        """Return ``robot``, wrapped if this environment records alongside it.

        The wrapper must not change the observation the policy sees. Readings
        belong in a sidecar, not in ``observation.state``, or a policy trained
        without the instrumentation will behave differently with it attached.
        """

    def readings(self) -> Mapping[str, Reading]:
        """Current value of every channel, empty if nothing has been heard."""

    def begin_episode(self) -> Iterable[str]:
        """Open an episode window and report anything wrong with the scene.

        Returns human-readable problems — a goal already satisfied before the
        rollout starts, say — which make the episode unscorable rather than
        failed. An empty iterable means the scene is ready.
        """

    def episode_status(self, *, final: bool = False) -> CompletionStatus:
        """Whether the instrumentation can call this episode yet.

        ``complete=False`` means "no verdict", whether because the goal is
        unmet or because nothing readable bears on it. Callers fall back to
        another judge; they cannot distinguish, and should not need to.

        ``final=True`` says the episode is out of time and asks for a last
        word. An unmet goal is then a failure rather than an abstention: an
        operator should not be asked to confirm a failure the instrumentation
        can see. Channels that read nothing still abstain.
        """

    def reset_scene(
        self,
        *,
        confirm: bool,
        reset_cell: Callable[[bool], None],
        set_rail: Callable[[float], None] | None = None,
    ) -> None:
        """Restore the state this task starts from.

        ``reset_cell`` returns the arm to rest and, when passed ``True``, waits
        for an operator. Implementations decide whether a human is needed;
        ``confirm=True`` means the caller has already decided one is.

        ``set_rail``, when provided, temporarily positions the cell's rails to a
        0..1 staging fraction before replaying keypoints authored at that pose.
        The runtime restores every rail to its job episode position after
        ``reset_scene`` returns or fails; environments must not treat the plan
        pose as a replacement for the job position.
        """

    def attach_dataset(self, dataset_root: Any) -> None:
        """Begin recording readings beside a dataset being written."""

    def record_frame(self) -> None:
        """Buffer the current readings against the frame just recorded."""

    def commit_episode(self, episode_index: int) -> None:
        """Persist buffered readings for a saved episode."""

    def discard_episode(self) -> None:
        """Drop buffered readings for an episode that will not be saved."""

    def close(self) -> None:
        """Release the connection."""

instrument 

instrument(robot: Any) -> Any

Return robot, wrapped if this environment records alongside it.

The wrapper must not change the observation the policy sees. Readings belong in a sidecar, not in observation.state, or a policy trained without the instrumentation will behave differently with it attached.

Source code in core/src/armnet_core/environment.py 
111
112
113
114
115
116
117
def instrument(self, robot: Any) -> Any:
    """Return ``robot``, wrapped if this environment records alongside it.

    The wrapper must not change the observation the policy sees. Readings
    belong in a sidecar, not in ``observation.state``, or a policy trained
    without the instrumentation will behave differently with it attached.
    """

readings 

readings() -> Mapping[str, Reading]

Current value of every channel, empty if nothing has been heard.

Source code in core/src/armnet_core/environment.py 
119
120
def readings(self) -> Mapping[str, Reading]:
    """Current value of every channel, empty if nothing has been heard."""

begin_episode 

begin_episode() -> Iterable[str]

Open an episode window and report anything wrong with the scene.

Returns human-readable problems — a goal already satisfied before the rollout starts, say — which make the episode unscorable rather than failed. An empty iterable means the scene is ready.

Source code in core/src/armnet_core/environment.py 
122
123
124
125
126
127
128
def begin_episode(self) -> Iterable[str]:
    """Open an episode window and report anything wrong with the scene.

    Returns human-readable problems — a goal already satisfied before the
    rollout starts, say — which make the episode unscorable rather than
    failed. An empty iterable means the scene is ready.
    """

episode_status 

episode_status(*, final: bool = False) -> CompletionStatus

Whether the instrumentation can call this episode yet.

complete=False means "no verdict", whether because the goal is unmet or because nothing readable bears on it. Callers fall back to another judge; they cannot distinguish, and should not need to.

final=True says the episode is out of time and asks for a last word. An unmet goal is then a failure rather than an abstention: an operator should not be asked to confirm a failure the instrumentation can see. Channels that read nothing still abstain.

Source code in core/src/armnet_core/environment.py 
130
131
132
133
134
135
136
137
138
139
140
141
def episode_status(self, *, final: bool = False) -> CompletionStatus:
    """Whether the instrumentation can call this episode yet.

    ``complete=False`` means "no verdict", whether because the goal is
    unmet or because nothing readable bears on it. Callers fall back to
    another judge; they cannot distinguish, and should not need to.

    ``final=True`` says the episode is out of time and asks for a last
    word. An unmet goal is then a failure rather than an abstention: an
    operator should not be asked to confirm a failure the instrumentation
    can see. Channels that read nothing still abstain.
    """

reset_scene 

reset_scene(*, confirm: bool, reset_cell: Callable[[bool], None], set_rail: Callable[[float], None] | None = None) -> None

Restore the state this task starts from.

reset_cell returns the arm to rest and, when passed True, waits for an operator. Implementations decide whether a human is needed; confirm=True means the caller has already decided one is.

set_rail, when provided, temporarily positions the cell's rails to a 0..1 staging fraction before replaying keypoints authored at that pose. The runtime restores every rail to its job episode position after reset_scene returns or fails; environments must not treat the plan pose as a replacement for the job position.

Source code in core/src/armnet_core/environment.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def reset_scene(
    self,
    *,
    confirm: bool,
    reset_cell: Callable[[bool], None],
    set_rail: Callable[[float], None] | None = None,
) -> None:
    """Restore the state this task starts from.

    ``reset_cell`` returns the arm to rest and, when passed ``True``, waits
    for an operator. Implementations decide whether a human is needed;
    ``confirm=True`` means the caller has already decided one is.

    ``set_rail``, when provided, temporarily positions the cell's rails to a
    0..1 staging fraction before replaying keypoints authored at that pose.
    The runtime restores every rail to its job episode position after
    ``reset_scene`` returns or fails; environments must not treat the plan
    pose as a replacement for the job position.
    """

attach_dataset 

attach_dataset(dataset_root: Any) -> None

Begin recording readings beside a dataset being written.

Source code in core/src/armnet_core/environment.py 
163
164
def attach_dataset(self, dataset_root: Any) -> None:
    """Begin recording readings beside a dataset being written."""

record_frame 

record_frame() -> None

Buffer the current readings against the frame just recorded.

Source code in core/src/armnet_core/environment.py 
166
167
def record_frame(self) -> None:
    """Buffer the current readings against the frame just recorded."""

commit_episode 

commit_episode(episode_index: int) -> None

Persist buffered readings for a saved episode.

Source code in core/src/armnet_core/environment.py 
169
170
def commit_episode(self, episode_index: int) -> None:
    """Persist buffered readings for a saved episode."""

discard_episode 

discard_episode() -> None

Drop buffered readings for an episode that will not be saved.

Source code in core/src/armnet_core/environment.py 
172
173
def discard_episode(self) -> None:
    """Drop buffered readings for an episode that will not be saved."""

close 

close() -> None

Release the connection.

Source code in core/src/armnet_core/environment.py 
175
176
def close(self) -> None:
    """Release the connection."""

Reading dataclass

One instrumentation channel's current value.

continuous separates channels that sweep through values as they move, such as a slider or a dial, from ones that jump between a few states, such as a switch. Anything watching the instrumentation has to treat those differently: a switch is worth reporting on every change, while a dial being turned would otherwise report continuously.

Source code in core/src/armnet_core/environment.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class Reading:
    """One instrumentation channel's current value.

    ``continuous`` separates channels that sweep through values as they move,
    such as a slider or a dial, from ones that jump between a few states, such
    as a switch. Anything watching the instrumentation has to treat those
    differently: a switch is worth reporting on every change, while a dial
    being turned would otherwise report continuously.
    """

    name: str
    kind: str
    value: float
    continuous: bool = False
    label: Optional[str] = None

    def describe(self) -> str:
        return f"{self.name}: {self.label if self.label is not None else self.value}"

name instance-attribute 

name: str

kind instance-attribute 

kind: str

value instance-attribute 

value: float

continuous class-attribute instance-attribute 

continuous: bool = False

label class-attribute instance-attribute 

label: Optional[str] = None

describe 

describe() -> str
Source code in core/src/armnet_core/environment.py 
82
83
def describe(self) -> str:
    return f"{self.name}: {self.label if self.label is not None else self.value}"

TaskSpec dataclass

A task an environment offers.

instruction is the natural-language form handed to a policy, and is what the platform stores as the task description.

Source code in core/src/armnet_core/environment.py 
86
87
88
89
90
91
92
93
94
95
@dataclass(frozen=True)
class TaskSpec:
    """A task an environment offers.

    ``instruction`` is the natural-language form handed to a policy, and is
    what the platform stores as the task description.
    """

    slug: str
    instruction: str

slug instance-attribute 

slug: str

instruction instance-attribute 

instruction: str

ManualResetBoundingBox

Bases: _ManualResetElement

A rectangular object whose dimensions use the box shirt-size scale.

Source code in core/src/armnet_core/manual_reset.py 
 98
 99
100
101
102
103
class ManualResetBoundingBox(_ManualResetElement):
    """A rectangular object whose dimensions use the box shirt-size scale."""

    type: Literal["bounding_box"] = "bounding_box"
    horizontal_length: ShirtSize
    vertical_length: ShirtSize

type class-attribute instance-attribute 

type: Literal['bounding_box'] = 'bounding_box'

horizontal_length instance-attribute 

horizontal_length: ShirtSize

vertical_length instance-attribute 

vertical_length: ShirtSize

ManualResetBoundingBoxGeometry

Bases: _FrozenModel

Normalized rendering geometry for a sampled rotated rectangle.

Source code in core/src/armnet_core/manual_reset.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class ManualResetBoundingBoxGeometry(_FrozenModel):
    """Normalized rendering geometry for a sampled rotated rectangle."""

    type: Literal["bounding_box"] = "bounding_box"
    element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)
    name: str
    center: NormalizedPoint
    corners: tuple[
        NormalizedPoint,
        NormalizedPoint,
        NormalizedPoint,
        NormalizedPoint,
    ]
    rotation_degrees: float
    arrow: NormalizedLine

type class-attribute instance-attribute 

type: Literal['bounding_box'] = 'bounding_box'

element_index class-attribute instance-attribute 

element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)

name instance-attribute 

name: str

center instance-attribute 

center: NormalizedPoint

corners instance-attribute 

corners: tuple[NormalizedPoint, NormalizedPoint, NormalizedPoint, NormalizedPoint]

rotation_degrees instance-attribute 

rotation_degrees: float

arrow instance-attribute 

arrow: NormalizedLine

ManualResetConfig

Bases: _FrozenModel

Versioned task reset instructions in stable display/sampling order.

Source code in core/src/armnet_core/manual_reset.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class ManualResetConfig(_FrozenModel):
    """Versioned task reset instructions in stable display/sampling order."""

    schema_version: Literal[1] = 1
    elements: tuple[ManualResetElement, ...] = Field(
        default=(),
        max_length=MAX_MANUAL_RESET_ELEMENTS,
    )

    @model_validator(mode="after")
    def _unique_element_names(self) -> ManualResetConfig:
        names = [element.name.casefold() for element in self.elements]
        if len(set(names)) != len(names):
            raise ValueError("manual-reset element names must be unique")
        return self

schema_version class-attribute instance-attribute 

schema_version: Literal[1] = 1

elements class-attribute instance-attribute 

elements: tuple[ManualResetElement, ...] = Field(default=(), max_length=MAX_MANUAL_RESET_ELEMENTS)

ManualResetLayout

Bases: _FrozenModel

A deterministic layout ready for normalized image/SVG rendering.

Source code in core/src/armnet_core/manual_reset.py 
172
173
174
175
176
177
class ManualResetLayout(_FrozenModel):
    """A deterministic layout ready for normalized image/SVG rendering."""

    source_frame_width: int = Field(gt=0)
    source_frame_height: int = Field(gt=0)
    elements: tuple[ManualResetElementGeometry, ...]

source_frame_width class-attribute instance-attribute 

source_frame_width: int = Field(gt=0)

source_frame_height class-attribute instance-attribute 

source_frame_height: int = Field(gt=0)

elements instance-attribute 

elements: tuple[ManualResetElementGeometry, ...]

ManualResetLayoutError

Bases: ValueError

Raised when a configured layout cannot fit after bounded retries.

Source code in core/src/armnet_core/manual_reset.py 
180
181
class ManualResetLayoutError(ValueError):
    """Raised when a configured layout cannot fit after bounded retries."""

ManualResetPoint

Bases: _ManualResetElement

A small point-like object with an orientation indicator.

Source code in core/src/armnet_core/manual_reset.py 
92
93
94
95
class ManualResetPoint(_ManualResetElement):
    """A small point-like object with an orientation indicator."""

    type: Literal["point"] = "point"

type class-attribute instance-attribute 

type: Literal['point'] = 'point'

ManualResetPointGeometry

Bases: _FrozenModel

Normalized rendering geometry for a sampled point element.

Source code in core/src/armnet_core/manual_reset.py 
136
137
138
139
140
141
142
143
144
145
146
class ManualResetPointGeometry(_FrozenModel):
    """Normalized rendering geometry for a sampled point element."""

    type: Literal["point"] = "point"
    element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)
    name: str
    center: NormalizedPoint
    radius_x: float = Field(gt=0)
    radius_y: float = Field(gt=0)
    rotation_degrees: float
    arrow: NormalizedLine

type class-attribute instance-attribute 

type: Literal['point'] = 'point'

element_index class-attribute instance-attribute 

element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)

name instance-attribute 

name: str

center instance-attribute 

center: NormalizedPoint

radius_x class-attribute instance-attribute 

radius_x: float = Field(gt=0)

radius_y class-attribute instance-attribute 

radius_y: float = Field(gt=0)

rotation_degrees instance-attribute 

rotation_degrees: float

arrow instance-attribute 

arrow: NormalizedLine

NormalizedLine

Bases: _FrozenModel

A normalized line segment.

Source code in core/src/armnet_core/manual_reset.py 
129
130
131
132
133
class NormalizedLine(_FrozenModel):
    """A normalized line segment."""

    start: NormalizedPoint
    end: NormalizedPoint

start instance-attribute 

start: NormalizedPoint

end instance-attribute 

end: NormalizedPoint

ShirtSize

Bases: str, Enum

Discrete physical scale used by manual-reset instructions.

Source code in core/src/armnet_core/manual_reset.py 
44
45
46
47
48
49
50
51
class ShirtSize(str, Enum):
    """Discrete physical scale used by manual-reset instructions."""

    XS = "XS"
    S = "S"
    M = "M"
    L = "L"
    XL = "XL"

XS class-attribute instance-attribute 

XS = 'XS'

S class-attribute instance-attribute 

S = 'S'

M class-attribute instance-attribute 

M = 'M'

L class-attribute instance-attribute 

L = 'L'

XL class-attribute instance-attribute 

XL = 'XL'

ImageReferenceError

Bases: ValueError

An image ref cannot be safely canonicalized for one customer.

Source code in core/src/armnet_core/image_refs.py 
20
21
class ImageReferenceError(ValueError):
    """An image ref cannot be safely canonicalized for one customer."""

CameraMountVariation

Bases: BaseModel

Both axes of one pan/tilt camera mount.

The unit is the mount's own: degrees for a hobby arducam_mount, encoder steps for a serial servo_mount. Nothing compares the two, and each axis carries the limits it was derived from, so neither the sampler nor the edge has to know which kind it is holding.

Source code in core/src/armnet_core/models.py 
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
class CameraMountVariation(BaseModel):
    """Both axes of one pan/tilt camera mount.

    The unit is the mount's own: degrees for a hobby ``arducam_mount``, encoder
    steps for a serial ``servo_mount``. Nothing compares the two, and each
    axis carries the limits it was derived from, so neither the sampler nor the
    edge has to know which kind it is holding.
    """

    model_config = ConfigDict(extra="forbid")

    pan: VariationAxis
    tilt: VariationAxis

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

pan instance-attribute 

pan: VariationAxis

tilt instance-attribute 

tilt: VariationAxis

CellDeviceConfig

Bases: BaseModel

What a cell can vary, and by how much.

Derived from the cell's own config in one place so the client validating an operator's override before submission and the runtime sampling a scenario read the same envelope, instead of each deriving one and drifting apart.

Source code in core/src/armnet_core/models.py 
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
class CellDeviceConfig(BaseModel):
    """What a cell can vary, and by how much.

    Derived from the cell's own config in one place so the client validating an
    operator's override before submission and the runtime sampling a scenario
    read the same envelope, instead of each deriving one and drifting apart.
    """

    model_config = ConfigDict(extra="forbid")

    rails: list[RailVariation] = Field(default_factory=list)
    camera_mounts: dict[str, CameraMountVariation] = Field(default_factory=dict)
    lightbox_brightness: Optional[VariationAxis] = None

    @classmethod
    def from_cell_config(cls, config: "RobotCellConfig") -> "CellDeviceConfig":
        rails: list[RailVariation] = []
        if config.rail is not None:
            rails.append(_rail_variation(config.rail, arm=None))
        for arm_name, arm in (config.arms or {}).items():
            if arm.rail is not None:
                rails.append(_rail_variation(arm.rail, arm=arm_name))

        mounts: dict[str, CameraMountVariation] = {}
        for name, camera in (config.camera_configs or {}).items():
            if not isinstance(camera, dict):
                continue
            if isinstance(camera.get("arducam_mount"), dict):
                mounts[name] = _camera_mount_variation(camera["arducam_mount"])
            elif isinstance(camera.get("servo_mount"), dict):
                mounts[name] = _servo_mount_variation(camera["servo_mount"])

        # A cell with no lightbox URL has no light to vary, however the
        # brightness defaults are set.
        brightness = (
            VariationAxis(
                default=float(config.lightbox_default_brightness),
                std=float(config.lightbox_variation_std),
                minimum=0.0,
                maximum=100.0,
            )
            if config.lightbox_url
            else None
        )
        return cls(rails=rails, camera_mounts=mounts, lightbox_brightness=brightness)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

rails class-attribute instance-attribute 

rails: list[RailVariation] = Field(default_factory=list)

camera_mounts class-attribute instance-attribute 

camera_mounts: dict[str, CameraMountVariation] = Field(default_factory=dict)

lightbox_brightness class-attribute instance-attribute 

lightbox_brightness: Optional[VariationAxis] = None

from_cell_config classmethod 

from_cell_config(config: 'RobotCellConfig') -> 'CellDeviceConfig'
Source code in core/src/armnet_core/models.py 
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
@classmethod
def from_cell_config(cls, config: "RobotCellConfig") -> "CellDeviceConfig":
    rails: list[RailVariation] = []
    if config.rail is not None:
        rails.append(_rail_variation(config.rail, arm=None))
    for arm_name, arm in (config.arms or {}).items():
        if arm.rail is not None:
            rails.append(_rail_variation(arm.rail, arm=arm_name))

    mounts: dict[str, CameraMountVariation] = {}
    for name, camera in (config.camera_configs or {}).items():
        if not isinstance(camera, dict):
            continue
        if isinstance(camera.get("arducam_mount"), dict):
            mounts[name] = _camera_mount_variation(camera["arducam_mount"])
        elif isinstance(camera.get("servo_mount"), dict):
            mounts[name] = _servo_mount_variation(camera["servo_mount"])

    # A cell with no lightbox URL has no light to vary, however the
    # brightness defaults are set.
    brightness = (
        VariationAxis(
            default=float(config.lightbox_default_brightness),
            std=float(config.lightbox_variation_std),
            minimum=0.0,
            maximum=100.0,
        )
        if config.lightbox_url
        else None
    )
    return cls(rails=rails, camera_mounts=mounts, lightbox_brightness=brightness)

CellRuntimeStatus

Bases: str, Enum

Current availability state for a robot cell.

Source code in core/src/armnet_core/models.py 
81
82
83
84
85
86
87
class CellRuntimeStatus(str, Enum):
    """Current availability state for a robot cell."""

    AVAILABLE = "available"
    OCCUPIED = "occupied"
    MAINTENANCE = "maintenance"
    OFFLINE = "offline"

AVAILABLE class-attribute instance-attribute 

AVAILABLE = 'available'

OCCUPIED class-attribute instance-attribute 

OCCUPIED = 'occupied'

MAINTENANCE class-attribute instance-attribute 

MAINTENANCE = 'maintenance'

OFFLINE class-attribute instance-attribute 

OFFLINE = 'offline'

CellStatus

Bases: BaseModel

Current status for one cell as reported by recent heartbeats.

Source code in core/src/armnet_core/models.py 
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
class CellStatus(BaseModel):
    """Current status for one cell as reported by recent heartbeats."""

    cell_id: str
    embodiment: Embodiment
    # What the cell is set up to work on. It can run any task belonging to this,
    # so the cell reports the environment and the job carries the task.
    environment: Environment
    status: CellRuntimeStatus
    current_job_id: Optional[str] = None
    last_heartbeat_at: Optional[datetime] = None
    # What this cell can vary and within what bounds, so a client can check an
    # override before submitting rather than learning from a failed job. None
    # from a cell too old to report it, which reads as "cannot check here".
    device_config: Optional[CellDeviceConfig] = None

cell_id instance-attribute 

cell_id: str

embodiment instance-attribute 

embodiment: Embodiment

environment instance-attribute 

environment: Environment

status instance-attribute 

status: CellRuntimeStatus

current_job_id class-attribute instance-attribute 

current_job_id: Optional[str] = None

last_heartbeat_at class-attribute instance-attribute 

last_heartbeat_at: Optional[datetime] = None

device_config class-attribute instance-attribute 

device_config: Optional[CellDeviceConfig] = None

CellStatusSummary

Bases: BaseModel

Aggregate availability for an embodiment, optionally scoped to a task.

Callers ask by task, which is what they think in; the orchestrator resolves it to the environment that can run it and matches cells on that. Both are echoed back. task is None when the summary aggregates across the whole embodiment, which is what a task-less job needs.

Source code in core/src/armnet_core/models.py 
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
class CellStatusSummary(BaseModel):
    """Aggregate availability for an embodiment, optionally scoped to a task.

    Callers ask by ``task``, which is what they think in; the orchestrator
    resolves it to the environment that can run it and matches cells on that.
    Both are echoed back. ``task`` is ``None`` when the summary aggregates
    across the whole embodiment, which is what a task-less job needs.
    """

    embodiment: Embodiment
    task: Optional[Task] = None
    environment: Optional[Environment] = None
    status: CellRuntimeStatus
    cells: list[CellStatus] = Field(default_factory=list)
    heartbeat_timeout_seconds: int = 20

embodiment instance-attribute 

embodiment: Embodiment

task class-attribute instance-attribute 

task: Optional[Task] = None

environment class-attribute instance-attribute 

environment: Optional[Environment] = None

status instance-attribute 

status: CellRuntimeStatus

cells class-attribute instance-attribute 

cells: list[CellStatus] = Field(default_factory=list)

heartbeat_timeout_seconds class-attribute instance-attribute 

heartbeat_timeout_seconds: int = 20

EmbodimentInfo

Bases: BaseModel

One known embodiment, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
1594
1595
1596
1597
1598
class EmbodimentInfo(BaseModel):
    """One known embodiment, as stored in the orchestrator database."""

    slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")

description class-attribute instance-attribute 

description: Optional[str] = None

EmbodimentList

Bases: BaseModel

The set of embodiments the platform currently accepts (GET /embodiments).

Source code in core/src/armnet_core/models.py 
1601
1602
1603
1604
class EmbodimentList(BaseModel):
    """The set of embodiments the platform currently accepts (GET /embodiments)."""

    embodiments: list[EmbodimentInfo] = Field(default_factory=list)

embodiments class-attribute instance-attribute 

embodiments: list[EmbodimentInfo] = Field(default_factory=list)

Job

Bases: BaseModel

The orchestrator's view of a job (response body of GET /jobs/{id}).

Source code in core/src/armnet_core/models.py 
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
class Job(BaseModel):
    """The orchestrator's view of a job (response body of GET /jobs/{id})."""

    id: JobId = Field(default_factory=_new_job_id)
    spec: JobSpec
    status: JobStatus = JobStatus.SUBMITTED
    created_at: datetime = Field(default_factory=_utcnow)
    updated_at: datetime = Field(default_factory=_utcnow)
    cell_id: Optional[str] = Field(
        default=None,
        description="ID of the cell that picked up the job, set on dispatch.",
    )
    result: Optional[JobResult] = None
    dispatched: Optional[bool] = Field(
        default=None,
        description=(
            "Only set on the POST /jobs response: True if the job was dispatched "
            "to a cell immediately, False if it was accepted but queued (no "
            "matching cell was free). None on all other responses. Clients can use "
            "this to decide whether to wait for logs/result or just report 'queued'."
        ),
    )

    def is_terminal(self) -> bool:
        return self.status in TerminalStatus

id class-attribute instance-attribute 

id: JobId = Field(default_factory=_new_job_id)

spec instance-attribute 

spec: JobSpec

status class-attribute instance-attribute 

status: JobStatus = JobStatus.SUBMITTED

created_at class-attribute instance-attribute 

created_at: datetime = Field(default_factory=_utcnow)

updated_at class-attribute instance-attribute 

updated_at: datetime = Field(default_factory=_utcnow)

cell_id class-attribute instance-attribute 

cell_id: Optional[str] = Field(default=None, description='ID of the cell that picked up the job, set on dispatch.')

result class-attribute instance-attribute 

result: Optional[JobResult] = None

dispatched class-attribute instance-attribute 

dispatched: Optional[bool] = Field(default=None, description="Only set on the POST /jobs response: True if the job was dispatched to a cell immediately, False if it was accepted but queued (no matching cell was free). None on all other responses. Clients can use this to decide whether to wait for logs/result or just report 'queued'.")

is_terminal 

is_terminal() -> bool
Source code in core/src/armnet_core/models.py 
1538
1539
def is_terminal(self) -> bool:
    return self.status in TerminalStatus

JobResult

Bases: BaseModel

Terminal result published by a cell on the results subject.

Also returned (embedded in :class:Job) by GET /jobs/{id} once the job is in a terminal state.

Source code in core/src/armnet_core/models.py 
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
class JobResult(BaseModel):
    """Terminal result published by a cell on the results subject.

    Also returned (embedded in :class:`Job`) by ``GET /jobs/{id}`` once the
    job is in a terminal state.
    """

    status: JobStatus = Field(
        ...,
        description="One of the terminal statuses (succeeded/failed/timeout/cancelled).",
    )
    exit_code: Optional[int] = Field(
        default=None,
        description="Container process exit code if the container ran to completion.",
    )
    stdout: str = Field(default="", description="Captured container stdout.")
    stderr: str = Field(default="", description="Captured container stderr.")
    error: Optional[str] = Field(
        default=None,
        description="Short, infra-side reason this job didn't run user code "
        "to completion: image pull failure, docker error, timeout, etc. "
        "Mutually exclusive with `traceback` in practice (one is a platform "
        "failure, the other is a user-code failure).",
    )
    traceback: Optional[str] = Field(
        default=None,
        description="Python traceback from the customer's `@main`-decorated "
        "function if it raised. Extracted by the cell from the "
        "`[armnet:traceback]:json` marker line in stdout. Capped at "
        "~64 KiB on the cell side; the full untruncated traceback is also "
        "in `stderr` for power users. None for successful jobs and for "
        "infra-side failures (those go in `error` instead).",
    )
    return_value: Optional[Any] = Field(
        default=None,
        description="Value returned by the customer's `@main`-decorated "
        "function. Extracted by the cell from the marker line that the "
        "`armnet-runtime` entrypoint prints to stdout. None if the "
        "function returned None or did not run to completion.",
    )
    started_at: Optional[datetime] = None
    finished_at: Optional[datetime] = None

    def raise_for_status(self) -> None:
        """Raise :class:`RemoteExecutionError` iff this result isn't ``SUCCEEDED``.

        The httpx-style "opt-in raising" pattern. Use it when you'd rather
        bail than branch on ``result.status``::

            result = execute(...)
            result.raise_for_status()
            do_thing(result.return_value)

        For successful results this is a no-op.
        """

        if self.status != JobStatus.SUCCEEDED:
            raise RemoteExecutionError(self)

    def __str__(self) -> str:
        """Human-readable rendering. Uses indentation to make tracebacks scannable.

        ``print(result)`` is intended to be the one-liner that tells you
        what happened. ``repr(result)`` (pydantic's default) still shows
        every field for debugging.
        """

        lines: list[str] = []
        head = f"JobResult(status={self.status.value}"
        if self.exit_code is not None:
            head += f", exit_code={self.exit_code}"
        head += ")"
        lines.append(head)
        if self.return_value is not None:
            lines.append(f"  return_value: {self.return_value!r}")
        if self.error:
            lines.append(f"  error: {self.error}")
        if self.traceback:
            lines.append("  traceback (from @main):")
            for tb_line in self.traceback.splitlines():
                lines.append(f"    {tb_line}")
        return "\n".join(lines)

status class-attribute instance-attribute 

status: JobStatus = Field(..., description='One of the terminal statuses (succeeded/failed/timeout/cancelled).')

exit_code class-attribute instance-attribute 

exit_code: Optional[int] = Field(default=None, description='Container process exit code if the container ran to completion.')

stdout class-attribute instance-attribute 

stdout: str = Field(default='', description='Captured container stdout.')

stderr class-attribute instance-attribute 

stderr: str = Field(default='', description='Captured container stderr.')

error class-attribute instance-attribute 

error: Optional[str] = Field(default=None, description="Short, infra-side reason this job didn't run user code to completion: image pull failure, docker error, timeout, etc. Mutually exclusive with `traceback` in practice (one is a platform failure, the other is a user-code failure).")

traceback class-attribute instance-attribute 

traceback: Optional[str] = Field(default=None, description="Python traceback from the customer's `@main`-decorated function if it raised. Extracted by the cell from the `[armnet:traceback]:json` marker line in stdout. Capped at ~64 KiB on the cell side; the full untruncated traceback is also in `stderr` for power users. None for successful jobs and for infra-side failures (those go in `error` instead).")

return_value class-attribute instance-attribute 

return_value: Optional[Any] = Field(default=None, description="Value returned by the customer's `@main`-decorated function. Extracted by the cell from the marker line that the `armnet-runtime` entrypoint prints to stdout. None if the function returned None or did not run to completion.")

started_at class-attribute instance-attribute 

started_at: Optional[datetime] = None

finished_at class-attribute instance-attribute 

finished_at: Optional[datetime] = None

raise_for_status 

raise_for_status() -> None

Raise :class:RemoteExecutionError iff this result isn't SUCCEEDED.

The httpx-style "opt-in raising" pattern. Use it when you'd rather bail than branch on result.status::

result = execute(...)
result.raise_for_status()
do_thing(result.return_value)

For successful results this is a no-op.

Source code in core/src/armnet_core/models.py 
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
def raise_for_status(self) -> None:
    """Raise :class:`RemoteExecutionError` iff this result isn't ``SUCCEEDED``.

    The httpx-style "opt-in raising" pattern. Use it when you'd rather
    bail than branch on ``result.status``::

        result = execute(...)
        result.raise_for_status()
        do_thing(result.return_value)

    For successful results this is a no-op.
    """

    if self.status != JobStatus.SUCCEEDED:
        raise RemoteExecutionError(self)

JobSpec

Bases: BaseModel

The fields a client supplies when creating a job.

This is the body of POST /jobs. The orchestrator wraps it into a :class:Job (assigning id, status, timestamps) before persisting.

Source code in core/src/armnet_core/models.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
class JobSpec(BaseModel):
    """The fields a client supplies when creating a job.

    This is the body of ``POST /jobs``. The orchestrator wraps it into a
    :class:`Job` (assigning ``id``, ``status``, timestamps) before
    persisting.
    """

    image: str = Field(
        ...,
        description=(
            "Container image selector/reference. In production, the orchestrator "
            "uses only the validated image selector and reconstructs registry, "
            "repository, and owner namespace from authenticated platform state. "
            "Local development may use a local `my-image:tag` reference."
        ),
    )
    args: dict[str, Any] = Field(
        default_factory=dict,
        description="Keyword arguments passed to the customer's @main-decorated "
        "function as `ctx.args`. JSON-encoded into the `ARMNET_ARGS` env "
        "var by the cell; decoded by the `armnet-runtime` entrypoint "
        "before calling user code. Must be JSON-serialisable.",
    )
    embodiment: Embodiment = Field(
        ...,
        description="Required robot embodiment. The orchestrator routes the "
        "job onto the NATS subject for this embodiment+task pair, where the "
        "matching cell picks it up.",
    )
    task: Optional[Task] = Field(
        default=None,
        description="Optional task. When set, only cells configured for this "
        "(embodiment, task) pair run the job. When omitted, any cell of the "
        "embodiment may pick it up regardless of its configured task.",
    )
    timeout_seconds: int = Field(
        default=120,
        ge=1,
        description="Wall-clock cap on container execution.",
    )
    secrets: dict[str, str] = Field(
        default_factory=dict,
        description=(
            "Mapping of environment variable name to user secret name. "
            "Example: {'HF_TOKEN': 'huggingface-token'} resolves the "
            "authenticated user's secret and injects it as HF_TOKEN."
        ),
    )
    detach: bool = Field(
        default=False,
        description=(
            "If false, losing the client log WebSocket requests graceful job "
            "cancellation. If true, the job keeps running after client disconnect."
        ),
    )
    skip_operator_dispatch: bool = Field(
        default=False,
        description=(
            "Run immediately on a manual-environment cell instead of waiting "
            "for an operator to press Dispatch in the FMS. Set by interactive "
            "teleop and recording, where the submitter is at the cell. Policy "
            "evals leave this false so the scene is staged first."
        ),
    )
    username: Optional[str] = Field(
        default=None,
        description=(
            "Authenticated armnet username. Set by the orchestrator from "
            "the API key; clients should not rely on supplied values being preserved."
        ),
    )

image class-attribute instance-attribute 

image: str = Field(..., description='Container image selector/reference. In production, the orchestrator uses only the validated image selector and reconstructs registry, repository, and owner namespace from authenticated platform state. Local development may use a local `my-image:tag` reference.')

args class-attribute instance-attribute 

args: dict[str, Any] = Field(default_factory=dict, description="Keyword arguments passed to the customer's @main-decorated function as `ctx.args`. JSON-encoded into the `ARMNET_ARGS` env var by the cell; decoded by the `armnet-runtime` entrypoint before calling user code. Must be JSON-serialisable.")

embodiment class-attribute instance-attribute 

embodiment: Embodiment = Field(..., description='Required robot embodiment. The orchestrator routes the job onto the NATS subject for this embodiment+task pair, where the matching cell picks it up.')

task class-attribute instance-attribute 

task: Optional[Task] = Field(default=None, description='Optional task. When set, only cells configured for this (embodiment, task) pair run the job. When omitted, any cell of the embodiment may pick it up regardless of its configured task.')

timeout_seconds class-attribute instance-attribute 

timeout_seconds: int = Field(default=120, ge=1, description='Wall-clock cap on container execution.')

secrets class-attribute instance-attribute 

secrets: dict[str, str] = Field(default_factory=dict, description="Mapping of environment variable name to user secret name. Example: {'HF_TOKEN': 'huggingface-token'} resolves the authenticated user's secret and injects it as HF_TOKEN.")

detach class-attribute instance-attribute 

detach: bool = Field(default=False, description='If false, losing the client log WebSocket requests graceful job cancellation. If true, the job keeps running after client disconnect.')

skip_operator_dispatch class-attribute instance-attribute 

skip_operator_dispatch: bool = Field(default=False, description='Run immediately on a manual-environment cell instead of waiting for an operator to press Dispatch in the FMS. Set by interactive teleop and recording, where the submitter is at the cell. Policy evals leave this false so the scene is staged first.')

username class-attribute instance-attribute 

username: Optional[str] = Field(default=None, description='Authenticated armnet username. Set by the orchestrator from the API key; clients should not rely on supplied values being preserved.')

JobStatus

Bases: str, Enum

Job lifecycle states.

A job is created SUBMITTED, becomes QUEUED after JetStream confirms durable storage, then DISPATCHED when an idle cell atomically claims it, RUNNING when that cell heartbeats ownership, and finally terminal.

Source code in core/src/armnet_core/models.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class JobStatus(str, Enum):
    """Job lifecycle states.

    A job is created ``SUBMITTED``, becomes ``QUEUED`` after JetStream confirms
    durable storage, then ``DISPATCHED`` when an idle cell atomically claims it,
    ``RUNNING`` when that cell heartbeats ownership, and finally terminal.
    """

    SUBMITTED = "submitted"
    QUEUED = "queued"
    DISPATCHED = "dispatched"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    TIMEOUT = "timeout"
    CANCELLED = "cancelled"

SUBMITTED class-attribute instance-attribute 

SUBMITTED = 'submitted'

QUEUED class-attribute instance-attribute 

QUEUED = 'queued'

DISPATCHED class-attribute instance-attribute 

DISPATCHED = 'dispatched'

RUNNING class-attribute instance-attribute 

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute 

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute 

FAILED = 'failed'

TIMEOUT class-attribute instance-attribute 

TIMEOUT = 'timeout'

CANCELLED class-attribute instance-attribute 

CANCELLED = 'cancelled'

JobStatusTransition

Bases: BaseModel

One entry in a job's status-change audit trail.

Emitted by the orchestrator's GET /jobs/{id}/status-transitions so users can trace a job's lifecycle. detail carries optional context (e.g. the cell id on dispatch/run, or the failure message on a terminal state).

Source code in core/src/armnet_core/models.py 
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
class JobStatusTransition(BaseModel):
    """One entry in a job's status-change audit trail.

    Emitted by the orchestrator's ``GET /jobs/{id}/status-transitions`` so users
    can trace a job's lifecycle. ``detail`` carries optional context (e.g. the
    cell id on dispatch/run, or the failure message on a terminal state).
    """

    status: str
    detail: Optional[str] = None
    timestamp: datetime

status instance-attribute 

status: str

detail class-attribute instance-attribute 

detail: Optional[str] = None

timestamp instance-attribute 

timestamp: datetime

OffCurrentGateConfig

Bases: BaseModel

Explicit operator override that disables current gating.

Source code in core/src/armnet_core/models.py 
481
482
483
484
485
486
class OffCurrentGateConfig(BaseModel):
    """Explicit operator override that disables current gating."""

    model_config = ConfigDict(extra="forbid")

    mode: Literal["off"]

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

mode instance-attribute 

mode: Literal['off']

PercentileCurrentGateConfig

Bases: BaseModel

Per-joint limits derived from the packaged normal-current profile.

Source code in core/src/armnet_core/models.py 
469
470
471
472
473
474
475
476
477
478
class PercentileCurrentGateConfig(BaseModel):
    """Per-joint limits derived from the packaged normal-current profile."""

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    mode: Literal["percentile"]
    percentile: Literal[95, 99] = 99
    percent_above: float = Field(default=100.0, gt=0)
    tiny_max_threshold: int = Field(default=15, gt=0)
    duration_ms: float = Field(default=200.0, gt=0)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

mode instance-attribute 

mode: Literal['percentile']

percentile class-attribute instance-attribute 

percentile: Literal[95, 99] = 99

percent_above class-attribute instance-attribute 

percent_above: float = Field(default=100.0, gt=0)

tiny_max_threshold class-attribute instance-attribute 

tiny_max_threshold: int = Field(default=15, gt=0)

duration_ms class-attribute instance-attribute 

duration_ms: float = Field(default=200.0, gt=0)

RailConfig

Bases: BaseModel

Measured geometry, travel direction, and default job position for one rail.

default_position_percent is where the carriage should sit when a job gives no override (0 = left, 100 = right). Jobs override it with rail_position as a 0–1 fraction. left_dir is the sign of servo rotation that moves the carriage left; it is a property of how the belt and motor are fitted, not of any homing routine.

Source code in core/src/armnet_core/models.py 
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
class RailConfig(BaseModel):
    """Measured geometry, travel direction, and default job position for one rail.

    ``default_position_percent`` is where the carriage should sit when a job
    gives no override (0 = left, 100 = right). Jobs override it with
    ``rail_position`` as a 0–1 fraction. ``left_dir`` is the sign of servo
    rotation that moves the carriage left; it is a property of how the belt
    and motor are fitted, not of any homing routine.
    """

    model_config = ConfigDict(extra="forbid")

    servo_id: int = Field(default=7, ge=0, le=253)
    total_rail_steps: int = Field(gt=0)
    left_dir: Literal[-1, 1] = 1
    default_position_percent: float = Field(default=50, ge=0, le=100)
    variation_std_percent: float = Field(default=10.0, ge=0)
    """Spread of the structured-variation draw around ``default_position_percent``.

    Per rail rather than global because the usable spread depends on where the
    carriage parks: a cell that sits mid-rail can swing both ways, while one
    parked near an end has the draw clipped against the stop and wants a
    smaller figure. Zero pins the rail to its default even when a job asks for
    variation.
    """
    home_end: Literal["left", "right"] = "right"
    """Physical end the carriage homes to when it rebuilds its position.

    Jobs home to this end by default before the container starts. A job that
    explicitly opts into conditional homing may trust the recorded position
    only when its calibration and settled live encoder tick still match; any
    uncertainty re-homes automatically. The end is not computed ("nearest"
    would need an untrusted position); the right end is the calibration
    reference and default.
    """

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

servo_id class-attribute instance-attribute 

servo_id: int = Field(default=7, ge=0, le=253)

total_rail_steps class-attribute instance-attribute 

total_rail_steps: int = Field(gt=0)

left_dir class-attribute instance-attribute 

left_dir: Literal[-1, 1] = 1

default_position_percent class-attribute instance-attribute 

default_position_percent: float = Field(default=50, ge=0, le=100)

variation_std_percent class-attribute instance-attribute 

variation_std_percent: float = Field(default=10.0, ge=0)

Spread of the structured-variation draw around default_position_percent.

Per rail rather than global because the usable spread depends on where the carriage parks: a cell that sits mid-rail can swing both ways, while one parked near an end has the draw clipped against the stop and wants a smaller figure. Zero pins the rail to its default even when a job asks for variation.

home_end class-attribute instance-attribute 

home_end: Literal['left', 'right'] = 'right'

Physical end the carriage homes to when it rebuilds its position.

Jobs home to this end by default before the container starts. A job that explicitly opts into conditional homing may trust the recorded position only when its calibration and settled live encoder tick still match; any uncertainty re-homes automatically. The end is not computed ("nearest" would need an untrusted position); the right end is the calibration reference and default.

RailVariation

Bases: BaseModel

One rail's position axis, in the 0-1 fraction the rail API takes.

Config declares the default and spread in percent, because that reads naturally beside default_position_percent; the conversion to the fraction every caller actually passes happens once here rather than at each comparison.

Source code in core/src/armnet_core/models.py 
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
class RailVariation(BaseModel):
    """One rail's position axis, in the 0-1 fraction the rail API takes.

    Config declares the default and spread in percent, because that reads
    naturally beside ``default_position_percent``; the conversion to the
    fraction every caller actually passes happens once here rather than at
    each comparison.
    """

    model_config = ConfigDict(extra="forbid")

    arm: Optional[str] = None
    position: VariationAxis

    @property
    def label(self) -> str:
        return f"rail {self.arm}" if self.arm else "rail"

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

arm class-attribute instance-attribute 

arm: Optional[str] = None

position instance-attribute 

position: VariationAxis

label property 

label: str

RegistryCredentials

Bases: BaseModel

Short-lived credentials for pushing to the platform's image registry.

Returned by POST /registry/credentials. The orchestrator mints these on demand by impersonating a dedicated image-pusher service account. Customers never need to know about Google Cloud, IAM, or service accounts — they just Image.build(...).push().

The password is an OAuth2 access token valid for ~1 hour. The SDK caches it on disk and refreshes within 5 minutes of expiry.

Source code in core/src/armnet_core/models.py 
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
class RegistryCredentials(BaseModel):
    """Short-lived credentials for pushing to the platform's image registry.

    Returned by ``POST /registry/credentials``. The orchestrator mints
    these on demand by impersonating a dedicated image-pusher service
    account. Customers never need to know about Google Cloud, IAM, or
    service accounts \u2014 they just ``Image.build(...).push()``.

    The ``password`` is an OAuth2 access token valid for ~1 hour. The
    SDK caches it on disk and refreshes within 5 minutes of expiry.
    """

    registry: str = Field(
        ...,
        description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.",
    )
    namespace: str = Field(
        ...,
        description="Path under the registry the customer is allowed to "
        "push to: '<project>/<repo>/<customer-id>'. The full image ref "
        "is '<registry>/<namespace>/<image-name>:<tag>'.",
    )
    username: str = Field(
        ...,
        description="docker login username. Always 'oauth2accesstoken' for AR.",
    )
    password: str = Field(
        ...,
        description="OAuth2 access token. Treat as a secret \u2014 valid for "
        "~1 hour and grants write access to the namespace above.",
    )
    expires_at: datetime = Field(
        ...,
        description="Wall-clock time at which the password becomes invalid.",
    )

registry class-attribute instance-attribute 

registry: str = Field(..., description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.")

namespace class-attribute instance-attribute 

namespace: str = Field(..., description="Path under the registry the customer is allowed to push to: '<project>/<repo>/<customer-id>'. The full image ref is '<registry>/<namespace>/<image-name>:<tag>'.")

username class-attribute instance-attribute 

username: str = Field(..., description="docker login username. Always 'oauth2accesstoken' for AR.")

password class-attribute instance-attribute 

password: str = Field(..., description='OAuth2 access token. Treat as a secret — valid for ~1 hour and grants write access to the namespace above.')

expires_at class-attribute instance-attribute 

expires_at: datetime = Field(..., description='Wall-clock time at which the password becomes invalid.')

RemoteExecutionError

Bases: RuntimeError

Raised by :meth:JobResult.raise_for_status for non-succeeded results.

The exception's __str__ includes the underlying status, any platform-side error, and the user's traceback (if any), so an unhandled raise prints all the diagnostic context an operator needs. The original :class:JobResult is available as :attr:result for structured access.

Source code in core/src/armnet_core/models.py 
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
class RemoteExecutionError(RuntimeError):
    """Raised by :meth:`JobResult.raise_for_status` for non-succeeded results.

    The exception's ``__str__`` includes the underlying status, any
    platform-side ``error``, and the user's ``traceback`` (if any), so an
    unhandled raise prints all the diagnostic context an operator needs.
    The original :class:`JobResult` is available as :attr:`result` for
    structured access.
    """

    def __init__(self, result: "JobResult") -> None:
        self.result = result
        super().__init__(str(result))

result instance-attribute 

result = result

RobotArmConfig

Bases: BaseModel

Named arm configuration for multi-arm (bimanual) cells.

The runtime derives each arm's connector endpoint from the cell-level robot_connector_endpoint plus the arm name; port is the physical serial device or SocketCAN interface that arm is wired to on the edge Pi, used by the Fleet Agent to provision the (bimanual) edge connector with the per-arm port mapping (--arm-ports). Optional because the edge can also be provisioned out-of-band, but set it so the agent can bring a bimanual cell up itself.

Source code in core/src/armnet_core/models.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
class RobotArmConfig(BaseModel):
    """Named arm configuration for multi-arm (bimanual) cells.

    The runtime derives each arm's connector endpoint from the cell-level
    ``robot_connector_endpoint`` plus the arm name; ``port`` is the physical
    serial device or SocketCAN interface that arm is wired to on the edge Pi,
    used by the Fleet Agent to provision the (bimanual) edge connector with
    the per-arm port mapping (``--arm-ports``). Optional because the edge can
    also be provisioned out-of-band, but set it so the agent can bring a
    bimanual cell up itself.
    """

    model_config = ConfigDict(extra="forbid")

    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    rest_position: Optional[dict[str, float]] = None
    safety_limit: Optional[float] = None
    safety_delta_degrees: Optional[float] = None
    power_plug_ip: Optional[str] = None
    port: Optional[str] = None
    rail: Optional[RailConfig] = None
    yam: Optional[YamArmDriverConfig] = None

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

robot_id class-attribute instance-attribute 

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute 

calibration_dir: Optional[Path] = None

calibration_file_path class-attribute instance-attribute 

calibration_file_path: Optional[Path] = None

rest_position class-attribute instance-attribute 

rest_position: Optional[dict[str, float]] = None

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: Optional[float] = None

power_plug_ip class-attribute instance-attribute 

power_plug_ip: Optional[str] = None

port class-attribute instance-attribute 

port: Optional[str] = None

rail class-attribute instance-attribute 

rail: Optional[RailConfig] = None

yam class-attribute instance-attribute 

yam: Optional[YamArmDriverConfig] = None

RobotCellConfig

Bases: BaseModel

Physical robot cell configuration loaded from robot_cell.json.

This is operator-owned configuration, not job input. The cell and local Docker runner use it to inject the same robot metadata into the runtime context for every job that lands on this cell.

The JSON file may use a nested structure with top-level groups: cell, robot, data_interface, policy. A model validator flattens nested input into the canonical flat field set for backward compatibility with all existing consumers.

Source code in core/src/armnet_core/models.py 
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
class RobotCellConfig(BaseModel):
    """Physical robot cell configuration loaded from ``robot_cell.json``.

    This is operator-owned configuration, not job input. The cell and local
    Docker runner use it to inject the same robot metadata into the runtime
    context for every job that lands on this cell.

    The JSON file may use a nested structure with top-level groups:
    ``cell``, ``robot``, ``data_interface``, ``policy``. A model validator
    flattens nested input into the canonical flat field set for backward
    compatibility with all existing consumers.
    """

    nats_url: Optional[str] = None
    cell_id: Optional[str] = None
    embodiment: Embodiment = DEFAULT_EMBODIMENT
    # The task being run right now, stamped in per job. None outside a job: the
    # cell is set up for an environment, not a task, and hosts all of them.
    task: Optional[Task] = None
    connector_socket_path: Optional[str] = None
    connector_tcp: Optional[str] = None
    robot_connector_endpoint: Optional[str] = None
    operator_call_endpoint: Optional[str] = "tcp://127.0.0.1:9877"
    robot_port: Optional[str] = None
    robot_power_plug_ip: Optional[str] = None
    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    camera_configs: dict[str, dict[str, Any]] = Field(default_factory=dict)
    # Immutable DB snapshot selected when the cell starts. Keeping it in the
    # runtime config prevents a later activation from changing an active job.
    scene_calibration: Optional[SceneCalibration] = None
    arms: dict[str, RobotArmConfig] = Field(default_factory=dict)
    rail: Optional[RailConfig] = None
    safety_limit: Optional[float] = Field(
        default=None,
        description=(
            "Runtime-facing relative action safety limit for robot SDK configs. "
            "SO-101 cells expose 30 degrees; embodiments with connector-only "
            "safety, such as ARX5, leave this unset."
        ),
    )
    docker_gpus: Optional[str] = None
    language_instruction: Optional[str] = None
    completion_enabled: bool = False
    completion_threshold: float = 0.9
    completion_camera: Optional[str] = None
    completion_min_interval_s: float = 1.0
    completion_model_path: str = "robometer/Robometer-4B"
    completion_chunk_seconds: float = 2.0
    completion_sample_fps: float = 4.0
    arm_rest_position: Optional[dict[str, float]] = None
    # Robot motion/safety params (pushed to edge connector at job start)
    safety_delta_degrees: Optional[float] = 30.0
    goal_velocity_limit: Optional[int] = Field(default=None, ge=1, le=3250)
    acceleration_limit: Optional[int] = Field(default=None, ge=1, le=254)
    max_command_speed_deg_s: Optional[float] = Field(default=300.0, gt=0)
    max_command_oscillation_deg_s2: Optional[float] = Field(default=4800.0, gt=0)
    rest_return_seconds: float = 4.0
    rest_return_hz: float = 20.0
    gripper_release_seconds: float = 1.0
    gripper_motor_name: str = "gripper"
    gripper_release_offset: float = 10.0
    current_gate: CurrentGateConfig = Field(default_factory=UniformCurrentGateConfig)
    tracking_stall: TrackingStallConfig = Field(
        default_factory=TrackingStallConfig
    )
    # Data interface params
    tcp_read_timeout: float = 10.0
    jpeg_quality: int = 0
    # Cell infra
    shelly_timeout: float = 2.0
    # What this cell is set up to work on, and that environment's own settings.
    # The settings stay an opaque dict: only the package implementing the
    # environment understands them, and armnet-core must remain pydantic-only.
    # from_file() may load them from environment_config_file; they are always
    # serialized inline into ARMNET_CELL_CONFIG so containers need no bind mount.
    environment: Optional[str] = None
    environment_config: dict[str, Any] = Field(default_factory=dict)
    # Workcell lightbox: HTTP-controlled LED dimmer on the cell's own AP (e.g.
    # "http://lightbox.local"). Only the edge host can reach it, so the cell
    # routes brightness through the edge connector at job start. None = no
    # lightbox on this cell. Brightness/freq/fade are cell defaults (JSON);
    # jobs may override brightness via args.lightbox_brightness and frequency
    # via args.lightbox_frequency. size_cm / light_sets identify the physical
    # box (not sent to the ESP) so PWM defaults are lightbox-specific, not
    # cell-specific.
    lightbox_url: Optional[str] = None
    lightbox_size_cm: Optional[int] = None  # enclosure edge length; identity only
    lightbox_light_sets: Optional[int] = None  # LED strips in the box; identity only
    lightbox_default_brightness: int = 30  # percent 0-100
    lightbox_default_frequency: int = 2000  # PWM Hz
    lightbox_default_fade: int = 1000  # ms ramp on /set
    # Spread of the structured-variation draw around the default brightness,
    # in the same percent units. Zero holds the light at its default even when
    # a job asks for variation.
    #
    # 11 rather than a round 10 because a dry run found the light barely
    # changing: it widens the envelope to [2.5, 57.5], which is about as far as
    # a symmetric draw around a default of 30 can reach before the low end
    # bottoms out at zero.
    lightbox_variation_std: float = Field(default=11.0, ge=0)

    @classmethod
    def _flatten_nested(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Flatten nested JSON groups into the canonical flat field set."""
        flat: dict[str, Any] = {}

        cell = data.pop("cell", None)
        if isinstance(cell, dict):
            flat["nats_url"] = cell.get("nats_url")
            flat["cell_id"] = cell.get("cell_id")
            flat["connector_socket_path"] = cell.get("connector_socket_path")
            flat["connector_tcp"] = cell.get("connector_tcp")
            flat["robot_connector_endpoint"] = cell.get("robot_connector_endpoint")
            flat["operator_call_endpoint"] = cell.get("operator_call_endpoint")
            flat["robot_port"] = cell.get("robot_port")
            flat["docker_gpus"] = cell.get("docker_gpus")
            flat["robot_power_plug_ip"] = cell.get("power_plug_ip")
            flat["shelly_timeout"] = cell.get("shelly_timeout", 2.0)

        robot = data.pop("robot", None)
        if isinstance(robot, dict):
            if "current_gate_enabled" in robot:
                raise ValueError(
                    "robot.current_gate_enabled is no longer supported; use "
                    "robot.current_gate.mode (uniform, percentile, or off)"
                )
            flat["embodiment"] = robot.get("embodiment")
            flat["robot_id"] = robot.get("robot_id")
            flat["calibration_dir"] = robot.get("calibration_dir")
            flat["calibration_file_path"] = robot.get("calibration_file_path")
            flat["arm_rest_position"] = robot.get("rest_position")
            flat["camera_configs"] = robot.get("camera_configs", {})
            flat["arms"] = robot.get("arms", {})
            flat["rail"] = robot.get("rail")
            if robot.get("embodiment") == BIMANUAL_YAM_EMBODIMENT:
                if "safety_delta_degrees" in robot:
                    flat["safety_delta_degrees"] = robot["safety_delta_degrees"]
            else:
                flat["safety_delta_degrees"] = robot.get("safety_delta_degrees", 30.0)
            flat["goal_velocity_limit"] = robot.get("goal_velocity_limit")
            flat["acceleration_limit"] = robot.get("acceleration_limit")
            flat["max_command_speed_deg_s"] = robot.get(
                "max_command_speed_deg_s", 300.0
            )
            flat["max_command_oscillation_deg_s2"] = robot.get(
                "max_command_oscillation_deg_s2", 4800.0
            )
            flat["rest_return_seconds"] = robot.get("rest_return_seconds", 4.0)
            flat["rest_return_hz"] = robot.get("rest_return_hz", 20.0)
            flat["gripper_release_seconds"] = robot.get("gripper_release_seconds", 1.0)
            flat["gripper_motor_name"] = robot.get("gripper_motor_name", "gripper")
            flat["gripper_release_offset"] = robot.get("gripper_release_offset", 10.0)
            if "current_gate" in robot:
                flat["current_gate"] = robot["current_gate"]
            if "tracking_stall" in robot:
                flat["tracking_stall"] = robot["tracking_stall"]

        data_interface = data.pop("data_interface", None)
        if isinstance(data_interface, dict):
            flat["tcp_read_timeout"] = data_interface.get("tcp_read_timeout", 10.0)
            flat["jpeg_quality"] = data_interface.get("jpeg_quality", 0)

        # The environment's settings live in a block named after it and are
        # passed through whole. Listing the keys we expect, as every other block
        # here does, is exactly how a new setting gets silently dropped on its
        # way to the cell.
        environment = data.pop("environment", None)
        if environment:
            flat["environment"] = environment
            flat["environment_config"] = data.pop(environment, None) or {}

        lightbox = data.pop("lightbox", None)
        if isinstance(lightbox, dict):
            flat["lightbox_url"] = lightbox.get("url")
            if lightbox.get("size_cm") is not None:
                flat["lightbox_size_cm"] = lightbox["size_cm"]
            if lightbox.get("light_sets") is not None:
                flat["lightbox_light_sets"] = lightbox["light_sets"]
            if lightbox.get("default_brightness") is not None:
                flat["lightbox_default_brightness"] = lightbox["default_brightness"]
            if lightbox.get("default_frequency") is not None:
                flat["lightbox_default_frequency"] = lightbox["default_frequency"]
            if lightbox.get("default_fade") is not None:
                flat["lightbox_default_fade"] = lightbox["default_fade"]
            if lightbox.get("variation_std") is not None:
                flat["lightbox_variation_std"] = lightbox["variation_std"]

        policy = data.pop("policy", None)
        if isinstance(policy, dict):
            # NOTE: ``task`` / ``language_instruction`` are intentionally NOT read
            # from the config file. They belong to the job, not the cell: a cell
            # hosts every task its environment offers and is told which one this
            # is when the job arrives. Any ``policy.task`` /
            # ``policy.language_instruction`` left in a JSON file is ignored so a
            # stale copy can't override it.
            completion = policy.get("completion", {})
            if isinstance(completion, dict):
                flat["completion_enabled"] = completion.get("enabled", False)
                flat["completion_threshold"] = completion.get("threshold", 0.9)
                flat["completion_camera"] = completion.get("camera")
                flat["completion_min_interval_s"] = completion.get("min_interval_s", 1.0)
                flat["completion_model_path"] = completion.get(
                    "model_path", "robometer/Robometer-4B"
                )
                flat["completion_chunk_seconds"] = completion.get("chunk_seconds", 2.0)
                flat["completion_sample_fps"] = completion.get("sample_fps", 4.0)

        keep_none = set()
        if (
            isinstance(robot, dict)
            and robot.get("embodiment") == BIMANUAL_YAM_EMBODIMENT
        ):
            keep_none.add("safety_delta_degrees")
        if isinstance(robot, dict):
            keep_none.update(
                key
                for key in (
                    "max_command_speed_deg_s",
                    "max_command_oscillation_deg_s2",
                )
                if key in robot
            )
        # Remove None values so Pydantic defaults apply, except fields whose
        # explicit null has operator-owned disable semantics.
        flat = {k: v for k, v in flat.items() if v is not None or k in keep_none}
        # Merge: explicit flat fields in data take precedence over unpacked nested
        flat.update(data)
        return flat

    @model_validator(mode="before")
    @classmethod
    def _accept_nested_format(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        robot = data.get("robot")
        embodiment = data.get("embodiment")
        if isinstance(robot, dict):
            embodiment = robot.get("embodiment", embodiment)
        if embodiment == BIMANUAL_YAM_EMBODIMENT:
            if "lightbox" in data:
                raise ValueError("bimanual-yam rejects lightbox")
            if isinstance(robot, dict):
                for key in _BIMANUAL_YAM_ROBOT_FORBIDDEN:
                    if key in robot:
                        raise ValueError(f"bimanual-yam rejects {key}")
        if "current_gate_enabled" in data or (
            isinstance(robot, dict) and "current_gate_enabled" in robot
        ):
            raise ValueError(
                "current_gate_enabled is no longer supported; use "
                "robot.current_gate.mode (uniform, percentile, or off)"
            )
        # ``environment`` is both a nested-file group and a flat RobotCellConfig
        # field. Runner dumps include it without a ``robot`` block; those are
        # already-flat runtime payloads, not operator JSON.
        had_nested_robot = isinstance(robot, dict)
        if any(key in data for key in ("cell", "robot", "policy", "data_interface", "environment", "lightbox")):
            data = cls._flatten_nested(dict(data))
            embodiment = data.get("embodiment", embodiment)
        if (
            embodiment == BIMANUAL_YAM_EMBODIMENT
            and not had_nested_robot
            and "safety_delta_degrees" not in data
        ):
            if not isinstance(data, dict):
                return data
            data = dict(data)
            data["safety_delta_degrees"] = None
        return data

    @model_validator(mode="after")
    def _validate_named_arms(self) -> "RobotCellConfig":
        if self.embodiment == BIMANUAL_YAM_EMBODIMENT:
            _require_yam_serial_safety_explicitly_off(self)
            return _validate_bimanual_yam_cell(self)
        if self.safety_delta_degrees is None:
            self = self.model_copy(update={"safety_delta_degrees": 30.0})
        if self.arms and not {"left", "right"}.issubset(self.arms):
            raise ValueError("bimanual robot.arms must include both 'left' and 'right'")
        return self

    @classmethod
    def from_file(cls, path: str | Path) -> "RobotCellConfig":
        config_path = Path(path).expanduser().resolve()
        raw_config = json.loads(config_path.read_text())
        environment_config_file = raw_config.pop("environment_config_file", None)
        if environment_config_file:
            environment = raw_config.get("environment")
            if "environment_config" in raw_config or (
                isinstance(environment, str) and environment in raw_config
            ):
                raise ValueError(
                    "robot cell config cannot define both environment_config "
                    "and environment_config_file"
                )
            environment_path = Path(environment_config_file).expanduser()
            if not environment_path.is_absolute():
                environment_path = config_path.parent / environment_path
            raw_config["environment_config"] = json.loads(environment_path.read_text())
        config = cls.model_validate(raw_config)
        update: dict[str, Any] = {}
        if config.calibration_dir and not config.calibration_dir.is_absolute():
            update["calibration_dir"] = config_path.parent / config.calibration_dir
        if config.calibration_file_path and not config.calibration_file_path.is_absolute():
            update["calibration_file_path"] = config_path.parent / config.calibration_file_path
        if config.arms:
            arms: dict[str, RobotArmConfig] = {}
            changed = False
            for name, arm in config.arms.items():
                arm_update: dict[str, Path] = {}
                if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                    arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
                if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                    arm_update["calibration_file_path"] = (
                        config_path.parent / arm.calibration_file_path
                    )
                arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
                changed = changed or bool(arm_update)
            if changed:
                update["arms"] = arms
        return config.model_copy(update=update) if update else config

nats_url class-attribute instance-attribute 

nats_url: Optional[str] = None

cell_id class-attribute instance-attribute 

cell_id: Optional[str] = None

embodiment class-attribute instance-attribute 

embodiment: Embodiment = DEFAULT_EMBODIMENT

task class-attribute instance-attribute 

task: Optional[Task] = None

connector_socket_path class-attribute instance-attribute 

connector_socket_path: Optional[str] = None

connector_tcp class-attribute instance-attribute 

connector_tcp: Optional[str] = None

robot_connector_endpoint class-attribute instance-attribute 

robot_connector_endpoint: Optional[str] = None

operator_call_endpoint class-attribute instance-attribute 

operator_call_endpoint: Optional[str] = 'tcp://127.0.0.1:9877'

robot_port class-attribute instance-attribute 

robot_port: Optional[str] = None

robot_power_plug_ip class-attribute instance-attribute 

robot_power_plug_ip: Optional[str] = None

robot_id class-attribute instance-attribute 

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute 

calibration_dir: Optional[Path] = None

calibration_file_path class-attribute instance-attribute 

calibration_file_path: Optional[Path] = None

camera_configs class-attribute instance-attribute 

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

scene_calibration class-attribute instance-attribute 

scene_calibration: Optional[SceneCalibration] = None

arms class-attribute instance-attribute 

arms: dict[str, RobotArmConfig] = Field(default_factory=dict)

rail class-attribute instance-attribute 

rail: Optional[RailConfig] = None

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = Field(default=None, description='Runtime-facing relative action safety limit for robot SDK configs. SO-101 cells expose 30 degrees; embodiments with connector-only safety, such as ARX5, leave this unset.')

docker_gpus class-attribute instance-attribute 

docker_gpus: Optional[str] = None

language_instruction class-attribute instance-attribute 

language_instruction: Optional[str] = None

completion_enabled class-attribute instance-attribute 

completion_enabled: bool = False

completion_threshold class-attribute instance-attribute 

completion_threshold: float = 0.9

completion_camera class-attribute instance-attribute 

completion_camera: Optional[str] = None

completion_min_interval_s class-attribute instance-attribute 

completion_min_interval_s: float = 1.0

completion_model_path class-attribute instance-attribute 

completion_model_path: str = 'robometer/Robometer-4B'

completion_chunk_seconds class-attribute instance-attribute 

completion_chunk_seconds: float = 2.0

completion_sample_fps class-attribute instance-attribute 

completion_sample_fps: float = 4.0

arm_rest_position class-attribute instance-attribute 

arm_rest_position: Optional[dict[str, float]] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: Optional[float] = 30.0

goal_velocity_limit class-attribute instance-attribute 

goal_velocity_limit: Optional[int] = Field(default=None, ge=1, le=3250)

acceleration_limit class-attribute instance-attribute 

acceleration_limit: Optional[int] = Field(default=None, ge=1, le=254)

max_command_speed_deg_s class-attribute instance-attribute 

max_command_speed_deg_s: Optional[float] = Field(default=300.0, gt=0)

max_command_oscillation_deg_s2 class-attribute instance-attribute 

max_command_oscillation_deg_s2: Optional[float] = Field(default=4800.0, gt=0)

rest_return_seconds class-attribute instance-attribute 

rest_return_seconds: float = 4.0

rest_return_hz class-attribute instance-attribute 

rest_return_hz: float = 20.0

gripper_release_seconds class-attribute instance-attribute 

gripper_release_seconds: float = 1.0

gripper_motor_name class-attribute instance-attribute 

gripper_motor_name: str = 'gripper'

gripper_release_offset class-attribute instance-attribute 

gripper_release_offset: float = 10.0

current_gate class-attribute instance-attribute 

current_gate: CurrentGateConfig = Field(default_factory=UniformCurrentGateConfig)

tracking_stall class-attribute instance-attribute 

tracking_stall: TrackingStallConfig = Field(default_factory=TrackingStallConfig)

tcp_read_timeout class-attribute instance-attribute 

tcp_read_timeout: float = 10.0

jpeg_quality class-attribute instance-attribute 

jpeg_quality: int = 0

shelly_timeout class-attribute instance-attribute 

shelly_timeout: float = 2.0

environment class-attribute instance-attribute 

environment: Optional[str] = None

environment_config class-attribute instance-attribute 

environment_config: dict[str, Any] = Field(default_factory=dict)

lightbox_url class-attribute instance-attribute 

lightbox_url: Optional[str] = None

lightbox_size_cm class-attribute instance-attribute 

lightbox_size_cm: Optional[int] = None

lightbox_light_sets class-attribute instance-attribute 

lightbox_light_sets: Optional[int] = None

lightbox_default_brightness class-attribute instance-attribute 

lightbox_default_brightness: int = 30

lightbox_default_frequency class-attribute instance-attribute 

lightbox_default_frequency: int = 2000

lightbox_default_fade class-attribute instance-attribute 

lightbox_default_fade: int = 1000

lightbox_variation_std class-attribute instance-attribute 

lightbox_variation_std: float = Field(default=11.0, ge=0)

from_file classmethod 

from_file(path: str | Path) -> 'RobotCellConfig'
Source code in core/src/armnet_core/models.py 
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
@classmethod
def from_file(cls, path: str | Path) -> "RobotCellConfig":
    config_path = Path(path).expanduser().resolve()
    raw_config = json.loads(config_path.read_text())
    environment_config_file = raw_config.pop("environment_config_file", None)
    if environment_config_file:
        environment = raw_config.get("environment")
        if "environment_config" in raw_config or (
            isinstance(environment, str) and environment in raw_config
        ):
            raise ValueError(
                "robot cell config cannot define both environment_config "
                "and environment_config_file"
            )
        environment_path = Path(environment_config_file).expanduser()
        if not environment_path.is_absolute():
            environment_path = config_path.parent / environment_path
        raw_config["environment_config"] = json.loads(environment_path.read_text())
    config = cls.model_validate(raw_config)
    update: dict[str, Any] = {}
    if config.calibration_dir and not config.calibration_dir.is_absolute():
        update["calibration_dir"] = config_path.parent / config.calibration_dir
    if config.calibration_file_path and not config.calibration_file_path.is_absolute():
        update["calibration_file_path"] = config_path.parent / config.calibration_file_path
    if config.arms:
        arms: dict[str, RobotArmConfig] = {}
        changed = False
        for name, arm in config.arms.items():
            arm_update: dict[str, Path] = {}
            if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
            if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                arm_update["calibration_file_path"] = (
                    config_path.parent / arm.calibration_file_path
                )
            arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
            changed = changed or bool(arm_update)
        if changed:
            update["arms"] = arms
    return config.model_copy(update=update) if update else config

SecretCreateRequest

Bases: BaseModel

Create or replace one user secret.

Source code in core/src/armnet_core/models.py 
1632
1633
1634
1635
1636
class SecretCreateRequest(BaseModel):
    """Create or replace one user secret."""

    name: str
    value: str

name instance-attribute 

name: str

value instance-attribute 

value: str

SecretInfo

Bases: BaseModel

Secret metadata returned to clients. Never includes secret values.

Source code in core/src/armnet_core/models.py 
1620
1621
1622
1623
class SecretInfo(BaseModel):
    """Secret metadata returned to clients. Never includes secret values."""

    name: str

name instance-attribute 

name: str

SecretList

Bases: BaseModel

List of secret metadata for the authenticated user.

Source code in core/src/armnet_core/models.py 
1626
1627
1628
1629
class SecretList(BaseModel):
    """List of secret metadata for the authenticated user."""

    secrets: list[SecretInfo] = Field(default_factory=list)

secrets class-attribute instance-attribute 

secrets: list[SecretInfo] = Field(default_factory=list)

TaskInfo

Bases: BaseModel

One known task, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
1607
1608
1609
1610
1611
class TaskInfo(BaseModel):
    """One known task, as stored in the orchestrator database."""

    slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")

description class-attribute instance-attribute 

description: Optional[str] = None

TaskList

Bases: BaseModel

The set of tasks the platform currently accepts (GET /tasks).

Source code in core/src/armnet_core/models.py 
1614
1615
1616
1617
class TaskList(BaseModel):
    """The set of tasks the platform currently accepts (GET /tasks)."""

    tasks: list[TaskInfo] = Field(default_factory=list)

tasks class-attribute instance-attribute 

tasks: list[TaskInfo] = Field(default_factory=list)

TeleopAction

Bases: BaseModel

A single remote-teleoperation action for a running job.

Sent by the client (sampling a local leader arm) to the orchestrator over a websocket, forwarded to the cell on the job's teleop NATS subject, and kept by the cell as a most-recent-value register. timestamp (unix seconds, on the client clock) is used to drop out-of-order/stale messages so the cell always holds the freshest action.

Source code in core/src/armnet_core/models.py 
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
class TeleopAction(BaseModel):
    """A single remote-teleoperation action for a running job.

    Sent by the client (sampling a local leader arm) to the orchestrator over a
    websocket, forwarded to the cell on the job's teleop NATS subject, and kept
    by the cell as a most-recent-value register. ``timestamp`` (unix seconds, on
    the client clock) is used to drop out-of-order/stale messages so the cell
    always holds the freshest action.
    """

    job_id: JobId
    timestamp: float
    # Joint-space command keyed like LeRobot's send_action input, e.g.
    # {"shoulder_pan.pos": 12.3, ...}.
    action: dict[str, float] = Field(default_factory=dict)

job_id instance-attribute 

job_id: JobId

timestamp instance-attribute 

timestamp: float

action class-attribute instance-attribute 

action: dict[str, float] = Field(default_factory=dict)

TeleopEvent

Bases: BaseModel

A discrete recording-control event for a running teleop job.

Travels the same path as :class:TeleopAction (client websocket → orchestrator → the job's teleop NATS subject → cell), but with different delivery semantics: actions are a freshest-wins stream where drops are fine, whereas events are rare, user-initiated commands that the cell queues in order so none is lost between runtime polls.

The event values mirror LeRobot's dataset-recording keyboard controls: Right Arrow ends the current episode and moves on (next_episode), Left Arrow discards and re-records it (rerecord_episode), and Esc stops the whole session (stop_recording).

Source code in core/src/armnet_core/models.py 
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
class TeleopEvent(BaseModel):
    """A discrete recording-control event for a running teleop job.

    Travels the same path as :class:`TeleopAction` (client websocket →
    orchestrator → the job's teleop NATS subject → cell), but with different
    delivery semantics: actions are a freshest-wins stream where drops are
    fine, whereas events are rare, user-initiated commands that the cell queues
    in order so none is lost between runtime polls.

    The ``event`` values mirror LeRobot's dataset-recording keyboard controls:
    Right Arrow ends the current episode and moves on (``next_episode``), Left
    Arrow discards and re-records it (``rerecord_episode``), and Esc stops the
    whole session (``stop_recording``).
    """

    job_id: JobId
    timestamp: float
    event: str = Field(..., pattern="^(next_episode|rerecord_episode|stop_recording)$")

job_id instance-attribute 

job_id: JobId

timestamp instance-attribute 

timestamp: float

event class-attribute instance-attribute 

event: str = Field(..., pattern='^(next_episode|rerecord_episode|stop_recording)$')

TrackingStallConfig

Bases: BaseModel

Persistent goal-versus-position obstacle detector owned by the cell.

Source code in core/src/armnet_core/models.py 
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
class TrackingStallConfig(BaseModel):
    """Persistent goal-versus-position obstacle detector owned by the cell."""

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    mode: Literal["off", "shadow", "enforce"] = "shadow"
    min_error_deg: float = Field(default=8.0, gt=0)
    release_error_deg: float = Field(default=4.0, gt=0)
    max_position_range_deg: float = Field(default=2.0, gt=0)
    duration_s: float = Field(default=2.0, gt=0)
    max_sample_gap_s: float = Field(default=0.5, gt=0)
    excluded_motors: tuple[str, ...] = ("gripper",)

    @model_validator(mode="after")
    def _release_has_hysteresis(self) -> "TrackingStallConfig":
        if self.release_error_deg >= self.min_error_deg:
            raise ValueError(
                "tracking_stall.release_error_deg must be below min_error_deg"
            )
        return self

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

mode class-attribute instance-attribute 

mode: Literal['off', 'shadow', 'enforce'] = 'shadow'

min_error_deg class-attribute instance-attribute 

min_error_deg: float = Field(default=8.0, gt=0)

release_error_deg class-attribute instance-attribute 

release_error_deg: float = Field(default=4.0, gt=0)

max_position_range_deg class-attribute instance-attribute 

max_position_range_deg: float = Field(default=2.0, gt=0)

duration_s class-attribute instance-attribute 

duration_s: float = Field(default=2.0, gt=0)

max_sample_gap_s class-attribute instance-attribute 

max_sample_gap_s: float = Field(default=0.5, gt=0)

excluded_motors class-attribute instance-attribute 

excluded_motors: tuple[str, ...] = ('gripper',)

UniformCurrentGateConfig

Bases: BaseModel

Two current limits applied to every joint: a fast one and a slow one.

One threshold cannot describe motor damage. Winding heat goes as the square of current against a time constant of tens of seconds, so a brief excursion near stall is harmless while a moderate load held for a minute is not. limit_counts is the peak tier and catches a crash or hard stall in duration_ms; sustained_limit_counts sits near the servo's continuous rating and catches the slow cook over sustained_duration_ms. Set the sustained limit to null to run the peak tier alone.

Source code in core/src/armnet_core/models.py 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
class UniformCurrentGateConfig(BaseModel):
    """Two current limits applied to every joint: a fast one and a slow one.

    One threshold cannot describe motor damage. Winding heat goes as the
    square of current against a time constant of tens of seconds, so a brief
    excursion near stall is harmless while a moderate load held for a minute
    is not. ``limit_counts`` is the peak tier and catches a crash or hard
    stall in ``duration_ms``; ``sustained_limit_counts`` sits near the servo's
    continuous rating and catches the slow cook over
    ``sustained_duration_ms``. Set the sustained limit to null to run the peak
    tier alone.
    """

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    mode: Literal["uniform"] = "uniform"
    limit_counts: int = Field(default=275, gt=0)
    duration_ms: float = Field(default=500.0, gt=0)
    sustained_limit_counts: Optional[int] = Field(default=120, gt=0)
    sustained_duration_ms: float = Field(default=5000.0, gt=0)

    @model_validator(mode="after")
    def _sustained_tier_is_the_lower_slower_one(self) -> "UniformCurrentGateConfig":
        # Either inversion parses fine and then quietly does nothing: a
        # sustained limit at or above the peak can never be the tier that
        # fires, and a window no longer than the peak's makes both tiers the
        # same test.
        if self.sustained_limit_counts is None:
            return self
        if self.sustained_limit_counts >= self.limit_counts:
            raise ValueError(
                "current_gate.sustained_limit_counts must be below limit_counts"
            )
        if self.sustained_duration_ms <= self.duration_ms:
            raise ValueError(
                "current_gate.sustained_duration_ms must exceed duration_ms"
            )
        return self

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

mode class-attribute instance-attribute 

mode: Literal['uniform'] = 'uniform'

limit_counts class-attribute instance-attribute 

limit_counts: int = Field(default=275, gt=0)

duration_ms class-attribute instance-attribute 

duration_ms: float = Field(default=500.0, gt=0)

sustained_limit_counts class-attribute instance-attribute 

sustained_limit_counts: Optional[int] = Field(default=120, gt=0)

sustained_duration_ms class-attribute instance-attribute 

sustained_duration_ms: float = Field(default=5000.0, gt=0)

VariationAxis

Bases: BaseModel

One continuously varied device setting, in the unit its API accepts.

default is the cell's calibrated value and std the spread of the draw around it. std_below and std_above replace that spread on one side of the default, so an axis can reach further one way than the other. minimum and maximum are the device's hard travel limits. What matters to callers is envelope: the clipped normal intersected with those limits, which is both the range the sampler can produce and the range an operator may ask for by hand.

Source code in core/src/armnet_core/models.py 
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
class VariationAxis(BaseModel):
    """One continuously varied device setting, in the unit its API accepts.

    ``default`` is the cell's calibrated value and ``std`` the spread of the
    draw around it. ``std_below`` and ``std_above`` replace that spread on one
    side of the default, so an axis can reach further one way than the other.
    ``minimum`` and ``maximum`` are the device's hard travel limits. What
    matters to callers is ``envelope``: the clipped normal intersected with
    those limits, which is both the range the sampler can produce and the
    range an operator may ask for by hand.
    """

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    default: float
    std: float = Field(default=0.0, ge=0.0)
    # Unset on a symmetric axis. They must not be serialized as null: clients
    # that predate these fields reject unknown keys, which is what took the
    # demo Space's availability lights offline.
    std_below: Optional[float] = Field(default=None, ge=0.0)
    std_above: Optional[float] = Field(default=None, ge=0.0)
    minimum: float
    maximum: float

    @model_serializer(mode="wrap")
    def _omit_unset_side_spreads(self, handler):  # noqa: ANN001
        dumped = handler(self)
        if isinstance(dumped, dict):
            if dumped.get("std_below") is None:
                dumped.pop("std_below", None)
            if dumped.get("std_above") is None:
                dumped.pop("std_above", None)
        return dumped

    @model_validator(mode="after")
    def _default_sits_inside_the_device_range(self) -> "VariationAxis":
        if self.minimum > self.maximum:
            raise ValueError(
                f"variation minimum {self.minimum} exceeds maximum {self.maximum}"
            )
        if not self.minimum <= self.default <= self.maximum:
            raise ValueError(
                f"variation default {self.default} is outside the device range "
                f"[{self.minimum}, {self.maximum}]"
            )
        return self

    @property
    def spread_below(self) -> float:
        """Spread applied when a draw lands below the default."""
        return self.std if self.std_below is None else self.std_below

    @property
    def spread_above(self) -> float:
        """Spread applied when a draw lands above the default."""
        return self.std if self.std_above is None else self.std_above

    def spread_for(self, z: float) -> float:
        """Which spread a signed normal draw uses. Zero stays on the default."""
        if z < 0:
            return self.spread_below
        if z > 0:
            return self.spread_above
        return 0.0

    @property
    def envelope(self) -> tuple[float, float]:
        """Lowest and highest value this axis may take."""
        return (
            max(self.minimum, self.default - VARIATION_CLIP_SIGMA * self.spread_below),
            min(self.maximum, self.default + VARIATION_CLIP_SIGMA * self.spread_above),
        )

    @property
    def is_clipped_by_the_device(self) -> bool:
        """Whether travel limits, not the sigma bound, decide the envelope.

        True means part of the intended distribution falls off the end of the
        device and piles up on a stop, so this cell varies less, and less
        symmetrically, than the configured spread suggests.
        """
        return (
            self.default - VARIATION_CLIP_SIGMA * self.spread_below < self.minimum
            or self.default + VARIATION_CLIP_SIGMA * self.spread_above > self.maximum
        )

    def permits(self, value: float) -> bool:
        """Whether ``value`` is one this axis could itself have produced."""
        low, high = self.envelope
        return low - _VARIATION_EPSILON <= value <= high + _VARIATION_EPSILON

    def describe_envelope(self, *, unit: str = "") -> str:
        low, high = self.envelope
        suffix = f" {unit}" if unit else ""
        return f"{low:g}{suffix} to {high:g}{suffix}"

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

default instance-attribute 

default: float

std class-attribute instance-attribute 

std: float = Field(default=0.0, ge=0.0)

std_below class-attribute instance-attribute 

std_below: Optional[float] = Field(default=None, ge=0.0)

std_above class-attribute instance-attribute 

std_above: Optional[float] = Field(default=None, ge=0.0)

minimum instance-attribute 

minimum: float

maximum instance-attribute 

maximum: float

spread_below property 

spread_below: float

Spread applied when a draw lands below the default.

spread_above property 

spread_above: float

Spread applied when a draw lands above the default.

envelope property 

envelope: tuple[float, float]

Lowest and highest value this axis may take.

is_clipped_by_the_device property 

is_clipped_by_the_device: bool

Whether travel limits, not the sigma bound, decide the envelope.

True means part of the intended distribution falls off the end of the device and piles up on a stop, so this cell varies less, and less symmetrically, than the configured spread suggests.

spread_for 

spread_for(z: float) -> float

Which spread a signed normal draw uses. Zero stays on the default.

Source code in core/src/armnet_core/models.py 
1022
1023
1024
1025
1026
1027
1028
def spread_for(self, z: float) -> float:
    """Which spread a signed normal draw uses. Zero stays on the default."""
    if z < 0:
        return self.spread_below
    if z > 0:
        return self.spread_above
    return 0.0

permits 

permits(value: float) -> bool

Whether value is one this axis could itself have produced.

Source code in core/src/armnet_core/models.py 
1051
1052
1053
1054
def permits(self, value: float) -> bool:
    """Whether ``value`` is one this axis could itself have produced."""
    low, high = self.envelope
    return low - _VARIATION_EPSILON <= value <= high + _VARIATION_EPSILON

describe_envelope 

describe_envelope(*, unit: str = '') -> str
Source code in core/src/armnet_core/models.py 
1056
1057
1058
1059
def describe_envelope(self, *, unit: str = "") -> str:
    low, high = self.envelope
    suffix = f" {unit}" if unit else ""
    return f"{low:g}{suffix} to {high:g}{suffix}"

VolumeCredentials

Bases: BaseModel

Short-lived credentials for a user's cloud-backed volume prefix.

Source code in core/src/armnet_core/models.py 
1579
1580
1581
1582
1583
1584
1585
class VolumeCredentials(BaseModel):
    """Short-lived credentials for a user's cloud-backed volume prefix."""

    bucket: str
    prefix: str
    access_token: str
    expires_at: datetime

bucket instance-attribute 

bucket: str

prefix instance-attribute 

prefix: str

access_token instance-attribute 

access_token: str

expires_at instance-attribute 

expires_at: datetime

WhoAmI

Bases: BaseModel

Authenticated API identity.

Source code in core/src/armnet_core/models.py 
1588
1589
1590
1591
class WhoAmI(BaseModel):
    """Authenticated API identity."""

    username: str

username instance-attribute 

username: str

YamArmDriverConfig

Bases: BaseModel

Pydantic copy of the YAM follower hardware settings owned by a cell.

This is a wire/config schema only: armnet-core stays pydantic-only and does not depend on the YAM hardware package. port lives on :class:RobotArmConfig. Per-arm rest uses six arm-joint radians plus a public gripper value in [0, 1], not internal gripper motor radians. Interactive zero-gravity is refused because the cell Pi has no operator at a prompt.

Source code in core/src/armnet_core/models.py 
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
class YamArmDriverConfig(BaseModel):
    """Pydantic copy of the YAM follower hardware settings owned by a cell.

    This is a wire/config schema only: armnet-core stays pydantic-only and does
    not depend on the YAM hardware package. ``port`` lives on
    :class:`RobotArmConfig`. Per-arm rest uses six arm-joint radians plus a
    public gripper value in ``[0, 1]``, not internal gripper motor radians.
    Interactive zero-gravity is refused because the cell Pi has no operator at
    a prompt.
    """

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    can_id_path: str
    bitrate: int = Field(default=1_000_000, gt=0)
    bustype: Literal["socketcan"] = "socketcan"
    motor_offsets: dict[str, float]
    motor_directions: dict[str, int]
    kp_gains: dict[str, float]
    kd_gains: dict[str, float]
    joint_limits: dict[str, tuple[float, float]]
    gripper_limits: tuple[float, float]
    use_gravity_compensation: bool = True
    gravity_comp_factor: float = 1.3
    gripper_type: Literal["crank_4310", "linear_3507", "linear_4310"] = "crank_4310"
    zero_gravity_mode: Literal[False] = False
    shutdown_zero_gravity_wait_for_enter: Literal[False] = False
    limit_gripper_force: float = Field(default=50.0, gt=0)
    lerobot_max_step: float | None = Field(default=None, gt=0)
    lerobot_gripper_max_step: float | None = Field(default=None, gt=0)

    @field_validator("can_id_path")
    @classmethod
    def _stable_can_id_path(cls, value: str) -> str:
        return _stable_nonempty_id_path("can_id_path", value, require_usb=True)

    @field_validator("motor_offsets", "kp_gains", "kd_gains")
    @classmethod
    def _seven_finite_gains(cls, value: dict[str, Any], info: Any) -> dict[str, float]:
        return _finite_mapping(info.field_name, value, YAM_JOINTS)

    @field_validator("motor_directions")
    @classmethod
    def _plus_or_minus_one(cls, value: dict[str, Any]) -> dict[str, int]:
        _require_exact_keys("motor_directions", value, YAM_JOINTS)
        directions: dict[str, int] = {}
        for name in YAM_JOINTS:
            direction = int(value[name])
            if direction not in (-1, 1):
                raise ValueError(f"motor_directions[{name!r}] must be 1 or -1")
            directions[name] = direction
        return directions

    @field_validator("joint_limits")
    @classmethod
    def _six_ordered_limits(cls, value: dict[str, Any]) -> dict[str, tuple[float, float]]:
        _require_exact_keys("joint_limits", value, YAM_ARM_JOINTS)
        return {
            name: _ordered_limit_pair(f"joint_limits[{name!r}]", value[name])
            for name in YAM_ARM_JOINTS
        }

    @field_validator("gripper_limits")
    @classmethod
    def _finite_gripper_limits(cls, value: Any) -> tuple[float, float]:
        return _finite_pair("gripper_limits", value)

    @field_validator("gravity_comp_factor")
    @classmethod
    def _finite_gravity_comp_factor(cls, value: Any) -> float:
        return _finite("gravity_comp_factor", value)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

can_id_path instance-attribute 

can_id_path: str

bitrate class-attribute instance-attribute 

bitrate: int = Field(default=1000000, gt=0)

bustype class-attribute instance-attribute 

bustype: Literal['socketcan'] = 'socketcan'

motor_offsets instance-attribute 

motor_offsets: dict[str, float]

motor_directions instance-attribute 

motor_directions: dict[str, int]

kp_gains instance-attribute 

kp_gains: dict[str, float]

kd_gains instance-attribute 

kd_gains: dict[str, float]

joint_limits instance-attribute 

joint_limits: dict[str, tuple[float, float]]

gripper_limits instance-attribute 

gripper_limits: tuple[float, float]

use_gravity_compensation class-attribute instance-attribute 

use_gravity_compensation: bool = True

gravity_comp_factor class-attribute instance-attribute 

gravity_comp_factor: float = 1.3

gripper_type class-attribute instance-attribute 

gripper_type: Literal['crank_4310', 'linear_3507', 'linear_4310'] = 'crank_4310'

zero_gravity_mode class-attribute instance-attribute 

zero_gravity_mode: Literal[False] = False

shutdown_zero_gravity_wait_for_enter class-attribute instance-attribute 

shutdown_zero_gravity_wait_for_enter: Literal[False] = False

limit_gripper_force class-attribute instance-attribute 

limit_gripper_force: float = Field(default=50.0, gt=0)

lerobot_max_step class-attribute instance-attribute 

lerobot_max_step: float | None = Field(default=None, gt=0)

lerobot_gripper_max_step class-attribute instance-attribute 

lerobot_gripper_max_step: float | None = Field(default=None, gt=0)

SceneCalibration

Bases: BaseModel

One immutable manipulation-area calibration snapshot.

Source code in core/src/armnet_core/scene_calibration.py 
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
class SceneCalibration(BaseModel):
    """One immutable manipulation-area calibration snapshot."""

    model_config = ConfigDict(
        extra="forbid",
        frozen=True,
        allow_inf_nan=False,
    )

    id: str = Field(min_length=1)
    cell_id: str = Field(min_length=1)
    camera_name: str = Field(min_length=1)
    vertices: tuple[NormalizedPoint, ...]
    source_frame_width: int = Field(gt=0)
    source_frame_height: int = Field(gt=0)
    schema_version: int = Field(default=1, gt=0)
    note: str | None = None
    created_at: datetime

    @field_validator("id", "cell_id", "camera_name")
    @classmethod
    def _nonempty_identifier(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("identifier must be nonempty")
        return value

    @field_validator("vertices", mode="before")
    @classmethod
    def _valid_polygon(cls, value: Any) -> tuple[NormalizedPoint, ...]:
        return validate_normalized_polygon(value)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', frozen=True, allow_inf_nan=False)

id class-attribute instance-attribute 

id: str = Field(min_length=1)

cell_id class-attribute instance-attribute 

cell_id: str = Field(min_length=1)

camera_name class-attribute instance-attribute 

camera_name: str = Field(min_length=1)

vertices instance-attribute 

vertices: tuple[NormalizedPoint, ...]

source_frame_width class-attribute instance-attribute 

source_frame_width: int = Field(gt=0)

source_frame_height class-attribute instance-attribute 

source_frame_height: int = Field(gt=0)

schema_version class-attribute instance-attribute 

schema_version: int = Field(default=1, gt=0)

note class-attribute instance-attribute 

note: str | None = None

created_at instance-attribute 

created_at: datetime

available_environments 

available_environments() -> list[str]

Names of every environment provided by an installed package.

Source code in core/src/armnet_core/environment.py 
212
213
214
215
def available_environments() -> list[str]:
    """Names of every environment provided by an installed package."""

    return sorted({entry.name for entry in _entry_points()})

environment_for 

environment_for(name: str) -> Environment

Load the environment registered under name.

Loading is deferred to here so that importing this module costs nothing: an environment's drivers are only imported by a process that actually intends to talk to one.

Source code in core/src/armnet_core/environment.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def environment_for(name: str) -> Environment:
    """Load the environment registered under ``name``.

    Loading is deferred to here so that importing this module costs nothing:
    an environment's drivers are only imported by a process that actually
    intends to talk to one.
    """

    for entry in _entry_points():
        if entry.name == name:
            return entry.load()()
    installed = available_environments()
    raise EnvironmentNotFound(
        f"no environment named {name!r} is installed"
        + (f" (installed: {', '.join(installed)})" if installed else "")
    )

task_environment 

task_environment(task: Optional[str]) -> Optional[str]

Name the environment that owns task, or None if unowned.

Used where the mapping is needed without a database — chiefly tests and local runs. The orchestrator resolves through the tasks table instead, so routing never depends on which environment packages it has installed.

Source code in core/src/armnet_core/environment.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def task_environment(task: Optional[str]) -> Optional[str]:
    """Name the environment that owns ``task``, or ``None`` if unowned.

    Used where the mapping is needed without a database — chiefly tests and
    local runs. The orchestrator resolves through the ``tasks`` table instead,
    so routing never depends on which environment packages it has installed.
    """

    if not task:
        return None
    for name in available_environments():
        if task in environment_for(name).tasks():
            return name
    return None

sample_manual_reset_layout 

sample_manual_reset_layout(config: ManualResetConfig, calibration: SceneCalibration | None, *, source_frame_width: int | None = None, source_frame_height: int | None = None, seed: int, reset_index: int, cell_id: str, task_slug: str, max_attempts: int = DEFAULT_LAYOUT_ATTEMPTS) -> ManualResetLayout

Sample a stable, non-overlapping task reset layout.

source_frame_width and source_frame_height may be omitted when a calibration is supplied, in which case its source dimensions are used. Every element gets at most max_attempts deterministic candidates.

Source code in core/src/armnet_core/manual_reset.py 
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
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
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def sample_manual_reset_layout(
    config: ManualResetConfig,
    calibration: SceneCalibration | None,
    *,
    source_frame_width: int | None = None,
    source_frame_height: int | None = None,
    seed: int,
    reset_index: int,
    cell_id: str,
    task_slug: str,
    max_attempts: int = DEFAULT_LAYOUT_ATTEMPTS,
) -> ManualResetLayout:
    """Sample a stable, non-overlapping task reset layout.

    ``source_frame_width`` and ``source_frame_height`` may be omitted when a
    calibration is supplied, in which case its source dimensions are used.
    Every element gets at most ``max_attempts`` deterministic candidates.
    """

    config = ManualResetConfig.model_validate(config)
    if reset_index < 0:
        raise ValueError("reset_index must be non-negative")
    if not cell_id.strip():
        raise ValueError("cell_id must be nonempty")
    if not task_slug.strip():
        raise ValueError("task_slug must be nonempty")
    if max_attempts <= 0:
        raise ValueError("max_attempts must be positive")

    if source_frame_width is None:
        source_frame_width = (
            calibration.source_frame_width if calibration is not None else None
        )
    if source_frame_height is None:
        source_frame_height = (
            calibration.source_frame_height if calibration is not None else None
        )
    if (
        source_frame_width is None
        or source_frame_height is None
        or source_frame_width <= 0
        or source_frame_height <= 0
    ):
        raise ValueError("positive source frame width and height are required")

    normalized_scene = (
        calibration.vertices if calibration is not None else FULL_FRAME_VERTICES
    )
    pixel_scene = tuple(
        normalized_to_pixel(
            vertex,
            width=source_frame_width,
            height=source_frame_height,
        )
        for vertex in normalized_scene
    )
    # Shirt sizes are camera-frame references, matching the rendered size guide.
    # The manipulation polygon still controls where objects may be placed and
    # which edges clearance is measured from, but changing that polygon must not
    # silently change what XS/S/M/L/XL mean.
    short_axis = min(source_frame_width, source_frame_height)

    triangles = tuple(
        tuple(
            normalized_to_pixel(
                point,
                width=source_frame_width,
                height=source_frame_height,
            )
            for point in triangle
        )
        for triangle in _triangulate(normalized_scene)
    )
    weighted_triangles = tuple(
        (triangle, polygon_area(triangle)) for triangle in triangles
    )
    total_area = sum(area for _, area in weighted_triangles)
    calibration_id = (
        calibration.id if calibration is not None else "full-normalized-frame"
    )

    render_elements: list[ManualResetElementGeometry] = []
    footprints: list[PixelFootprint] = []
    for element_index, element in enumerate(config.elements):
        clearance = EDGE_DISTANCE_SCALES[element.min_edge_distance] * short_axis
        for attempt in range(max_attempts):
            random = _StableRandom(
                [
                    int(seed),
                    reset_index,
                    cell_id,
                    calibration_id,
                    task_slug,
                    element_index,
                    element.name,
                    attempt,
                ]
            )
            rotation = (
                (2.0 * random.unit() - 1.0)
                * element.rotation_range_degrees
            )
            center = _sample_weighted_triangle(
                weighted_triangles,
                total_area,
                random,
            )
            footprint = _make_footprint(
                element,
                center=center,
                rotation_degrees=rotation,
                short_axis=short_axis,
            )
            if not _footprint_inside_scene(
                footprint,
                pixel_scene,
                clearance=clearance,
            ):
                continue
            if any(_footprints_overlap(footprint, other) for other in footprints):
                continue

            footprints.append(footprint)
            render_elements.append(
                _render_geometry(
                    element,
                    element_index=element_index,
                    footprint=footprint,
                    rotation_degrees=rotation,
                    short_axis=short_axis,
                    width=source_frame_width,
                    height=source_frame_height,
                )
            )
            break
        else:
            raise ManualResetLayoutError(
                "unable to place manual-reset element "
                f"{element_index + 1} ({element.name!r}) after "
                f"{max_attempts} deterministic attempts; its footprint and "
                "edge clearance do not fit without overlap"
            )

    return ManualResetLayout(
        source_frame_width=source_frame_width,
        source_frame_height=source_frame_height,
        elements=tuple(render_elements),
    )

canonical_customer_image 

canonical_customer_image(requested: str, *, registry: str, namespace_prefix: str, username: str) -> str

Build the only image ref an authenticated customer may execute.

Source code in core/src/armnet_core/image_refs.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def canonical_customer_image(
    requested: str,
    *,
    registry: str,
    namespace_prefix: str,
    username: str,
) -> str:
    """Build the only image ref an authenticated customer may execute."""

    selector = customer_image_selector(
        requested,
        registry=registry,
        namespace_prefix=namespace_prefix,
    )
    slug = customer_registry_slug(username)
    return f"{registry.rstrip('/')}/{namespace_prefix.strip('/')}/{slug}/{selector}"

customer_image_selector 

customer_image_selector(requested: str, *, registry: str, namespace_prefix: str) -> str

Return the customer-controlled suffix from a relative or platform ref.

For a full ref, the shape must be <registry>/<namespace-prefix>/<supplied-user>/<selector>. The supplied user component is deliberately ignored; callers reconstruct it from the authenticated identity.

Source code in core/src/armnet_core/image_refs.py 
 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
def customer_image_selector(
    requested: str,
    *,
    registry: str,
    namespace_prefix: str,
) -> str:
    """Return the customer-controlled suffix from a relative or platform ref.

    For a full ref, the shape must be
    ``<registry>/<namespace-prefix>/<supplied-user>/<selector>``. The supplied
    user component is deliberately ignored; callers reconstruct it from the
    authenticated identity.
    """

    registry = registry.strip().rstrip("/")
    prefix = namespace_prefix.strip().strip("/")
    if not registry or "/" in registry or not prefix:
        raise ImageReferenceError("platform image registry configuration is invalid")

    first = requested.split("/", 1)[0]
    explicit_registry = "/" in requested and (
        "." in first or ":" in first or first == "localhost"
    )
    if not explicit_registry:
        return validate_image_selector(requested)

    platform_base = f"{registry}/{prefix}/"
    if not requested.startswith(platform_base):
        raise ImageReferenceError(
            f"image must come from the platform registry {registry}/{prefix}"
        )
    remainder = requested[len(platform_base) :]
    supplied_user, separator, selector = remainder.partition("/")
    if not separator or not _CUSTOMER_SLUG.fullmatch(supplied_user):
        raise ImageReferenceError(
            "fully-qualified customer image must include a user namespace and image selector"
        )
    return validate_image_selector(selector)

customer_registry_slug 

customer_registry_slug(username: str) -> str

Current username→registry mapping, validated as one path component.

Existing production namespaces use lowercase usernames, so preserve that mapping. Account provisioning must prevent case-fold collisions (e.g. Alice and alice); this function rejects path separators and other characters rather than silently sanitizing them into collisions.

Source code in core/src/armnet_core/image_refs.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def customer_registry_slug(username: str) -> str:
    """Current username→registry mapping, validated as one path component.

    Existing production namespaces use lowercase usernames, so preserve that
    mapping. Account provisioning must prevent case-fold collisions (e.g.
    ``Alice`` and ``alice``); this function rejects path separators and other
    characters rather than silently sanitizing them into collisions.
    """

    slug = username.strip().lower()
    if not slug or not _CUSTOMER_SLUG.fullmatch(slug):
        raise ImageReferenceError(
            f"username {username!r} cannot be represented as a registry namespace"
        )
    return slug

require_owned_customer_image 

require_owned_customer_image(image: str, *, registry: str, namespace_prefix: str, username: str) -> None

Reject a persisted/dispatch image unless it is already canonical.

Source code in core/src/armnet_core/image_refs.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def require_owned_customer_image(
    image: str,
    *,
    registry: str,
    namespace_prefix: str,
    username: str,
) -> None:
    """Reject a persisted/dispatch image unless it is already canonical."""

    canonical = canonical_customer_image(
        image,
        registry=registry,
        namespace_prefix=namespace_prefix,
        username=username,
    )
    if image != canonical:
        raise ImageReferenceError(
            f"image is not owned by authenticated user {username!r}; expected {canonical!r}"
        )

validate_image_selector 

validate_image_selector(selector: str) -> str

Validate a repository path plus explicit tag or sha256 digest.

Source code in core/src/armnet_core/image_refs.py 
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
def validate_image_selector(selector: str) -> str:
    """Validate a repository path plus explicit tag or sha256 digest."""

    if not selector or selector != selector.strip():
        raise ImageReferenceError("image selector must be nonempty with no surrounding whitespace")
    if any(ch.isspace() for ch in selector) or "\\" in selector:
        raise ImageReferenceError("image selector must not contain whitespace or backslashes")
    if selector.startswith(("-", "/", ".")) or "//" in selector:
        raise ImageReferenceError("invalid image selector path")

    name_and_tag, at, digest = selector.partition("@")
    if at:
        if not _SHA256.fullmatch(digest):
            raise ImageReferenceError("image digest must be lowercase sha256:<64 hex>")
    elif "@" in digest:
        raise ImageReferenceError("invalid image digest")

    last_slash = name_and_tag.rfind("/")
    last_colon = name_and_tag.rfind(":")
    tag = None
    name = name_and_tag
    if last_colon > last_slash:
        name, tag = name_and_tag[:last_colon], name_and_tag[last_colon + 1 :]
        if not _TAG.fullmatch(tag):
            raise ImageReferenceError("invalid image tag")

    if tag is None and not at:
        raise ImageReferenceError("image selector must include an explicit tag or digest")
    components = name.split("/")
    if not components or any(not _NAME_COMPONENT.fullmatch(part) for part in components):
        raise ImageReferenceError(
            "image name must use lowercase path components containing letters, "
            "digits, '.', '_', or '-'"
        )
    return selector

canonical_job_id 

canonical_job_id(value: Any) -> str

Return one textual representation for UUID job ids.

Postgres returns dashed UUID text while older clients and NATS messages used the equivalent 32-character hex form. Canonicalizing at every model boundary prevents those spellings from becoming separate in-memory cache keys. Synthetic local-development ids remain unchanged.

Source code in core/src/armnet_core/models.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
def canonical_job_id(value: Any) -> str:
    """Return one textual representation for UUID job ids.

    Postgres returns dashed UUID text while older clients and NATS messages used
    the equivalent 32-character hex form. Canonicalizing at every model boundary
    prevents those spellings from becoming separate in-memory cache keys.
    Synthetic local-development ids remain unchanged.
    """
    text = str(value)
    try:
        return str(UUID(text))
    except ValueError:
        return text

normalized_to_pixel 

normalized_to_pixel(point: NormalizedPoint, *, width: int, height: int) -> NormalizedPoint

Scale a normalized point to source-frame pixel coordinates.

Source code in core/src/armnet_core/scene_calibration.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
def normalized_to_pixel(
    point: NormalizedPoint,
    *,
    width: int,
    height: int,
) -> NormalizedPoint:
    """Scale a normalized point to source-frame pixel coordinates."""
    if width <= 0 or height <= 0:
        raise ValueError("frame width and height must be positive")
    return point[0] * width, point[1] * height

pixel_to_normalized 

pixel_to_normalized(point: NormalizedPoint, *, width: int, height: int) -> NormalizedPoint

Scale source-frame pixel coordinates into the unit frame.

Source code in core/src/armnet_core/scene_calibration.py 
105
106
107
108
109
110
111
112
113
114
def pixel_to_normalized(
    point: NormalizedPoint,
    *,
    width: int,
    height: int,
) -> NormalizedPoint:
    """Scale source-frame pixel coordinates into the unit frame."""
    if width <= 0 or height <= 0:
        raise ValueError("frame width and height must be positive")
    return point[0] / width, point[1] / height

point_in_polygon 

point_in_polygon(point: NormalizedPoint, vertices: Iterable[NormalizedPoint]) -> bool

Return whether a point is inside or on the boundary of a polygon.

Source code in core/src/armnet_core/scene_calibration.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def point_in_polygon(
    point: NormalizedPoint,
    vertices: Iterable[NormalizedPoint],
) -> bool:
    """Return whether a point is inside or on the boundary of a polygon."""
    points = tuple(vertices)
    if len(points) < 3:
        return False
    x, y = point
    inside = False
    for index, start in enumerate(points):
        end = points[(index + 1) % len(points)]
        if _point_on_segment(point, start, end):
            return True
        if (start[1] > y) != (end[1] > y):
            crossing_x = start[0] + (y - start[1]) * (
                end[0] - start[0]
            ) / (end[1] - start[1])
            if x < crossing_x:
                inside = not inside
    return inside

polygon_area 

polygon_area(vertices: Iterable[NormalizedPoint]) -> float

Return the unsigned area of an ordered polygon.

Source code in core/src/armnet_core/scene_calibration.py 
25
26
27
28
def polygon_area(vertices: Iterable[NormalizedPoint]) -> float:
    """Return the unsigned area of an ordered polygon."""
    points = tuple(vertices)
    return abs(_signed_area(points))

sample_scene_target 

sample_scene_target(calibration: SceneCalibration | None, *, seed: int, reset_index: int, cell_id: str, edge_clearance: float = 0.01) -> NormalizedPoint

Choose a stable reset target inside the active calibrated polygon.

With no active calibration, the whole normalized frame is used. Candidate points are drawn from an ear-clipped triangulation, then the point furthest from an edge is selected. This makes a small edge clearance best-effort without ever moving a point outside a narrow polygon.

Source code in core/src/armnet_core/scene_calibration.py 
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
def sample_scene_target(
    calibration: SceneCalibration | None,
    *,
    seed: int,
    reset_index: int,
    cell_id: str,
    edge_clearance: float = 0.01,
) -> NormalizedPoint:
    """Choose a stable reset target inside the active calibrated polygon.

    With no active calibration, the whole normalized frame is used. Candidate
    points are drawn from an ear-clipped triangulation, then the point furthest
    from an edge is selected. This makes a small edge clearance best-effort
    without ever moving a point outside a narrow polygon.
    """
    if reset_index < 0:
        raise ValueError("reset_index must be non-negative")
    if not cell_id:
        raise ValueError("cell_id must be nonempty")
    if not math.isfinite(edge_clearance) or edge_clearance < 0:
        raise ValueError("edge_clearance must be finite and non-negative")

    vertices = calibration.vertices if calibration else FULL_FRAME_VERTICES
    calibration_id = calibration.id if calibration else "full-normalized-frame"
    material = json.dumps(
        [int(seed), reset_index, cell_id, calibration_id],
        separators=(",", ":"),
        ensure_ascii=True,
    ).encode("utf-8")
    random = _StableRandom(material)
    triangles = _triangulate(vertices)
    weighted = tuple((triangle, polygon_area(triangle)) for triangle in triangles)
    total_area = sum(area for _, area in weighted)

    best: NormalizedPoint | None = None
    best_clearance = -1.0
    # Multiple deterministic candidates make the requested clearance likely
    # for ordinary work areas while preserving support for very narrow shapes.
    for _ in range(64):
        selection = random.unit() * total_area
        triangle = weighted[-1][0]
        for candidate_triangle, area in weighted:
            selection -= area
            if selection <= 0:
                triangle = candidate_triangle
                break
        candidate = _sample_triangle(triangle, random.unit(), random.unit())
        if not point_in_polygon(candidate, vertices):
            continue
        clearance = min(
            _distance_to_segment(
                candidate,
                vertices[index],
                vertices[(index + 1) % len(vertices)],
            )
            for index in range(len(vertices))
        )
        if clearance > best_clearance:
            best = candidate
            best_clearance = clearance
        if best_clearance >= edge_clearance:
            break

    if best is None:  # Defensive fallback for floating-point edge cases.
        best = _triangle_centroid(weighted[0][0])
    if not point_in_polygon(best, vertices):
        raise RuntimeError("failed to sample a point inside the calibration polygon")
    return best

validate_normalized_polygon 

validate_normalized_polygon(vertices: Any, *, minimum_area: float = MIN_NORMALIZED_POLYGON_AREA) -> tuple[NormalizedPoint, ...]

Validate and normalize an ordered simple polygon in the unit frame.

Source code in core/src/armnet_core/scene_calibration.py 
31
32
33
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
def validate_normalized_polygon(
    vertices: Any,
    *,
    minimum_area: float = MIN_NORMALIZED_POLYGON_AREA,
) -> tuple[NormalizedPoint, ...]:
    """Validate and normalize an ordered simple polygon in the unit frame."""
    if not isinstance(vertices, (list, tuple)) or len(vertices) < 3:
        raise ValueError("polygon must contain at least 3 ordered vertices")

    points: list[NormalizedPoint] = []
    for index, vertex in enumerate(vertices):
        if not isinstance(vertex, (list, tuple)) or len(vertex) != 2:
            raise ValueError(f"vertex {index} must be an [x, y] pair")
        x, y = vertex
        if (
            isinstance(x, bool)
            or isinstance(y, bool)
            or not isinstance(x, (int, float))
            or not isinstance(y, (int, float))
        ):
            raise ValueError(  # noqa: TRY004 - public validator contract
                f"vertex {index} coordinates must be numbers"
            )
        point = (float(x), float(y))
        if not math.isfinite(point[0]) or not math.isfinite(point[1]):
            raise ValueError(f"vertex {index} coordinates must be finite")
        if not 0.0 <= point[0] <= 1.0 or not 0.0 <= point[1] <= 1.0:
            raise ValueError(f"vertex {index} coordinates must be in [0, 1]")
        points.append(point)

    if len(set(points)) != len(points):
        raise ValueError("polygon vertices must be distinct")
    if _has_self_intersection(points):
        raise ValueError("polygon must not self-intersect")
    if polygon_area(points) < minimum_area:
        raise ValueError(f"polygon area must be at least {minimum_area:g}")
    return tuple(points)

armnet_core.enums

Embodiment, environment and task wire-type aliases.

These are stable string slugs that appear in HTTP bodies, NATS subjects, env vars injected into customer containers, and cell configuration (for example "lerobot/so-101", "busybox", "busybox_left_toggle_down").

The three name different things:

Embodiment is the robot — what arms the cell has.

Environment is what the robot is working on: an instrumented BusyBox panel, a table of wooden blocks. A cell is assigned one, and can then run any task that environment offers without being reconfigured.

Task is a single job to perform within an environment. Users request a task; the platform resolves it to an environment to find a cell that can run it.

Valid embodiments and environments live in the orchestrator's Postgres database (see db/), which routing treats as the source of truth so it never depends on which environment packages happen to be installed. Task behaviour — the goal, how it is scored, how the scene is restored — lives in the environment's own package, because it is inseparable from that environment's instrumentation. The database mirrors those tasks for validation and routing.

Embodiment module-attribute 

Embodiment = str

Environment module-attribute 

Environment = str

Task module-attribute 

Task = str

DEFAULT_EMBODIMENT module-attribute 

DEFAULT_EMBODIMENT: Embodiment = 'lerobot/so-101'

DEFAULT_ENVIRONMENT module-attribute 

DEFAULT_ENVIRONMENT: Environment = 'busybox'

DEFAULT_TASK module-attribute 

DEFAULT_TASK: Task = 'assemble_block_tower'

armnet_core.models

External-facing wire types for the armnet platform.

Customer-touching models only — request/response bodies for the orchestrator HTTP API and the result payloads embedded in those responses. Internal NATS message envelopes (JobDispatch, CellHeartbeat) live in the separate armnet-protocol package alongside the NATS subject taxonomy.

Keep this module dependency-light (pydantic + stdlib only) so it can be imported from any component without dragging in HTTP or NATS clients.

JobId module-attribute 

JobId = Annotated[str, BeforeValidator(canonical_job_id)]

TerminalStatus module-attribute 

TerminalStatus = frozenset({JobStatus.SUCCEEDED, JobStatus.FAILED, JobStatus.TIMEOUT, JobStatus.CANCELLED})

BIMANUAL_YAM_EMBODIMENT module-attribute 

BIMANUAL_YAM_EMBODIMENT = 'lerobot/bimanual_yam'

YAM_ARM_JOINTS module-attribute 

YAM_ARM_JOINTS = ('shoulder_pan', 'shoulder_lift', 'elbow_flex', 'wrist_flex', 'wrist_roll', 'wrist_yaw')

YAM_JOINTS module-attribute 

YAM_JOINTS = YAM_ARM_JOINTS + ('gripper',)

YAM_CAMERA_NAMES module-attribute 

YAM_CAMERA_NAMES = ('top', 'left_wrist', 'right_wrist')

YAM_CAMERA_INDEX_OR_PATH module-attribute 

YAM_CAMERA_INDEX_OR_PATH = {'top': '/dev/video_top', 'left_wrist': '/dev/video_left_wrist', 'right_wrist': '/dev/video_right_wrist'}

YAM_ARM_PORTS module-attribute 

YAM_ARM_PORTS = {'left': 'can_yam_left', 'right': 'can_yam_right'}

CurrentGateConfig module-attribute 

CurrentGateConfig = Annotated[UniformCurrentGateConfig | PercentileCurrentGateConfig | OffCurrentGateConfig, Field(discriminator='mode')]

VARIATION_CLIP_SIGMA module-attribute 

VARIATION_CLIP_SIGMA = 2.5

CAMERA_PAN_RANGE module-attribute 

CAMERA_PAN_RANGE = (0.0, 180.0)

CAMERA_TILT_RANGE module-attribute 

CAMERA_TILT_RANGE = (15.0, 145.0)

DEFAULT_CAMERA_MOUNT_VARIATION_STD_DEGREES module-attribute 

DEFAULT_CAMERA_MOUNT_VARIATION_STD_DEGREES = 5.0

CAMERA_ENCODER_RANGE module-attribute 

CAMERA_ENCODER_RANGE = (0.0, 4095.0)

SERVO_MOUNT_STEPS_PER_DEGREE module-attribute 

SERVO_MOUNT_STEPS_PER_DEGREE = 4096.0 / 360.0

DEFAULT_SERVO_MOUNT_VARIATION_STD_STEPS module-attribute 

DEFAULT_SERVO_MOUNT_VARIATION_STD_STEPS = DEFAULT_CAMERA_MOUNT_VARIATION_STD_DEGREES * SERVO_MOUNT_STEPS_PER_DEGREE

TELEOP_EVENT_NEXT_EPISODE module-attribute 

TELEOP_EVENT_NEXT_EPISODE = 'next_episode'

TELEOP_EVENT_RERECORD_EPISODE module-attribute 

TELEOP_EVENT_RERECORD_EPISODE = 'rerecord_episode'

TELEOP_EVENT_STOP_RECORDING module-attribute 

TELEOP_EVENT_STOP_RECORDING = 'stop_recording'

TELEOP_EVENTS module-attribute 

TELEOP_EVENTS = (TELEOP_EVENT_NEXT_EPISODE, TELEOP_EVENT_RERECORD_EPISODE, TELEOP_EVENT_STOP_RECORDING)

JobStatus

Bases: str, Enum

Job lifecycle states.

A job is created SUBMITTED, becomes QUEUED after JetStream confirms durable storage, then DISPATCHED when an idle cell atomically claims it, RUNNING when that cell heartbeats ownership, and finally terminal.

Source code in core/src/armnet_core/models.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class JobStatus(str, Enum):
    """Job lifecycle states.

    A job is created ``SUBMITTED``, becomes ``QUEUED`` after JetStream confirms
    durable storage, then ``DISPATCHED`` when an idle cell atomically claims it,
    ``RUNNING`` when that cell heartbeats ownership, and finally terminal.
    """

    SUBMITTED = "submitted"
    QUEUED = "queued"
    DISPATCHED = "dispatched"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    TIMEOUT = "timeout"
    CANCELLED = "cancelled"

SUBMITTED class-attribute instance-attribute 

SUBMITTED = 'submitted'

QUEUED class-attribute instance-attribute 

QUEUED = 'queued'

DISPATCHED class-attribute instance-attribute 

DISPATCHED = 'dispatched'

RUNNING class-attribute instance-attribute 

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute 

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute 

FAILED = 'failed'

TIMEOUT class-attribute instance-attribute 

TIMEOUT = 'timeout'

CANCELLED class-attribute instance-attribute 

CANCELLED = 'cancelled'

CellRuntimeStatus

Bases: str, Enum

Current availability state for a robot cell.

Source code in core/src/armnet_core/models.py 
81
82
83
84
85
86
87
class CellRuntimeStatus(str, Enum):
    """Current availability state for a robot cell."""

    AVAILABLE = "available"
    OCCUPIED = "occupied"
    MAINTENANCE = "maintenance"
    OFFLINE = "offline"

AVAILABLE class-attribute instance-attribute 

AVAILABLE = 'available'

OCCUPIED class-attribute instance-attribute 

OCCUPIED = 'occupied'

MAINTENANCE class-attribute instance-attribute 

MAINTENANCE = 'maintenance'

OFFLINE class-attribute instance-attribute 

OFFLINE = 'offline'

YamArmDriverConfig

Bases: BaseModel

Pydantic copy of the YAM follower hardware settings owned by a cell.

This is a wire/config schema only: armnet-core stays pydantic-only and does not depend on the YAM hardware package. port lives on :class:RobotArmConfig. Per-arm rest uses six arm-joint radians plus a public gripper value in [0, 1], not internal gripper motor radians. Interactive zero-gravity is refused because the cell Pi has no operator at a prompt.

Source code in core/src/armnet_core/models.py 
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
class YamArmDriverConfig(BaseModel):
    """Pydantic copy of the YAM follower hardware settings owned by a cell.

    This is a wire/config schema only: armnet-core stays pydantic-only and does
    not depend on the YAM hardware package. ``port`` lives on
    :class:`RobotArmConfig`. Per-arm rest uses six arm-joint radians plus a
    public gripper value in ``[0, 1]``, not internal gripper motor radians.
    Interactive zero-gravity is refused because the cell Pi has no operator at
    a prompt.
    """

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    can_id_path: str
    bitrate: int = Field(default=1_000_000, gt=0)
    bustype: Literal["socketcan"] = "socketcan"
    motor_offsets: dict[str, float]
    motor_directions: dict[str, int]
    kp_gains: dict[str, float]
    kd_gains: dict[str, float]
    joint_limits: dict[str, tuple[float, float]]
    gripper_limits: tuple[float, float]
    use_gravity_compensation: bool = True
    gravity_comp_factor: float = 1.3
    gripper_type: Literal["crank_4310", "linear_3507", "linear_4310"] = "crank_4310"
    zero_gravity_mode: Literal[False] = False
    shutdown_zero_gravity_wait_for_enter: Literal[False] = False
    limit_gripper_force: float = Field(default=50.0, gt=0)
    lerobot_max_step: float | None = Field(default=None, gt=0)
    lerobot_gripper_max_step: float | None = Field(default=None, gt=0)

    @field_validator("can_id_path")
    @classmethod
    def _stable_can_id_path(cls, value: str) -> str:
        return _stable_nonempty_id_path("can_id_path", value, require_usb=True)

    @field_validator("motor_offsets", "kp_gains", "kd_gains")
    @classmethod
    def _seven_finite_gains(cls, value: dict[str, Any], info: Any) -> dict[str, float]:
        return _finite_mapping(info.field_name, value, YAM_JOINTS)

    @field_validator("motor_directions")
    @classmethod
    def _plus_or_minus_one(cls, value: dict[str, Any]) -> dict[str, int]:
        _require_exact_keys("motor_directions", value, YAM_JOINTS)
        directions: dict[str, int] = {}
        for name in YAM_JOINTS:
            direction = int(value[name])
            if direction not in (-1, 1):
                raise ValueError(f"motor_directions[{name!r}] must be 1 or -1")
            directions[name] = direction
        return directions

    @field_validator("joint_limits")
    @classmethod
    def _six_ordered_limits(cls, value: dict[str, Any]) -> dict[str, tuple[float, float]]:
        _require_exact_keys("joint_limits", value, YAM_ARM_JOINTS)
        return {
            name: _ordered_limit_pair(f"joint_limits[{name!r}]", value[name])
            for name in YAM_ARM_JOINTS
        }

    @field_validator("gripper_limits")
    @classmethod
    def _finite_gripper_limits(cls, value: Any) -> tuple[float, float]:
        return _finite_pair("gripper_limits", value)

    @field_validator("gravity_comp_factor")
    @classmethod
    def _finite_gravity_comp_factor(cls, value: Any) -> float:
        return _finite("gravity_comp_factor", value)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

can_id_path instance-attribute 

can_id_path: str

bitrate class-attribute instance-attribute 

bitrate: int = Field(default=1000000, gt=0)

bustype class-attribute instance-attribute 

bustype: Literal['socketcan'] = 'socketcan'

motor_offsets instance-attribute 

motor_offsets: dict[str, float]

motor_directions instance-attribute 

motor_directions: dict[str, int]

kp_gains instance-attribute 

kp_gains: dict[str, float]

kd_gains instance-attribute 

kd_gains: dict[str, float]

joint_limits instance-attribute 

joint_limits: dict[str, tuple[float, float]]

gripper_limits instance-attribute 

gripper_limits: tuple[float, float]

use_gravity_compensation class-attribute instance-attribute 

use_gravity_compensation: bool = True

gravity_comp_factor class-attribute instance-attribute 

gravity_comp_factor: float = 1.3

gripper_type class-attribute instance-attribute 

gripper_type: Literal['crank_4310', 'linear_3507', 'linear_4310'] = 'crank_4310'

zero_gravity_mode class-attribute instance-attribute 

zero_gravity_mode: Literal[False] = False

shutdown_zero_gravity_wait_for_enter class-attribute instance-attribute 

shutdown_zero_gravity_wait_for_enter: Literal[False] = False

limit_gripper_force class-attribute instance-attribute 

limit_gripper_force: float = Field(default=50.0, gt=0)

lerobot_max_step class-attribute instance-attribute 

lerobot_max_step: float | None = Field(default=None, gt=0)

lerobot_gripper_max_step class-attribute instance-attribute 

lerobot_gripper_max_step: float | None = Field(default=None, gt=0)

RailConfig

Bases: BaseModel

Measured geometry, travel direction, and default job position for one rail.

default_position_percent is where the carriage should sit when a job gives no override (0 = left, 100 = right). Jobs override it with rail_position as a 0–1 fraction. left_dir is the sign of servo rotation that moves the carriage left; it is a property of how the belt and motor are fitted, not of any homing routine.

Source code in core/src/armnet_core/models.py 
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
class RailConfig(BaseModel):
    """Measured geometry, travel direction, and default job position for one rail.

    ``default_position_percent`` is where the carriage should sit when a job
    gives no override (0 = left, 100 = right). Jobs override it with
    ``rail_position`` as a 0–1 fraction. ``left_dir`` is the sign of servo
    rotation that moves the carriage left; it is a property of how the belt
    and motor are fitted, not of any homing routine.
    """

    model_config = ConfigDict(extra="forbid")

    servo_id: int = Field(default=7, ge=0, le=253)
    total_rail_steps: int = Field(gt=0)
    left_dir: Literal[-1, 1] = 1
    default_position_percent: float = Field(default=50, ge=0, le=100)
    variation_std_percent: float = Field(default=10.0, ge=0)
    """Spread of the structured-variation draw around ``default_position_percent``.

    Per rail rather than global because the usable spread depends on where the
    carriage parks: a cell that sits mid-rail can swing both ways, while one
    parked near an end has the draw clipped against the stop and wants a
    smaller figure. Zero pins the rail to its default even when a job asks for
    variation.
    """
    home_end: Literal["left", "right"] = "right"
    """Physical end the carriage homes to when it rebuilds its position.

    Jobs home to this end by default before the container starts. A job that
    explicitly opts into conditional homing may trust the recorded position
    only when its calibration and settled live encoder tick still match; any
    uncertainty re-homes automatically. The end is not computed ("nearest"
    would need an untrusted position); the right end is the calibration
    reference and default.
    """

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

servo_id class-attribute instance-attribute 

servo_id: int = Field(default=7, ge=0, le=253)

total_rail_steps class-attribute instance-attribute 

total_rail_steps: int = Field(gt=0)

left_dir class-attribute instance-attribute 

left_dir: Literal[-1, 1] = 1

default_position_percent class-attribute instance-attribute 

default_position_percent: float = Field(default=50, ge=0, le=100)

variation_std_percent class-attribute instance-attribute 

variation_std_percent: float = Field(default=10.0, ge=0)

Spread of the structured-variation draw around default_position_percent.

Per rail rather than global because the usable spread depends on where the carriage parks: a cell that sits mid-rail can swing both ways, while one parked near an end has the draw clipped against the stop and wants a smaller figure. Zero pins the rail to its default even when a job asks for variation.

home_end class-attribute instance-attribute 

home_end: Literal['left', 'right'] = 'right'

Physical end the carriage homes to when it rebuilds its position.

Jobs home to this end by default before the container starts. A job that explicitly opts into conditional homing may trust the recorded position only when its calibration and settled live encoder tick still match; any uncertainty re-homes automatically. The end is not computed ("nearest" would need an untrusted position); the right end is the calibration reference and default.

RobotArmConfig

Bases: BaseModel

Named arm configuration for multi-arm (bimanual) cells.

The runtime derives each arm's connector endpoint from the cell-level robot_connector_endpoint plus the arm name; port is the physical serial device or SocketCAN interface that arm is wired to on the edge Pi, used by the Fleet Agent to provision the (bimanual) edge connector with the per-arm port mapping (--arm-ports). Optional because the edge can also be provisioned out-of-band, but set it so the agent can bring a bimanual cell up itself.

Source code in core/src/armnet_core/models.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
class RobotArmConfig(BaseModel):
    """Named arm configuration for multi-arm (bimanual) cells.

    The runtime derives each arm's connector endpoint from the cell-level
    ``robot_connector_endpoint`` plus the arm name; ``port`` is the physical
    serial device or SocketCAN interface that arm is wired to on the edge Pi,
    used by the Fleet Agent to provision the (bimanual) edge connector with
    the per-arm port mapping (``--arm-ports``). Optional because the edge can
    also be provisioned out-of-band, but set it so the agent can bring a
    bimanual cell up itself.
    """

    model_config = ConfigDict(extra="forbid")

    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    rest_position: Optional[dict[str, float]] = None
    safety_limit: Optional[float] = None
    safety_delta_degrees: Optional[float] = None
    power_plug_ip: Optional[str] = None
    port: Optional[str] = None
    rail: Optional[RailConfig] = None
    yam: Optional[YamArmDriverConfig] = None

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

robot_id class-attribute instance-attribute 

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute 

calibration_dir: Optional[Path] = None

calibration_file_path class-attribute instance-attribute 

calibration_file_path: Optional[Path] = None

rest_position class-attribute instance-attribute 

rest_position: Optional[dict[str, float]] = None

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: Optional[float] = None

power_plug_ip class-attribute instance-attribute 

power_plug_ip: Optional[str] = None

port class-attribute instance-attribute 

port: Optional[str] = None

rail class-attribute instance-attribute 

rail: Optional[RailConfig] = None

yam class-attribute instance-attribute 

yam: Optional[YamArmDriverConfig] = None

UniformCurrentGateConfig

Bases: BaseModel

Two current limits applied to every joint: a fast one and a slow one.

One threshold cannot describe motor damage. Winding heat goes as the square of current against a time constant of tens of seconds, so a brief excursion near stall is harmless while a moderate load held for a minute is not. limit_counts is the peak tier and catches a crash or hard stall in duration_ms; sustained_limit_counts sits near the servo's continuous rating and catches the slow cook over sustained_duration_ms. Set the sustained limit to null to run the peak tier alone.

Source code in core/src/armnet_core/models.py 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
class UniformCurrentGateConfig(BaseModel):
    """Two current limits applied to every joint: a fast one and a slow one.

    One threshold cannot describe motor damage. Winding heat goes as the
    square of current against a time constant of tens of seconds, so a brief
    excursion near stall is harmless while a moderate load held for a minute
    is not. ``limit_counts`` is the peak tier and catches a crash or hard
    stall in ``duration_ms``; ``sustained_limit_counts`` sits near the servo's
    continuous rating and catches the slow cook over
    ``sustained_duration_ms``. Set the sustained limit to null to run the peak
    tier alone.
    """

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    mode: Literal["uniform"] = "uniform"
    limit_counts: int = Field(default=275, gt=0)
    duration_ms: float = Field(default=500.0, gt=0)
    sustained_limit_counts: Optional[int] = Field(default=120, gt=0)
    sustained_duration_ms: float = Field(default=5000.0, gt=0)

    @model_validator(mode="after")
    def _sustained_tier_is_the_lower_slower_one(self) -> "UniformCurrentGateConfig":
        # Either inversion parses fine and then quietly does nothing: a
        # sustained limit at or above the peak can never be the tier that
        # fires, and a window no longer than the peak's makes both tiers the
        # same test.
        if self.sustained_limit_counts is None:
            return self
        if self.sustained_limit_counts >= self.limit_counts:
            raise ValueError(
                "current_gate.sustained_limit_counts must be below limit_counts"
            )
        if self.sustained_duration_ms <= self.duration_ms:
            raise ValueError(
                "current_gate.sustained_duration_ms must exceed duration_ms"
            )
        return self

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

mode class-attribute instance-attribute 

mode: Literal['uniform'] = 'uniform'

limit_counts class-attribute instance-attribute 

limit_counts: int = Field(default=275, gt=0)

duration_ms class-attribute instance-attribute 

duration_ms: float = Field(default=500.0, gt=0)

sustained_limit_counts class-attribute instance-attribute 

sustained_limit_counts: Optional[int] = Field(default=120, gt=0)

sustained_duration_ms class-attribute instance-attribute 

sustained_duration_ms: float = Field(default=5000.0, gt=0)

PercentileCurrentGateConfig

Bases: BaseModel

Per-joint limits derived from the packaged normal-current profile.

Source code in core/src/armnet_core/models.py 
469
470
471
472
473
474
475
476
477
478
class PercentileCurrentGateConfig(BaseModel):
    """Per-joint limits derived from the packaged normal-current profile."""

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    mode: Literal["percentile"]
    percentile: Literal[95, 99] = 99
    percent_above: float = Field(default=100.0, gt=0)
    tiny_max_threshold: int = Field(default=15, gt=0)
    duration_ms: float = Field(default=200.0, gt=0)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

mode instance-attribute 

mode: Literal['percentile']

percentile class-attribute instance-attribute 

percentile: Literal[95, 99] = 99

percent_above class-attribute instance-attribute 

percent_above: float = Field(default=100.0, gt=0)

tiny_max_threshold class-attribute instance-attribute 

tiny_max_threshold: int = Field(default=15, gt=0)

duration_ms class-attribute instance-attribute 

duration_ms: float = Field(default=200.0, gt=0)

OffCurrentGateConfig

Bases: BaseModel

Explicit operator override that disables current gating.

Source code in core/src/armnet_core/models.py 
481
482
483
484
485
486
class OffCurrentGateConfig(BaseModel):
    """Explicit operator override that disables current gating."""

    model_config = ConfigDict(extra="forbid")

    mode: Literal["off"]

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

mode instance-attribute 

mode: Literal['off']

TrackingStallConfig

Bases: BaseModel

Persistent goal-versus-position obstacle detector owned by the cell.

Source code in core/src/armnet_core/models.py 
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
class TrackingStallConfig(BaseModel):
    """Persistent goal-versus-position obstacle detector owned by the cell."""

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    mode: Literal["off", "shadow", "enforce"] = "shadow"
    min_error_deg: float = Field(default=8.0, gt=0)
    release_error_deg: float = Field(default=4.0, gt=0)
    max_position_range_deg: float = Field(default=2.0, gt=0)
    duration_s: float = Field(default=2.0, gt=0)
    max_sample_gap_s: float = Field(default=0.5, gt=0)
    excluded_motors: tuple[str, ...] = ("gripper",)

    @model_validator(mode="after")
    def _release_has_hysteresis(self) -> "TrackingStallConfig":
        if self.release_error_deg >= self.min_error_deg:
            raise ValueError(
                "tracking_stall.release_error_deg must be below min_error_deg"
            )
        return self

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

mode class-attribute instance-attribute 

mode: Literal['off', 'shadow', 'enforce'] = 'shadow'

min_error_deg class-attribute instance-attribute 

min_error_deg: float = Field(default=8.0, gt=0)

release_error_deg class-attribute instance-attribute 

release_error_deg: float = Field(default=4.0, gt=0)

max_position_range_deg class-attribute instance-attribute 

max_position_range_deg: float = Field(default=2.0, gt=0)

duration_s class-attribute instance-attribute 

duration_s: float = Field(default=2.0, gt=0)

max_sample_gap_s class-attribute instance-attribute 

max_sample_gap_s: float = Field(default=0.5, gt=0)

excluded_motors class-attribute instance-attribute 

excluded_motors: tuple[str, ...] = ('gripper',)

JobSpec

Bases: BaseModel

The fields a client supplies when creating a job.

This is the body of POST /jobs. The orchestrator wraps it into a :class:Job (assigning id, status, timestamps) before persisting.

Source code in core/src/armnet_core/models.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
class JobSpec(BaseModel):
    """The fields a client supplies when creating a job.

    This is the body of ``POST /jobs``. The orchestrator wraps it into a
    :class:`Job` (assigning ``id``, ``status``, timestamps) before
    persisting.
    """

    image: str = Field(
        ...,
        description=(
            "Container image selector/reference. In production, the orchestrator "
            "uses only the validated image selector and reconstructs registry, "
            "repository, and owner namespace from authenticated platform state. "
            "Local development may use a local `my-image:tag` reference."
        ),
    )
    args: dict[str, Any] = Field(
        default_factory=dict,
        description="Keyword arguments passed to the customer's @main-decorated "
        "function as `ctx.args`. JSON-encoded into the `ARMNET_ARGS` env "
        "var by the cell; decoded by the `armnet-runtime` entrypoint "
        "before calling user code. Must be JSON-serialisable.",
    )
    embodiment: Embodiment = Field(
        ...,
        description="Required robot embodiment. The orchestrator routes the "
        "job onto the NATS subject for this embodiment+task pair, where the "
        "matching cell picks it up.",
    )
    task: Optional[Task] = Field(
        default=None,
        description="Optional task. When set, only cells configured for this "
        "(embodiment, task) pair run the job. When omitted, any cell of the "
        "embodiment may pick it up regardless of its configured task.",
    )
    timeout_seconds: int = Field(
        default=120,
        ge=1,
        description="Wall-clock cap on container execution.",
    )
    secrets: dict[str, str] = Field(
        default_factory=dict,
        description=(
            "Mapping of environment variable name to user secret name. "
            "Example: {'HF_TOKEN': 'huggingface-token'} resolves the "
            "authenticated user's secret and injects it as HF_TOKEN."
        ),
    )
    detach: bool = Field(
        default=False,
        description=(
            "If false, losing the client log WebSocket requests graceful job "
            "cancellation. If true, the job keeps running after client disconnect."
        ),
    )
    skip_operator_dispatch: bool = Field(
        default=False,
        description=(
            "Run immediately on a manual-environment cell instead of waiting "
            "for an operator to press Dispatch in the FMS. Set by interactive "
            "teleop and recording, where the submitter is at the cell. Policy "
            "evals leave this false so the scene is staged first."
        ),
    )
    username: Optional[str] = Field(
        default=None,
        description=(
            "Authenticated armnet username. Set by the orchestrator from "
            "the API key; clients should not rely on supplied values being preserved."
        ),
    )

image class-attribute instance-attribute 

image: str = Field(..., description='Container image selector/reference. In production, the orchestrator uses only the validated image selector and reconstructs registry, repository, and owner namespace from authenticated platform state. Local development may use a local `my-image:tag` reference.')

args class-attribute instance-attribute 

args: dict[str, Any] = Field(default_factory=dict, description="Keyword arguments passed to the customer's @main-decorated function as `ctx.args`. JSON-encoded into the `ARMNET_ARGS` env var by the cell; decoded by the `armnet-runtime` entrypoint before calling user code. Must be JSON-serialisable.")

embodiment class-attribute instance-attribute 

embodiment: Embodiment = Field(..., description='Required robot embodiment. The orchestrator routes the job onto the NATS subject for this embodiment+task pair, where the matching cell picks it up.')

task class-attribute instance-attribute 

task: Optional[Task] = Field(default=None, description='Optional task. When set, only cells configured for this (embodiment, task) pair run the job. When omitted, any cell of the embodiment may pick it up regardless of its configured task.')

timeout_seconds class-attribute instance-attribute 

timeout_seconds: int = Field(default=120, ge=1, description='Wall-clock cap on container execution.')

secrets class-attribute instance-attribute 

secrets: dict[str, str] = Field(default_factory=dict, description="Mapping of environment variable name to user secret name. Example: {'HF_TOKEN': 'huggingface-token'} resolves the authenticated user's secret and injects it as HF_TOKEN.")

detach class-attribute instance-attribute 

detach: bool = Field(default=False, description='If false, losing the client log WebSocket requests graceful job cancellation. If true, the job keeps running after client disconnect.')

skip_operator_dispatch class-attribute instance-attribute 

skip_operator_dispatch: bool = Field(default=False, description='Run immediately on a manual-environment cell instead of waiting for an operator to press Dispatch in the FMS. Set by interactive teleop and recording, where the submitter is at the cell. Policy evals leave this false so the scene is staged first.')

username class-attribute instance-attribute 

username: Optional[str] = Field(default=None, description='Authenticated armnet username. Set by the orchestrator from the API key; clients should not rely on supplied values being preserved.')

RobotCellConfig

Bases: BaseModel

Physical robot cell configuration loaded from robot_cell.json.

This is operator-owned configuration, not job input. The cell and local Docker runner use it to inject the same robot metadata into the runtime context for every job that lands on this cell.

The JSON file may use a nested structure with top-level groups: cell, robot, data_interface, policy. A model validator flattens nested input into the canonical flat field set for backward compatibility with all existing consumers.

Source code in core/src/armnet_core/models.py 
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
class RobotCellConfig(BaseModel):
    """Physical robot cell configuration loaded from ``robot_cell.json``.

    This is operator-owned configuration, not job input. The cell and local
    Docker runner use it to inject the same robot metadata into the runtime
    context for every job that lands on this cell.

    The JSON file may use a nested structure with top-level groups:
    ``cell``, ``robot``, ``data_interface``, ``policy``. A model validator
    flattens nested input into the canonical flat field set for backward
    compatibility with all existing consumers.
    """

    nats_url: Optional[str] = None
    cell_id: Optional[str] = None
    embodiment: Embodiment = DEFAULT_EMBODIMENT
    # The task being run right now, stamped in per job. None outside a job: the
    # cell is set up for an environment, not a task, and hosts all of them.
    task: Optional[Task] = None
    connector_socket_path: Optional[str] = None
    connector_tcp: Optional[str] = None
    robot_connector_endpoint: Optional[str] = None
    operator_call_endpoint: Optional[str] = "tcp://127.0.0.1:9877"
    robot_port: Optional[str] = None
    robot_power_plug_ip: Optional[str] = None
    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    camera_configs: dict[str, dict[str, Any]] = Field(default_factory=dict)
    # Immutable DB snapshot selected when the cell starts. Keeping it in the
    # runtime config prevents a later activation from changing an active job.
    scene_calibration: Optional[SceneCalibration] = None
    arms: dict[str, RobotArmConfig] = Field(default_factory=dict)
    rail: Optional[RailConfig] = None
    safety_limit: Optional[float] = Field(
        default=None,
        description=(
            "Runtime-facing relative action safety limit for robot SDK configs. "
            "SO-101 cells expose 30 degrees; embodiments with connector-only "
            "safety, such as ARX5, leave this unset."
        ),
    )
    docker_gpus: Optional[str] = None
    language_instruction: Optional[str] = None
    completion_enabled: bool = False
    completion_threshold: float = 0.9
    completion_camera: Optional[str] = None
    completion_min_interval_s: float = 1.0
    completion_model_path: str = "robometer/Robometer-4B"
    completion_chunk_seconds: float = 2.0
    completion_sample_fps: float = 4.0
    arm_rest_position: Optional[dict[str, float]] = None
    # Robot motion/safety params (pushed to edge connector at job start)
    safety_delta_degrees: Optional[float] = 30.0
    goal_velocity_limit: Optional[int] = Field(default=None, ge=1, le=3250)
    acceleration_limit: Optional[int] = Field(default=None, ge=1, le=254)
    max_command_speed_deg_s: Optional[float] = Field(default=300.0, gt=0)
    max_command_oscillation_deg_s2: Optional[float] = Field(default=4800.0, gt=0)
    rest_return_seconds: float = 4.0
    rest_return_hz: float = 20.0
    gripper_release_seconds: float = 1.0
    gripper_motor_name: str = "gripper"
    gripper_release_offset: float = 10.0
    current_gate: CurrentGateConfig = Field(default_factory=UniformCurrentGateConfig)
    tracking_stall: TrackingStallConfig = Field(
        default_factory=TrackingStallConfig
    )
    # Data interface params
    tcp_read_timeout: float = 10.0
    jpeg_quality: int = 0
    # Cell infra
    shelly_timeout: float = 2.0
    # What this cell is set up to work on, and that environment's own settings.
    # The settings stay an opaque dict: only the package implementing the
    # environment understands them, and armnet-core must remain pydantic-only.
    # from_file() may load them from environment_config_file; they are always
    # serialized inline into ARMNET_CELL_CONFIG so containers need no bind mount.
    environment: Optional[str] = None
    environment_config: dict[str, Any] = Field(default_factory=dict)
    # Workcell lightbox: HTTP-controlled LED dimmer on the cell's own AP (e.g.
    # "http://lightbox.local"). Only the edge host can reach it, so the cell
    # routes brightness through the edge connector at job start. None = no
    # lightbox on this cell. Brightness/freq/fade are cell defaults (JSON);
    # jobs may override brightness via args.lightbox_brightness and frequency
    # via args.lightbox_frequency. size_cm / light_sets identify the physical
    # box (not sent to the ESP) so PWM defaults are lightbox-specific, not
    # cell-specific.
    lightbox_url: Optional[str] = None
    lightbox_size_cm: Optional[int] = None  # enclosure edge length; identity only
    lightbox_light_sets: Optional[int] = None  # LED strips in the box; identity only
    lightbox_default_brightness: int = 30  # percent 0-100
    lightbox_default_frequency: int = 2000  # PWM Hz
    lightbox_default_fade: int = 1000  # ms ramp on /set
    # Spread of the structured-variation draw around the default brightness,
    # in the same percent units. Zero holds the light at its default even when
    # a job asks for variation.
    #
    # 11 rather than a round 10 because a dry run found the light barely
    # changing: it widens the envelope to [2.5, 57.5], which is about as far as
    # a symmetric draw around a default of 30 can reach before the low end
    # bottoms out at zero.
    lightbox_variation_std: float = Field(default=11.0, ge=0)

    @classmethod
    def _flatten_nested(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Flatten nested JSON groups into the canonical flat field set."""
        flat: dict[str, Any] = {}

        cell = data.pop("cell", None)
        if isinstance(cell, dict):
            flat["nats_url"] = cell.get("nats_url")
            flat["cell_id"] = cell.get("cell_id")
            flat["connector_socket_path"] = cell.get("connector_socket_path")
            flat["connector_tcp"] = cell.get("connector_tcp")
            flat["robot_connector_endpoint"] = cell.get("robot_connector_endpoint")
            flat["operator_call_endpoint"] = cell.get("operator_call_endpoint")
            flat["robot_port"] = cell.get("robot_port")
            flat["docker_gpus"] = cell.get("docker_gpus")
            flat["robot_power_plug_ip"] = cell.get("power_plug_ip")
            flat["shelly_timeout"] = cell.get("shelly_timeout", 2.0)

        robot = data.pop("robot", None)
        if isinstance(robot, dict):
            if "current_gate_enabled" in robot:
                raise ValueError(
                    "robot.current_gate_enabled is no longer supported; use "
                    "robot.current_gate.mode (uniform, percentile, or off)"
                )
            flat["embodiment"] = robot.get("embodiment")
            flat["robot_id"] = robot.get("robot_id")
            flat["calibration_dir"] = robot.get("calibration_dir")
            flat["calibration_file_path"] = robot.get("calibration_file_path")
            flat["arm_rest_position"] = robot.get("rest_position")
            flat["camera_configs"] = robot.get("camera_configs", {})
            flat["arms"] = robot.get("arms", {})
            flat["rail"] = robot.get("rail")
            if robot.get("embodiment") == BIMANUAL_YAM_EMBODIMENT:
                if "safety_delta_degrees" in robot:
                    flat["safety_delta_degrees"] = robot["safety_delta_degrees"]
            else:
                flat["safety_delta_degrees"] = robot.get("safety_delta_degrees", 30.0)
            flat["goal_velocity_limit"] = robot.get("goal_velocity_limit")
            flat["acceleration_limit"] = robot.get("acceleration_limit")
            flat["max_command_speed_deg_s"] = robot.get(
                "max_command_speed_deg_s", 300.0
            )
            flat["max_command_oscillation_deg_s2"] = robot.get(
                "max_command_oscillation_deg_s2", 4800.0
            )
            flat["rest_return_seconds"] = robot.get("rest_return_seconds", 4.0)
            flat["rest_return_hz"] = robot.get("rest_return_hz", 20.0)
            flat["gripper_release_seconds"] = robot.get("gripper_release_seconds", 1.0)
            flat["gripper_motor_name"] = robot.get("gripper_motor_name", "gripper")
            flat["gripper_release_offset"] = robot.get("gripper_release_offset", 10.0)
            if "current_gate" in robot:
                flat["current_gate"] = robot["current_gate"]
            if "tracking_stall" in robot:
                flat["tracking_stall"] = robot["tracking_stall"]

        data_interface = data.pop("data_interface", None)
        if isinstance(data_interface, dict):
            flat["tcp_read_timeout"] = data_interface.get("tcp_read_timeout", 10.0)
            flat["jpeg_quality"] = data_interface.get("jpeg_quality", 0)

        # The environment's settings live in a block named after it and are
        # passed through whole. Listing the keys we expect, as every other block
        # here does, is exactly how a new setting gets silently dropped on its
        # way to the cell.
        environment = data.pop("environment", None)
        if environment:
            flat["environment"] = environment
            flat["environment_config"] = data.pop(environment, None) or {}

        lightbox = data.pop("lightbox", None)
        if isinstance(lightbox, dict):
            flat["lightbox_url"] = lightbox.get("url")
            if lightbox.get("size_cm") is not None:
                flat["lightbox_size_cm"] = lightbox["size_cm"]
            if lightbox.get("light_sets") is not None:
                flat["lightbox_light_sets"] = lightbox["light_sets"]
            if lightbox.get("default_brightness") is not None:
                flat["lightbox_default_brightness"] = lightbox["default_brightness"]
            if lightbox.get("default_frequency") is not None:
                flat["lightbox_default_frequency"] = lightbox["default_frequency"]
            if lightbox.get("default_fade") is not None:
                flat["lightbox_default_fade"] = lightbox["default_fade"]
            if lightbox.get("variation_std") is not None:
                flat["lightbox_variation_std"] = lightbox["variation_std"]

        policy = data.pop("policy", None)
        if isinstance(policy, dict):
            # NOTE: ``task`` / ``language_instruction`` are intentionally NOT read
            # from the config file. They belong to the job, not the cell: a cell
            # hosts every task its environment offers and is told which one this
            # is when the job arrives. Any ``policy.task`` /
            # ``policy.language_instruction`` left in a JSON file is ignored so a
            # stale copy can't override it.
            completion = policy.get("completion", {})
            if isinstance(completion, dict):
                flat["completion_enabled"] = completion.get("enabled", False)
                flat["completion_threshold"] = completion.get("threshold", 0.9)
                flat["completion_camera"] = completion.get("camera")
                flat["completion_min_interval_s"] = completion.get("min_interval_s", 1.0)
                flat["completion_model_path"] = completion.get(
                    "model_path", "robometer/Robometer-4B"
                )
                flat["completion_chunk_seconds"] = completion.get("chunk_seconds", 2.0)
                flat["completion_sample_fps"] = completion.get("sample_fps", 4.0)

        keep_none = set()
        if (
            isinstance(robot, dict)
            and robot.get("embodiment") == BIMANUAL_YAM_EMBODIMENT
        ):
            keep_none.add("safety_delta_degrees")
        if isinstance(robot, dict):
            keep_none.update(
                key
                for key in (
                    "max_command_speed_deg_s",
                    "max_command_oscillation_deg_s2",
                )
                if key in robot
            )
        # Remove None values so Pydantic defaults apply, except fields whose
        # explicit null has operator-owned disable semantics.
        flat = {k: v for k, v in flat.items() if v is not None or k in keep_none}
        # Merge: explicit flat fields in data take precedence over unpacked nested
        flat.update(data)
        return flat

    @model_validator(mode="before")
    @classmethod
    def _accept_nested_format(cls, data: Any) -> Any:
        if not isinstance(data, dict):
            return data
        robot = data.get("robot")
        embodiment = data.get("embodiment")
        if isinstance(robot, dict):
            embodiment = robot.get("embodiment", embodiment)
        if embodiment == BIMANUAL_YAM_EMBODIMENT:
            if "lightbox" in data:
                raise ValueError("bimanual-yam rejects lightbox")
            if isinstance(robot, dict):
                for key in _BIMANUAL_YAM_ROBOT_FORBIDDEN:
                    if key in robot:
                        raise ValueError(f"bimanual-yam rejects {key}")
        if "current_gate_enabled" in data or (
            isinstance(robot, dict) and "current_gate_enabled" in robot
        ):
            raise ValueError(
                "current_gate_enabled is no longer supported; use "
                "robot.current_gate.mode (uniform, percentile, or off)"
            )
        # ``environment`` is both a nested-file group and a flat RobotCellConfig
        # field. Runner dumps include it without a ``robot`` block; those are
        # already-flat runtime payloads, not operator JSON.
        had_nested_robot = isinstance(robot, dict)
        if any(key in data for key in ("cell", "robot", "policy", "data_interface", "environment", "lightbox")):
            data = cls._flatten_nested(dict(data))
            embodiment = data.get("embodiment", embodiment)
        if (
            embodiment == BIMANUAL_YAM_EMBODIMENT
            and not had_nested_robot
            and "safety_delta_degrees" not in data
        ):
            if not isinstance(data, dict):
                return data
            data = dict(data)
            data["safety_delta_degrees"] = None
        return data

    @model_validator(mode="after")
    def _validate_named_arms(self) -> "RobotCellConfig":
        if self.embodiment == BIMANUAL_YAM_EMBODIMENT:
            _require_yam_serial_safety_explicitly_off(self)
            return _validate_bimanual_yam_cell(self)
        if self.safety_delta_degrees is None:
            self = self.model_copy(update={"safety_delta_degrees": 30.0})
        if self.arms and not {"left", "right"}.issubset(self.arms):
            raise ValueError("bimanual robot.arms must include both 'left' and 'right'")
        return self

    @classmethod
    def from_file(cls, path: str | Path) -> "RobotCellConfig":
        config_path = Path(path).expanduser().resolve()
        raw_config = json.loads(config_path.read_text())
        environment_config_file = raw_config.pop("environment_config_file", None)
        if environment_config_file:
            environment = raw_config.get("environment")
            if "environment_config" in raw_config or (
                isinstance(environment, str) and environment in raw_config
            ):
                raise ValueError(
                    "robot cell config cannot define both environment_config "
                    "and environment_config_file"
                )
            environment_path = Path(environment_config_file).expanduser()
            if not environment_path.is_absolute():
                environment_path = config_path.parent / environment_path
            raw_config["environment_config"] = json.loads(environment_path.read_text())
        config = cls.model_validate(raw_config)
        update: dict[str, Any] = {}
        if config.calibration_dir and not config.calibration_dir.is_absolute():
            update["calibration_dir"] = config_path.parent / config.calibration_dir
        if config.calibration_file_path and not config.calibration_file_path.is_absolute():
            update["calibration_file_path"] = config_path.parent / config.calibration_file_path
        if config.arms:
            arms: dict[str, RobotArmConfig] = {}
            changed = False
            for name, arm in config.arms.items():
                arm_update: dict[str, Path] = {}
                if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                    arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
                if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                    arm_update["calibration_file_path"] = (
                        config_path.parent / arm.calibration_file_path
                    )
                arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
                changed = changed or bool(arm_update)
            if changed:
                update["arms"] = arms
        return config.model_copy(update=update) if update else config

nats_url class-attribute instance-attribute 

nats_url: Optional[str] = None

cell_id class-attribute instance-attribute 

cell_id: Optional[str] = None

embodiment class-attribute instance-attribute 

embodiment: Embodiment = DEFAULT_EMBODIMENT

task class-attribute instance-attribute 

task: Optional[Task] = None

connector_socket_path class-attribute instance-attribute 

connector_socket_path: Optional[str] = None

connector_tcp class-attribute instance-attribute 

connector_tcp: Optional[str] = None

robot_connector_endpoint class-attribute instance-attribute 

robot_connector_endpoint: Optional[str] = None

operator_call_endpoint class-attribute instance-attribute 

operator_call_endpoint: Optional[str] = 'tcp://127.0.0.1:9877'

robot_port class-attribute instance-attribute 

robot_port: Optional[str] = None

robot_power_plug_ip class-attribute instance-attribute 

robot_power_plug_ip: Optional[str] = None

robot_id class-attribute instance-attribute 

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute 

calibration_dir: Optional[Path] = None

calibration_file_path class-attribute instance-attribute 

calibration_file_path: Optional[Path] = None

camera_configs class-attribute instance-attribute 

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

scene_calibration class-attribute instance-attribute 

scene_calibration: Optional[SceneCalibration] = None

arms class-attribute instance-attribute 

arms: dict[str, RobotArmConfig] = Field(default_factory=dict)

rail class-attribute instance-attribute 

rail: Optional[RailConfig] = None

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = Field(default=None, description='Runtime-facing relative action safety limit for robot SDK configs. SO-101 cells expose 30 degrees; embodiments with connector-only safety, such as ARX5, leave this unset.')

docker_gpus class-attribute instance-attribute 

docker_gpus: Optional[str] = None

language_instruction class-attribute instance-attribute 

language_instruction: Optional[str] = None

completion_enabled class-attribute instance-attribute 

completion_enabled: bool = False

completion_threshold class-attribute instance-attribute 

completion_threshold: float = 0.9

completion_camera class-attribute instance-attribute 

completion_camera: Optional[str] = None

completion_min_interval_s class-attribute instance-attribute 

completion_min_interval_s: float = 1.0

completion_model_path class-attribute instance-attribute 

completion_model_path: str = 'robometer/Robometer-4B'

completion_chunk_seconds class-attribute instance-attribute 

completion_chunk_seconds: float = 2.0

completion_sample_fps class-attribute instance-attribute 

completion_sample_fps: float = 4.0

arm_rest_position class-attribute instance-attribute 

arm_rest_position: Optional[dict[str, float]] = None

safety_delta_degrees class-attribute instance-attribute 

safety_delta_degrees: Optional[float] = 30.0

goal_velocity_limit class-attribute instance-attribute 

goal_velocity_limit: Optional[int] = Field(default=None, ge=1, le=3250)

acceleration_limit class-attribute instance-attribute 

acceleration_limit: Optional[int] = Field(default=None, ge=1, le=254)

max_command_speed_deg_s class-attribute instance-attribute 

max_command_speed_deg_s: Optional[float] = Field(default=300.0, gt=0)

max_command_oscillation_deg_s2 class-attribute instance-attribute 

max_command_oscillation_deg_s2: Optional[float] = Field(default=4800.0, gt=0)

rest_return_seconds class-attribute instance-attribute 

rest_return_seconds: float = 4.0

rest_return_hz class-attribute instance-attribute 

rest_return_hz: float = 20.0

gripper_release_seconds class-attribute instance-attribute 

gripper_release_seconds: float = 1.0

gripper_motor_name class-attribute instance-attribute 

gripper_motor_name: str = 'gripper'

gripper_release_offset class-attribute instance-attribute 

gripper_release_offset: float = 10.0

current_gate class-attribute instance-attribute 

current_gate: CurrentGateConfig = Field(default_factory=UniformCurrentGateConfig)

tracking_stall class-attribute instance-attribute 

tracking_stall: TrackingStallConfig = Field(default_factory=TrackingStallConfig)

tcp_read_timeout class-attribute instance-attribute 

tcp_read_timeout: float = 10.0

jpeg_quality class-attribute instance-attribute 

jpeg_quality: int = 0

shelly_timeout class-attribute instance-attribute 

shelly_timeout: float = 2.0

environment class-attribute instance-attribute 

environment: Optional[str] = None

environment_config class-attribute instance-attribute 

environment_config: dict[str, Any] = Field(default_factory=dict)

lightbox_url class-attribute instance-attribute 

lightbox_url: Optional[str] = None

lightbox_size_cm class-attribute instance-attribute 

lightbox_size_cm: Optional[int] = None

lightbox_light_sets class-attribute instance-attribute 

lightbox_light_sets: Optional[int] = None

lightbox_default_brightness class-attribute instance-attribute 

lightbox_default_brightness: int = 30

lightbox_default_frequency class-attribute instance-attribute 

lightbox_default_frequency: int = 2000

lightbox_default_fade class-attribute instance-attribute 

lightbox_default_fade: int = 1000

lightbox_variation_std class-attribute instance-attribute 

lightbox_variation_std: float = Field(default=11.0, ge=0)

from_file classmethod 

from_file(path: str | Path) -> 'RobotCellConfig'
Source code in core/src/armnet_core/models.py 
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
@classmethod
def from_file(cls, path: str | Path) -> "RobotCellConfig":
    config_path = Path(path).expanduser().resolve()
    raw_config = json.loads(config_path.read_text())
    environment_config_file = raw_config.pop("environment_config_file", None)
    if environment_config_file:
        environment = raw_config.get("environment")
        if "environment_config" in raw_config or (
            isinstance(environment, str) and environment in raw_config
        ):
            raise ValueError(
                "robot cell config cannot define both environment_config "
                "and environment_config_file"
            )
        environment_path = Path(environment_config_file).expanduser()
        if not environment_path.is_absolute():
            environment_path = config_path.parent / environment_path
        raw_config["environment_config"] = json.loads(environment_path.read_text())
    config = cls.model_validate(raw_config)
    update: dict[str, Any] = {}
    if config.calibration_dir and not config.calibration_dir.is_absolute():
        update["calibration_dir"] = config_path.parent / config.calibration_dir
    if config.calibration_file_path and not config.calibration_file_path.is_absolute():
        update["calibration_file_path"] = config_path.parent / config.calibration_file_path
    if config.arms:
        arms: dict[str, RobotArmConfig] = {}
        changed = False
        for name, arm in config.arms.items():
            arm_update: dict[str, Path] = {}
            if arm.calibration_dir and not arm.calibration_dir.is_absolute():
                arm_update["calibration_dir"] = config_path.parent / arm.calibration_dir
            if arm.calibration_file_path and not arm.calibration_file_path.is_absolute():
                arm_update["calibration_file_path"] = (
                    config_path.parent / arm.calibration_file_path
                )
            arms[name] = arm.model_copy(update=arm_update) if arm_update else arm
            changed = changed or bool(arm_update)
        if changed:
            update["arms"] = arms
    return config.model_copy(update=update) if update else config

VariationAxis

Bases: BaseModel

One continuously varied device setting, in the unit its API accepts.

default is the cell's calibrated value and std the spread of the draw around it. std_below and std_above replace that spread on one side of the default, so an axis can reach further one way than the other. minimum and maximum are the device's hard travel limits. What matters to callers is envelope: the clipped normal intersected with those limits, which is both the range the sampler can produce and the range an operator may ask for by hand.

Source code in core/src/armnet_core/models.py 
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
class VariationAxis(BaseModel):
    """One continuously varied device setting, in the unit its API accepts.

    ``default`` is the cell's calibrated value and ``std`` the spread of the
    draw around it. ``std_below`` and ``std_above`` replace that spread on one
    side of the default, so an axis can reach further one way than the other.
    ``minimum`` and ``maximum`` are the device's hard travel limits. What
    matters to callers is ``envelope``: the clipped normal intersected with
    those limits, which is both the range the sampler can produce and the
    range an operator may ask for by hand.
    """

    model_config = ConfigDict(extra="forbid", allow_inf_nan=False)

    default: float
    std: float = Field(default=0.0, ge=0.0)
    # Unset on a symmetric axis. They must not be serialized as null: clients
    # that predate these fields reject unknown keys, which is what took the
    # demo Space's availability lights offline.
    std_below: Optional[float] = Field(default=None, ge=0.0)
    std_above: Optional[float] = Field(default=None, ge=0.0)
    minimum: float
    maximum: float

    @model_serializer(mode="wrap")
    def _omit_unset_side_spreads(self, handler):  # noqa: ANN001
        dumped = handler(self)
        if isinstance(dumped, dict):
            if dumped.get("std_below") is None:
                dumped.pop("std_below", None)
            if dumped.get("std_above") is None:
                dumped.pop("std_above", None)
        return dumped

    @model_validator(mode="after")
    def _default_sits_inside_the_device_range(self) -> "VariationAxis":
        if self.minimum > self.maximum:
            raise ValueError(
                f"variation minimum {self.minimum} exceeds maximum {self.maximum}"
            )
        if not self.minimum <= self.default <= self.maximum:
            raise ValueError(
                f"variation default {self.default} is outside the device range "
                f"[{self.minimum}, {self.maximum}]"
            )
        return self

    @property
    def spread_below(self) -> float:
        """Spread applied when a draw lands below the default."""
        return self.std if self.std_below is None else self.std_below

    @property
    def spread_above(self) -> float:
        """Spread applied when a draw lands above the default."""
        return self.std if self.std_above is None else self.std_above

    def spread_for(self, z: float) -> float:
        """Which spread a signed normal draw uses. Zero stays on the default."""
        if z < 0:
            return self.spread_below
        if z > 0:
            return self.spread_above
        return 0.0

    @property
    def envelope(self) -> tuple[float, float]:
        """Lowest and highest value this axis may take."""
        return (
            max(self.minimum, self.default - VARIATION_CLIP_SIGMA * self.spread_below),
            min(self.maximum, self.default + VARIATION_CLIP_SIGMA * self.spread_above),
        )

    @property
    def is_clipped_by_the_device(self) -> bool:
        """Whether travel limits, not the sigma bound, decide the envelope.

        True means part of the intended distribution falls off the end of the
        device and piles up on a stop, so this cell varies less, and less
        symmetrically, than the configured spread suggests.
        """
        return (
            self.default - VARIATION_CLIP_SIGMA * self.spread_below < self.minimum
            or self.default + VARIATION_CLIP_SIGMA * self.spread_above > self.maximum
        )

    def permits(self, value: float) -> bool:
        """Whether ``value`` is one this axis could itself have produced."""
        low, high = self.envelope
        return low - _VARIATION_EPSILON <= value <= high + _VARIATION_EPSILON

    def describe_envelope(self, *, unit: str = "") -> str:
        low, high = self.envelope
        suffix = f" {unit}" if unit else ""
        return f"{low:g}{suffix} to {high:g}{suffix}"

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', allow_inf_nan=False)

default instance-attribute 

default: float

std class-attribute instance-attribute 

std: float = Field(default=0.0, ge=0.0)

std_below class-attribute instance-attribute 

std_below: Optional[float] = Field(default=None, ge=0.0)

std_above class-attribute instance-attribute 

std_above: Optional[float] = Field(default=None, ge=0.0)

minimum instance-attribute 

minimum: float

maximum instance-attribute 

maximum: float

spread_below property 

spread_below: float

Spread applied when a draw lands below the default.

spread_above property 

spread_above: float

Spread applied when a draw lands above the default.

envelope property 

envelope: tuple[float, float]

Lowest and highest value this axis may take.

is_clipped_by_the_device property 

is_clipped_by_the_device: bool

Whether travel limits, not the sigma bound, decide the envelope.

True means part of the intended distribution falls off the end of the device and piles up on a stop, so this cell varies less, and less symmetrically, than the configured spread suggests.

spread_for 

spread_for(z: float) -> float

Which spread a signed normal draw uses. Zero stays on the default.

Source code in core/src/armnet_core/models.py 
1022
1023
1024
1025
1026
1027
1028
def spread_for(self, z: float) -> float:
    """Which spread a signed normal draw uses. Zero stays on the default."""
    if z < 0:
        return self.spread_below
    if z > 0:
        return self.spread_above
    return 0.0

permits 

permits(value: float) -> bool

Whether value is one this axis could itself have produced.

Source code in core/src/armnet_core/models.py 
1051
1052
1053
1054
def permits(self, value: float) -> bool:
    """Whether ``value`` is one this axis could itself have produced."""
    low, high = self.envelope
    return low - _VARIATION_EPSILON <= value <= high + _VARIATION_EPSILON

describe_envelope 

describe_envelope(*, unit: str = '') -> str
Source code in core/src/armnet_core/models.py 
1056
1057
1058
1059
def describe_envelope(self, *, unit: str = "") -> str:
    low, high = self.envelope
    suffix = f" {unit}" if unit else ""
    return f"{low:g}{suffix} to {high:g}{suffix}"

CameraMountVariation

Bases: BaseModel

Both axes of one pan/tilt camera mount.

The unit is the mount's own: degrees for a hobby arducam_mount, encoder steps for a serial servo_mount. Nothing compares the two, and each axis carries the limits it was derived from, so neither the sampler nor the edge has to know which kind it is holding.

Source code in core/src/armnet_core/models.py 
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
class CameraMountVariation(BaseModel):
    """Both axes of one pan/tilt camera mount.

    The unit is the mount's own: degrees for a hobby ``arducam_mount``, encoder
    steps for a serial ``servo_mount``. Nothing compares the two, and each
    axis carries the limits it was derived from, so neither the sampler nor the
    edge has to know which kind it is holding.
    """

    model_config = ConfigDict(extra="forbid")

    pan: VariationAxis
    tilt: VariationAxis

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

pan instance-attribute 

pan: VariationAxis

tilt instance-attribute 

tilt: VariationAxis

RailVariation

Bases: BaseModel

One rail's position axis, in the 0-1 fraction the rail API takes.

Config declares the default and spread in percent, because that reads naturally beside default_position_percent; the conversion to the fraction every caller actually passes happens once here rather than at each comparison.

Source code in core/src/armnet_core/models.py 
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
class RailVariation(BaseModel):
    """One rail's position axis, in the 0-1 fraction the rail API takes.

    Config declares the default and spread in percent, because that reads
    naturally beside ``default_position_percent``; the conversion to the
    fraction every caller actually passes happens once here rather than at
    each comparison.
    """

    model_config = ConfigDict(extra="forbid")

    arm: Optional[str] = None
    position: VariationAxis

    @property
    def label(self) -> str:
        return f"rail {self.arm}" if self.arm else "rail"

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

arm class-attribute instance-attribute 

arm: Optional[str] = None

position instance-attribute 

position: VariationAxis

label property 

label: str

CellDeviceConfig

Bases: BaseModel

What a cell can vary, and by how much.

Derived from the cell's own config in one place so the client validating an operator's override before submission and the runtime sampling a scenario read the same envelope, instead of each deriving one and drifting apart.

Source code in core/src/armnet_core/models.py 
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
class CellDeviceConfig(BaseModel):
    """What a cell can vary, and by how much.

    Derived from the cell's own config in one place so the client validating an
    operator's override before submission and the runtime sampling a scenario
    read the same envelope, instead of each deriving one and drifting apart.
    """

    model_config = ConfigDict(extra="forbid")

    rails: list[RailVariation] = Field(default_factory=list)
    camera_mounts: dict[str, CameraMountVariation] = Field(default_factory=dict)
    lightbox_brightness: Optional[VariationAxis] = None

    @classmethod
    def from_cell_config(cls, config: "RobotCellConfig") -> "CellDeviceConfig":
        rails: list[RailVariation] = []
        if config.rail is not None:
            rails.append(_rail_variation(config.rail, arm=None))
        for arm_name, arm in (config.arms or {}).items():
            if arm.rail is not None:
                rails.append(_rail_variation(arm.rail, arm=arm_name))

        mounts: dict[str, CameraMountVariation] = {}
        for name, camera in (config.camera_configs or {}).items():
            if not isinstance(camera, dict):
                continue
            if isinstance(camera.get("arducam_mount"), dict):
                mounts[name] = _camera_mount_variation(camera["arducam_mount"])
            elif isinstance(camera.get("servo_mount"), dict):
                mounts[name] = _servo_mount_variation(camera["servo_mount"])

        # A cell with no lightbox URL has no light to vary, however the
        # brightness defaults are set.
        brightness = (
            VariationAxis(
                default=float(config.lightbox_default_brightness),
                std=float(config.lightbox_variation_std),
                minimum=0.0,
                maximum=100.0,
            )
            if config.lightbox_url
            else None
        )
        return cls(rails=rails, camera_mounts=mounts, lightbox_brightness=brightness)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid')

rails class-attribute instance-attribute 

rails: list[RailVariation] = Field(default_factory=list)

camera_mounts class-attribute instance-attribute 

camera_mounts: dict[str, CameraMountVariation] = Field(default_factory=dict)

lightbox_brightness class-attribute instance-attribute 

lightbox_brightness: Optional[VariationAxis] = None

from_cell_config classmethod 

from_cell_config(config: 'RobotCellConfig') -> 'CellDeviceConfig'
Source code in core/src/armnet_core/models.py 
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
@classmethod
def from_cell_config(cls, config: "RobotCellConfig") -> "CellDeviceConfig":
    rails: list[RailVariation] = []
    if config.rail is not None:
        rails.append(_rail_variation(config.rail, arm=None))
    for arm_name, arm in (config.arms or {}).items():
        if arm.rail is not None:
            rails.append(_rail_variation(arm.rail, arm=arm_name))

    mounts: dict[str, CameraMountVariation] = {}
    for name, camera in (config.camera_configs or {}).items():
        if not isinstance(camera, dict):
            continue
        if isinstance(camera.get("arducam_mount"), dict):
            mounts[name] = _camera_mount_variation(camera["arducam_mount"])
        elif isinstance(camera.get("servo_mount"), dict):
            mounts[name] = _servo_mount_variation(camera["servo_mount"])

    # A cell with no lightbox URL has no light to vary, however the
    # brightness defaults are set.
    brightness = (
        VariationAxis(
            default=float(config.lightbox_default_brightness),
            std=float(config.lightbox_variation_std),
            minimum=0.0,
            maximum=100.0,
        )
        if config.lightbox_url
        else None
    )
    return cls(rails=rails, camera_mounts=mounts, lightbox_brightness=brightness)

CellStatus

Bases: BaseModel

Current status for one cell as reported by recent heartbeats.

Source code in core/src/armnet_core/models.py 
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
class CellStatus(BaseModel):
    """Current status for one cell as reported by recent heartbeats."""

    cell_id: str
    embodiment: Embodiment
    # What the cell is set up to work on. It can run any task belonging to this,
    # so the cell reports the environment and the job carries the task.
    environment: Environment
    status: CellRuntimeStatus
    current_job_id: Optional[str] = None
    last_heartbeat_at: Optional[datetime] = None
    # What this cell can vary and within what bounds, so a client can check an
    # override before submitting rather than learning from a failed job. None
    # from a cell too old to report it, which reads as "cannot check here".
    device_config: Optional[CellDeviceConfig] = None

cell_id instance-attribute 

cell_id: str

embodiment instance-attribute 

embodiment: Embodiment

environment instance-attribute 

environment: Environment

status instance-attribute 

status: CellRuntimeStatus

current_job_id class-attribute instance-attribute 

current_job_id: Optional[str] = None

last_heartbeat_at class-attribute instance-attribute 

last_heartbeat_at: Optional[datetime] = None

device_config class-attribute instance-attribute 

device_config: Optional[CellDeviceConfig] = None

CellStatusSummary

Bases: BaseModel

Aggregate availability for an embodiment, optionally scoped to a task.

Callers ask by task, which is what they think in; the orchestrator resolves it to the environment that can run it and matches cells on that. Both are echoed back. task is None when the summary aggregates across the whole embodiment, which is what a task-less job needs.

Source code in core/src/armnet_core/models.py 
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
class CellStatusSummary(BaseModel):
    """Aggregate availability for an embodiment, optionally scoped to a task.

    Callers ask by ``task``, which is what they think in; the orchestrator
    resolves it to the environment that can run it and matches cells on that.
    Both are echoed back. ``task`` is ``None`` when the summary aggregates
    across the whole embodiment, which is what a task-less job needs.
    """

    embodiment: Embodiment
    task: Optional[Task] = None
    environment: Optional[Environment] = None
    status: CellRuntimeStatus
    cells: list[CellStatus] = Field(default_factory=list)
    heartbeat_timeout_seconds: int = 20

embodiment instance-attribute 

embodiment: Embodiment

task class-attribute instance-attribute 

task: Optional[Task] = None

environment class-attribute instance-attribute 

environment: Optional[Environment] = None

status instance-attribute 

status: CellRuntimeStatus

cells class-attribute instance-attribute 

cells: list[CellStatus] = Field(default_factory=list)

heartbeat_timeout_seconds class-attribute instance-attribute 

heartbeat_timeout_seconds: int = 20

TeleopAction

Bases: BaseModel

A single remote-teleoperation action for a running job.

Sent by the client (sampling a local leader arm) to the orchestrator over a websocket, forwarded to the cell on the job's teleop NATS subject, and kept by the cell as a most-recent-value register. timestamp (unix seconds, on the client clock) is used to drop out-of-order/stale messages so the cell always holds the freshest action.

Source code in core/src/armnet_core/models.py 
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
class TeleopAction(BaseModel):
    """A single remote-teleoperation action for a running job.

    Sent by the client (sampling a local leader arm) to the orchestrator over a
    websocket, forwarded to the cell on the job's teleop NATS subject, and kept
    by the cell as a most-recent-value register. ``timestamp`` (unix seconds, on
    the client clock) is used to drop out-of-order/stale messages so the cell
    always holds the freshest action.
    """

    job_id: JobId
    timestamp: float
    # Joint-space command keyed like LeRobot's send_action input, e.g.
    # {"shoulder_pan.pos": 12.3, ...}.
    action: dict[str, float] = Field(default_factory=dict)

job_id instance-attribute 

job_id: JobId

timestamp instance-attribute 

timestamp: float

action class-attribute instance-attribute 

action: dict[str, float] = Field(default_factory=dict)

TeleopEvent

Bases: BaseModel

A discrete recording-control event for a running teleop job.

Travels the same path as :class:TeleopAction (client websocket → orchestrator → the job's teleop NATS subject → cell), but with different delivery semantics: actions are a freshest-wins stream where drops are fine, whereas events are rare, user-initiated commands that the cell queues in order so none is lost between runtime polls.

The event values mirror LeRobot's dataset-recording keyboard controls: Right Arrow ends the current episode and moves on (next_episode), Left Arrow discards and re-records it (rerecord_episode), and Esc stops the whole session (stop_recording).

Source code in core/src/armnet_core/models.py 
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
class TeleopEvent(BaseModel):
    """A discrete recording-control event for a running teleop job.

    Travels the same path as :class:`TeleopAction` (client websocket →
    orchestrator → the job's teleop NATS subject → cell), but with different
    delivery semantics: actions are a freshest-wins stream where drops are
    fine, whereas events are rare, user-initiated commands that the cell queues
    in order so none is lost between runtime polls.

    The ``event`` values mirror LeRobot's dataset-recording keyboard controls:
    Right Arrow ends the current episode and moves on (``next_episode``), Left
    Arrow discards and re-records it (``rerecord_episode``), and Esc stops the
    whole session (``stop_recording``).
    """

    job_id: JobId
    timestamp: float
    event: str = Field(..., pattern="^(next_episode|rerecord_episode|stop_recording)$")

job_id instance-attribute 

job_id: JobId

timestamp instance-attribute 

timestamp: float

event class-attribute instance-attribute 

event: str = Field(..., pattern='^(next_episode|rerecord_episode|stop_recording)$')

JobResult

Bases: BaseModel

Terminal result published by a cell on the results subject.

Also returned (embedded in :class:Job) by GET /jobs/{id} once the job is in a terminal state.

Source code in core/src/armnet_core/models.py 
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
class JobResult(BaseModel):
    """Terminal result published by a cell on the results subject.

    Also returned (embedded in :class:`Job`) by ``GET /jobs/{id}`` once the
    job is in a terminal state.
    """

    status: JobStatus = Field(
        ...,
        description="One of the terminal statuses (succeeded/failed/timeout/cancelled).",
    )
    exit_code: Optional[int] = Field(
        default=None,
        description="Container process exit code if the container ran to completion.",
    )
    stdout: str = Field(default="", description="Captured container stdout.")
    stderr: str = Field(default="", description="Captured container stderr.")
    error: Optional[str] = Field(
        default=None,
        description="Short, infra-side reason this job didn't run user code "
        "to completion: image pull failure, docker error, timeout, etc. "
        "Mutually exclusive with `traceback` in practice (one is a platform "
        "failure, the other is a user-code failure).",
    )
    traceback: Optional[str] = Field(
        default=None,
        description="Python traceback from the customer's `@main`-decorated "
        "function if it raised. Extracted by the cell from the "
        "`[armnet:traceback]:json` marker line in stdout. Capped at "
        "~64 KiB on the cell side; the full untruncated traceback is also "
        "in `stderr` for power users. None for successful jobs and for "
        "infra-side failures (those go in `error` instead).",
    )
    return_value: Optional[Any] = Field(
        default=None,
        description="Value returned by the customer's `@main`-decorated "
        "function. Extracted by the cell from the marker line that the "
        "`armnet-runtime` entrypoint prints to stdout. None if the "
        "function returned None or did not run to completion.",
    )
    started_at: Optional[datetime] = None
    finished_at: Optional[datetime] = None

    def raise_for_status(self) -> None:
        """Raise :class:`RemoteExecutionError` iff this result isn't ``SUCCEEDED``.

        The httpx-style "opt-in raising" pattern. Use it when you'd rather
        bail than branch on ``result.status``::

            result = execute(...)
            result.raise_for_status()
            do_thing(result.return_value)

        For successful results this is a no-op.
        """

        if self.status != JobStatus.SUCCEEDED:
            raise RemoteExecutionError(self)

    def __str__(self) -> str:
        """Human-readable rendering. Uses indentation to make tracebacks scannable.

        ``print(result)`` is intended to be the one-liner that tells you
        what happened. ``repr(result)`` (pydantic's default) still shows
        every field for debugging.
        """

        lines: list[str] = []
        head = f"JobResult(status={self.status.value}"
        if self.exit_code is not None:
            head += f", exit_code={self.exit_code}"
        head += ")"
        lines.append(head)
        if self.return_value is not None:
            lines.append(f"  return_value: {self.return_value!r}")
        if self.error:
            lines.append(f"  error: {self.error}")
        if self.traceback:
            lines.append("  traceback (from @main):")
            for tb_line in self.traceback.splitlines():
                lines.append(f"    {tb_line}")
        return "\n".join(lines)

status class-attribute instance-attribute 

status: JobStatus = Field(..., description='One of the terminal statuses (succeeded/failed/timeout/cancelled).')

exit_code class-attribute instance-attribute 

exit_code: Optional[int] = Field(default=None, description='Container process exit code if the container ran to completion.')

stdout class-attribute instance-attribute 

stdout: str = Field(default='', description='Captured container stdout.')

stderr class-attribute instance-attribute 

stderr: str = Field(default='', description='Captured container stderr.')

error class-attribute instance-attribute 

error: Optional[str] = Field(default=None, description="Short, infra-side reason this job didn't run user code to completion: image pull failure, docker error, timeout, etc. Mutually exclusive with `traceback` in practice (one is a platform failure, the other is a user-code failure).")

traceback class-attribute instance-attribute 

traceback: Optional[str] = Field(default=None, description="Python traceback from the customer's `@main`-decorated function if it raised. Extracted by the cell from the `[armnet:traceback]:json` marker line in stdout. Capped at ~64 KiB on the cell side; the full untruncated traceback is also in `stderr` for power users. None for successful jobs and for infra-side failures (those go in `error` instead).")

return_value class-attribute instance-attribute 

return_value: Optional[Any] = Field(default=None, description="Value returned by the customer's `@main`-decorated function. Extracted by the cell from the marker line that the `armnet-runtime` entrypoint prints to stdout. None if the function returned None or did not run to completion.")

started_at class-attribute instance-attribute 

started_at: Optional[datetime] = None

finished_at class-attribute instance-attribute 

finished_at: Optional[datetime] = None

raise_for_status 

raise_for_status() -> None

Raise :class:RemoteExecutionError iff this result isn't SUCCEEDED.

The httpx-style "opt-in raising" pattern. Use it when you'd rather bail than branch on result.status::

result = execute(...)
result.raise_for_status()
do_thing(result.return_value)

For successful results this is a no-op.

Source code in core/src/armnet_core/models.py 
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
def raise_for_status(self) -> None:
    """Raise :class:`RemoteExecutionError` iff this result isn't ``SUCCEEDED``.

    The httpx-style "opt-in raising" pattern. Use it when you'd rather
    bail than branch on ``result.status``::

        result = execute(...)
        result.raise_for_status()
        do_thing(result.return_value)

    For successful results this is a no-op.
    """

    if self.status != JobStatus.SUCCEEDED:
        raise RemoteExecutionError(self)

JobStatusTransition

Bases: BaseModel

One entry in a job's status-change audit trail.

Emitted by the orchestrator's GET /jobs/{id}/status-transitions so users can trace a job's lifecycle. detail carries optional context (e.g. the cell id on dispatch/run, or the failure message on a terminal state).

Source code in core/src/armnet_core/models.py 
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
class JobStatusTransition(BaseModel):
    """One entry in a job's status-change audit trail.

    Emitted by the orchestrator's ``GET /jobs/{id}/status-transitions`` so users
    can trace a job's lifecycle. ``detail`` carries optional context (e.g. the
    cell id on dispatch/run, or the failure message on a terminal state).
    """

    status: str
    detail: Optional[str] = None
    timestamp: datetime

status instance-attribute 

status: str

detail class-attribute instance-attribute 

detail: Optional[str] = None

timestamp instance-attribute 

timestamp: datetime

RemoteExecutionError

Bases: RuntimeError

Raised by :meth:JobResult.raise_for_status for non-succeeded results.

The exception's __str__ includes the underlying status, any platform-side error, and the user's traceback (if any), so an unhandled raise prints all the diagnostic context an operator needs. The original :class:JobResult is available as :attr:result for structured access.

Source code in core/src/armnet_core/models.py 
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
class RemoteExecutionError(RuntimeError):
    """Raised by :meth:`JobResult.raise_for_status` for non-succeeded results.

    The exception's ``__str__`` includes the underlying status, any
    platform-side ``error``, and the user's ``traceback`` (if any), so an
    unhandled raise prints all the diagnostic context an operator needs.
    The original :class:`JobResult` is available as :attr:`result` for
    structured access.
    """

    def __init__(self, result: "JobResult") -> None:
        self.result = result
        super().__init__(str(result))

result instance-attribute 

result = result

Job

Bases: BaseModel

The orchestrator's view of a job (response body of GET /jobs/{id}).

Source code in core/src/armnet_core/models.py 
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
class Job(BaseModel):
    """The orchestrator's view of a job (response body of GET /jobs/{id})."""

    id: JobId = Field(default_factory=_new_job_id)
    spec: JobSpec
    status: JobStatus = JobStatus.SUBMITTED
    created_at: datetime = Field(default_factory=_utcnow)
    updated_at: datetime = Field(default_factory=_utcnow)
    cell_id: Optional[str] = Field(
        default=None,
        description="ID of the cell that picked up the job, set on dispatch.",
    )
    result: Optional[JobResult] = None
    dispatched: Optional[bool] = Field(
        default=None,
        description=(
            "Only set on the POST /jobs response: True if the job was dispatched "
            "to a cell immediately, False if it was accepted but queued (no "
            "matching cell was free). None on all other responses. Clients can use "
            "this to decide whether to wait for logs/result or just report 'queued'."
        ),
    )

    def is_terminal(self) -> bool:
        return self.status in TerminalStatus

id class-attribute instance-attribute 

id: JobId = Field(default_factory=_new_job_id)

spec instance-attribute 

spec: JobSpec

status class-attribute instance-attribute 

status: JobStatus = JobStatus.SUBMITTED

created_at class-attribute instance-attribute 

created_at: datetime = Field(default_factory=_utcnow)

updated_at class-attribute instance-attribute 

updated_at: datetime = Field(default_factory=_utcnow)

cell_id class-attribute instance-attribute 

cell_id: Optional[str] = Field(default=None, description='ID of the cell that picked up the job, set on dispatch.')

result class-attribute instance-attribute 

result: Optional[JobResult] = None

dispatched class-attribute instance-attribute 

dispatched: Optional[bool] = Field(default=None, description="Only set on the POST /jobs response: True if the job was dispatched to a cell immediately, False if it was accepted but queued (no matching cell was free). None on all other responses. Clients can use this to decide whether to wait for logs/result or just report 'queued'.")

is_terminal 

is_terminal() -> bool
Source code in core/src/armnet_core/models.py 
1538
1539
def is_terminal(self) -> bool:
    return self.status in TerminalStatus

RegistryCredentials

Bases: BaseModel

Short-lived credentials for pushing to the platform's image registry.

Returned by POST /registry/credentials. The orchestrator mints these on demand by impersonating a dedicated image-pusher service account. Customers never need to know about Google Cloud, IAM, or service accounts — they just Image.build(...).push().

The password is an OAuth2 access token valid for ~1 hour. The SDK caches it on disk and refreshes within 5 minutes of expiry.

Source code in core/src/armnet_core/models.py 
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
class RegistryCredentials(BaseModel):
    """Short-lived credentials for pushing to the platform's image registry.

    Returned by ``POST /registry/credentials``. The orchestrator mints
    these on demand by impersonating a dedicated image-pusher service
    account. Customers never need to know about Google Cloud, IAM, or
    service accounts \u2014 they just ``Image.build(...).push()``.

    The ``password`` is an OAuth2 access token valid for ~1 hour. The
    SDK caches it on disk and refreshes within 5 minutes of expiry.
    """

    registry: str = Field(
        ...,
        description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.",
    )
    namespace: str = Field(
        ...,
        description="Path under the registry the customer is allowed to "
        "push to: '<project>/<repo>/<customer-id>'. The full image ref "
        "is '<registry>/<namespace>/<image-name>:<tag>'.",
    )
    username: str = Field(
        ...,
        description="docker login username. Always 'oauth2accesstoken' for AR.",
    )
    password: str = Field(
        ...,
        description="OAuth2 access token. Treat as a secret \u2014 valid for "
        "~1 hour and grants write access to the namespace above.",
    )
    expires_at: datetime = Field(
        ...,
        description="Wall-clock time at which the password becomes invalid.",
    )

registry class-attribute instance-attribute 

registry: str = Field(..., description="Registry hostname, e.g. 'europe-north1-docker.pkg.dev'.")

namespace class-attribute instance-attribute 

namespace: str = Field(..., description="Path under the registry the customer is allowed to push to: '<project>/<repo>/<customer-id>'. The full image ref is '<registry>/<namespace>/<image-name>:<tag>'.")

username class-attribute instance-attribute 

username: str = Field(..., description="docker login username. Always 'oauth2accesstoken' for AR.")

password class-attribute instance-attribute 

password: str = Field(..., description='OAuth2 access token. Treat as a secret — valid for ~1 hour and grants write access to the namespace above.')

expires_at class-attribute instance-attribute 

expires_at: datetime = Field(..., description='Wall-clock time at which the password becomes invalid.')

VolumeCredentials

Bases: BaseModel

Short-lived credentials for a user's cloud-backed volume prefix.

Source code in core/src/armnet_core/models.py 
1579
1580
1581
1582
1583
1584
1585
class VolumeCredentials(BaseModel):
    """Short-lived credentials for a user's cloud-backed volume prefix."""

    bucket: str
    prefix: str
    access_token: str
    expires_at: datetime

bucket instance-attribute 

bucket: str

prefix instance-attribute 

prefix: str

access_token instance-attribute 

access_token: str

expires_at instance-attribute 

expires_at: datetime

WhoAmI

Bases: BaseModel

Authenticated API identity.

Source code in core/src/armnet_core/models.py 
1588
1589
1590
1591
class WhoAmI(BaseModel):
    """Authenticated API identity."""

    username: str

username instance-attribute 

username: str

EmbodimentInfo

Bases: BaseModel

One known embodiment, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
1594
1595
1596
1597
1598
class EmbodimentInfo(BaseModel):
    """One known embodiment, as stored in the orchestrator database."""

    slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Embodiment = Field(..., description="Wire-format embodiment slug, e.g. 'lerobot/so-101'.")

description class-attribute instance-attribute 

description: Optional[str] = None

EmbodimentList

Bases: BaseModel

The set of embodiments the platform currently accepts (GET /embodiments).

Source code in core/src/armnet_core/models.py 
1601
1602
1603
1604
class EmbodimentList(BaseModel):
    """The set of embodiments the platform currently accepts (GET /embodiments)."""

    embodiments: list[EmbodimentInfo] = Field(default_factory=list)

embodiments class-attribute instance-attribute 

embodiments: list[EmbodimentInfo] = Field(default_factory=list)

TaskInfo

Bases: BaseModel

One known task, as stored in the orchestrator database.

Source code in core/src/armnet_core/models.py 
1607
1608
1609
1610
1611
class TaskInfo(BaseModel):
    """One known task, as stored in the orchestrator database."""

    slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")
    description: Optional[str] = None

slug class-attribute instance-attribute 

slug: Task = Field(..., description="Wire-format task slug, e.g. 'assemble_block_tower'.")

description class-attribute instance-attribute 

description: Optional[str] = None

TaskList

Bases: BaseModel

The set of tasks the platform currently accepts (GET /tasks).

Source code in core/src/armnet_core/models.py 
1614
1615
1616
1617
class TaskList(BaseModel):
    """The set of tasks the platform currently accepts (GET /tasks)."""

    tasks: list[TaskInfo] = Field(default_factory=list)

tasks class-attribute instance-attribute 

tasks: list[TaskInfo] = Field(default_factory=list)

SecretInfo

Bases: BaseModel

Secret metadata returned to clients. Never includes secret values.

Source code in core/src/armnet_core/models.py 
1620
1621
1622
1623
class SecretInfo(BaseModel):
    """Secret metadata returned to clients. Never includes secret values."""

    name: str

name instance-attribute 

name: str

SecretList

Bases: BaseModel

List of secret metadata for the authenticated user.

Source code in core/src/armnet_core/models.py 
1626
1627
1628
1629
class SecretList(BaseModel):
    """List of secret metadata for the authenticated user."""

    secrets: list[SecretInfo] = Field(default_factory=list)

secrets class-attribute instance-attribute 

secrets: list[SecretInfo] = Field(default_factory=list)

SecretCreateRequest

Bases: BaseModel

Create or replace one user secret.

Source code in core/src/armnet_core/models.py 
1632
1633
1634
1635
1636
class SecretCreateRequest(BaseModel):
    """Create or replace one user secret."""

    name: str
    value: str

name instance-attribute 

name: str

value instance-attribute 

value: str

canonical_job_id 

canonical_job_id(value: Any) -> str

Return one textual representation for UUID job ids.

Postgres returns dashed UUID text while older clients and NATS messages used the equivalent 32-character hex form. Canonicalizing at every model boundary prevents those spellings from becoming separate in-memory cache keys. Synthetic local-development ids remain unchanged.

Source code in core/src/armnet_core/models.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
def canonical_job_id(value: Any) -> str:
    """Return one textual representation for UUID job ids.

    Postgres returns dashed UUID text while older clients and NATS messages used
    the equivalent 32-character hex form. Canonicalizing at every model boundary
    prevents those spellings from becoming separate in-memory cache keys.
    Synthetic local-development ids remain unchanged.
    """
    text = str(value)
    try:
        return str(UUID(text))
    except ValueError:
        return text

armnet_core.auth

Shared authentication constants.

API_KEY_ENV module-attribute 

API_KEY_ENV = 'ARMNET_API_KEY'

API_KEY_HEADER module-attribute 

API_KEY_HEADER = 'X-Armnet-Api-Key'

armnet_core.environment

Contracts for instrumented workcell environments.

An environment is a kind of physical workcell together with the tasks that can be performed in it: the BusyBox panel and its switches, a table of wooden blocks, a kitchen worktop. A cell is assigned one environment and can then run any task that environment offers without being reconfigured.

The contract has two halves, deliberately kept apart.

:class:Environment is a static description. It names its tasks and knows how to open a session against the hardware, but holds no connection itself, so the orchestrator can ask which environment owns a task without any driver being installed.

:class:InstrumentedCell is a live session against one cell for one job. It reads the instrumentation, decides whether an episode finished, restores the scene between episodes, and contributes its readings to the recorded dataset. Everything specific to a kind of workcell lives behind it, which is what keeps job code free of any particular environment's vocabulary.

Environments register themselves under the armnet.environments entry-point group, so installing a package is all it takes to add one.

ENTRY_POINT_GROUP module-attribute 

ENTRY_POINT_GROUP = 'armnet.environments'

CompletionStatus

Bases: NamedTuple

Whether an episode has finished, how it went, and who decided.

bool(status) is status.complete, so a caller that only cares whether to keep going can use it directly.

scored_by names the judge — an environment's own name when its instrumentation decided, "operator" for a human verdict, "monitor" for the automated completion model. Recorded with results so a success rate can be read knowing what produced it.

Source code in core/src/armnet_core/environment.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class CompletionStatus(NamedTuple):
    """Whether an episode has finished, how it went, and who decided.

    ``bool(status)`` is ``status.complete``, so a caller that only cares
    whether to keep going can use it directly.

    ``scored_by`` names the judge — an environment's own name when its
    instrumentation decided, ``"operator"`` for a human verdict, ``"monitor"``
    for the automated completion model. Recorded with results so a success rate
    can be read knowing what produced it.
    """

    complete: bool
    success: bool
    scored_by: Optional[str] = None

    def __bool__(self) -> bool:
        return self.complete

complete instance-attribute 

complete: bool

success instance-attribute 

success: bool

scored_by class-attribute instance-attribute 

scored_by: Optional[str] = None

Reading dataclass

One instrumentation channel's current value.

continuous separates channels that sweep through values as they move, such as a slider or a dial, from ones that jump between a few states, such as a switch. Anything watching the instrumentation has to treat those differently: a switch is worth reporting on every change, while a dial being turned would otherwise report continuously.

Source code in core/src/armnet_core/environment.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True)
class Reading:
    """One instrumentation channel's current value.

    ``continuous`` separates channels that sweep through values as they move,
    such as a slider or a dial, from ones that jump between a few states, such
    as a switch. Anything watching the instrumentation has to treat those
    differently: a switch is worth reporting on every change, while a dial
    being turned would otherwise report continuously.
    """

    name: str
    kind: str
    value: float
    continuous: bool = False
    label: Optional[str] = None

    def describe(self) -> str:
        return f"{self.name}: {self.label if self.label is not None else self.value}"

name instance-attribute 

name: str

kind instance-attribute 

kind: str

value instance-attribute 

value: float

continuous class-attribute instance-attribute 

continuous: bool = False

label class-attribute instance-attribute 

label: Optional[str] = None

describe 

describe() -> str
Source code in core/src/armnet_core/environment.py 
82
83
def describe(self) -> str:
    return f"{self.name}: {self.label if self.label is not None else self.value}"

TaskSpec dataclass

A task an environment offers.

instruction is the natural-language form handed to a policy, and is what the platform stores as the task description.

Source code in core/src/armnet_core/environment.py 
86
87
88
89
90
91
92
93
94
95
@dataclass(frozen=True)
class TaskSpec:
    """A task an environment offers.

    ``instruction`` is the natural-language form handed to a policy, and is
    what the platform stores as the task description.
    """

    slug: str
    instruction: str

slug instance-attribute 

slug: str

instruction instance-attribute 

instruction: str

InstrumentedCell

Bases: Protocol

A live session against one cell's instrumentation, for one job.

Implementations own everything about their kind of workcell: the transport, how a task's goal is expressed, and how the scene is restored between episodes. Job code sees only this interface, through ctx.cell.

No method here may raise because the hardware misbehaved. Instrumentation is an aid to a rollout, never the thing that fails one, so a silent panel reports "nothing known" and lets the operator decide.

Source code in core/src/armnet_core/environment.py 
 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
@runtime_checkable
class InstrumentedCell(Protocol):
    """A live session against one cell's instrumentation, for one job.

    Implementations own everything about their kind of workcell: the transport,
    how a task's goal is expressed, and how the scene is restored between
    episodes. Job code sees only this interface, through ``ctx.cell``.

    No method here may raise because the hardware misbehaved. Instrumentation
    is an aid to a rollout, never the thing that fails one, so a silent panel
    reports "nothing known" and lets the operator decide.
    """

    def instrument(self, robot: Any) -> Any:
        """Return ``robot``, wrapped if this environment records alongside it.

        The wrapper must not change the observation the policy sees. Readings
        belong in a sidecar, not in ``observation.state``, or a policy trained
        without the instrumentation will behave differently with it attached.
        """

    def readings(self) -> Mapping[str, Reading]:
        """Current value of every channel, empty if nothing has been heard."""

    def begin_episode(self) -> Iterable[str]:
        """Open an episode window and report anything wrong with the scene.

        Returns human-readable problems — a goal already satisfied before the
        rollout starts, say — which make the episode unscorable rather than
        failed. An empty iterable means the scene is ready.
        """

    def episode_status(self, *, final: bool = False) -> CompletionStatus:
        """Whether the instrumentation can call this episode yet.

        ``complete=False`` means "no verdict", whether because the goal is
        unmet or because nothing readable bears on it. Callers fall back to
        another judge; they cannot distinguish, and should not need to.

        ``final=True`` says the episode is out of time and asks for a last
        word. An unmet goal is then a failure rather than an abstention: an
        operator should not be asked to confirm a failure the instrumentation
        can see. Channels that read nothing still abstain.
        """

    def reset_scene(
        self,
        *,
        confirm: bool,
        reset_cell: Callable[[bool], None],
        set_rail: Callable[[float], None] | None = None,
    ) -> None:
        """Restore the state this task starts from.

        ``reset_cell`` returns the arm to rest and, when passed ``True``, waits
        for an operator. Implementations decide whether a human is needed;
        ``confirm=True`` means the caller has already decided one is.

        ``set_rail``, when provided, temporarily positions the cell's rails to a
        0..1 staging fraction before replaying keypoints authored at that pose.
        The runtime restores every rail to its job episode position after
        ``reset_scene`` returns or fails; environments must not treat the plan
        pose as a replacement for the job position.
        """

    def attach_dataset(self, dataset_root: Any) -> None:
        """Begin recording readings beside a dataset being written."""

    def record_frame(self) -> None:
        """Buffer the current readings against the frame just recorded."""

    def commit_episode(self, episode_index: int) -> None:
        """Persist buffered readings for a saved episode."""

    def discard_episode(self) -> None:
        """Drop buffered readings for an episode that will not be saved."""

    def close(self) -> None:
        """Release the connection."""

instrument 

instrument(robot: Any) -> Any

Return robot, wrapped if this environment records alongside it.

The wrapper must not change the observation the policy sees. Readings belong in a sidecar, not in observation.state, or a policy trained without the instrumentation will behave differently with it attached.

Source code in core/src/armnet_core/environment.py 
111
112
113
114
115
116
117
def instrument(self, robot: Any) -> Any:
    """Return ``robot``, wrapped if this environment records alongside it.

    The wrapper must not change the observation the policy sees. Readings
    belong in a sidecar, not in ``observation.state``, or a policy trained
    without the instrumentation will behave differently with it attached.
    """

readings 

readings() -> Mapping[str, Reading]

Current value of every channel, empty if nothing has been heard.

Source code in core/src/armnet_core/environment.py 
119
120
def readings(self) -> Mapping[str, Reading]:
    """Current value of every channel, empty if nothing has been heard."""

begin_episode 

begin_episode() -> Iterable[str]

Open an episode window and report anything wrong with the scene.

Returns human-readable problems — a goal already satisfied before the rollout starts, say — which make the episode unscorable rather than failed. An empty iterable means the scene is ready.

Source code in core/src/armnet_core/environment.py 
122
123
124
125
126
127
128
def begin_episode(self) -> Iterable[str]:
    """Open an episode window and report anything wrong with the scene.

    Returns human-readable problems — a goal already satisfied before the
    rollout starts, say — which make the episode unscorable rather than
    failed. An empty iterable means the scene is ready.
    """

episode_status 

episode_status(*, final: bool = False) -> CompletionStatus

Whether the instrumentation can call this episode yet.

complete=False means "no verdict", whether because the goal is unmet or because nothing readable bears on it. Callers fall back to another judge; they cannot distinguish, and should not need to.

final=True says the episode is out of time and asks for a last word. An unmet goal is then a failure rather than an abstention: an operator should not be asked to confirm a failure the instrumentation can see. Channels that read nothing still abstain.

Source code in core/src/armnet_core/environment.py 
130
131
132
133
134
135
136
137
138
139
140
141
def episode_status(self, *, final: bool = False) -> CompletionStatus:
    """Whether the instrumentation can call this episode yet.

    ``complete=False`` means "no verdict", whether because the goal is
    unmet or because nothing readable bears on it. Callers fall back to
    another judge; they cannot distinguish, and should not need to.

    ``final=True`` says the episode is out of time and asks for a last
    word. An unmet goal is then a failure rather than an abstention: an
    operator should not be asked to confirm a failure the instrumentation
    can see. Channels that read nothing still abstain.
    """

reset_scene 

reset_scene(*, confirm: bool, reset_cell: Callable[[bool], None], set_rail: Callable[[float], None] | None = None) -> None

Restore the state this task starts from.

reset_cell returns the arm to rest and, when passed True, waits for an operator. Implementations decide whether a human is needed; confirm=True means the caller has already decided one is.

set_rail, when provided, temporarily positions the cell's rails to a 0..1 staging fraction before replaying keypoints authored at that pose. The runtime restores every rail to its job episode position after reset_scene returns or fails; environments must not treat the plan pose as a replacement for the job position.

Source code in core/src/armnet_core/environment.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def reset_scene(
    self,
    *,
    confirm: bool,
    reset_cell: Callable[[bool], None],
    set_rail: Callable[[float], None] | None = None,
) -> None:
    """Restore the state this task starts from.

    ``reset_cell`` returns the arm to rest and, when passed ``True``, waits
    for an operator. Implementations decide whether a human is needed;
    ``confirm=True`` means the caller has already decided one is.

    ``set_rail``, when provided, temporarily positions the cell's rails to a
    0..1 staging fraction before replaying keypoints authored at that pose.
    The runtime restores every rail to its job episode position after
    ``reset_scene`` returns or fails; environments must not treat the plan
    pose as a replacement for the job position.
    """

attach_dataset 

attach_dataset(dataset_root: Any) -> None

Begin recording readings beside a dataset being written.

Source code in core/src/armnet_core/environment.py 
163
164
def attach_dataset(self, dataset_root: Any) -> None:
    """Begin recording readings beside a dataset being written."""

record_frame 

record_frame() -> None

Buffer the current readings against the frame just recorded.

Source code in core/src/armnet_core/environment.py 
166
167
def record_frame(self) -> None:
    """Buffer the current readings against the frame just recorded."""

commit_episode 

commit_episode(episode_index: int) -> None

Persist buffered readings for a saved episode.

Source code in core/src/armnet_core/environment.py 
169
170
def commit_episode(self, episode_index: int) -> None:
    """Persist buffered readings for a saved episode."""

discard_episode 

discard_episode() -> None

Drop buffered readings for an episode that will not be saved.

Source code in core/src/armnet_core/environment.py 
172
173
def discard_episode(self) -> None:
    """Drop buffered readings for an episode that will not be saved."""

close 

close() -> None

Release the connection.

Source code in core/src/armnet_core/environment.py 
175
176
def close(self) -> None:
    """Release the connection."""

Environment

Bases: Protocol

A kind of workcell and the tasks it offers.

Source code in core/src/armnet_core/environment.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@runtime_checkable
class Environment(Protocol):
    """A kind of workcell and the tasks it offers."""

    name: str

    def tasks(self) -> Mapping[str, TaskSpec]:
        """Every task this environment can run, keyed by slug."""

    def connect(
        self,
        config: Mapping[str, Any],
        *,
        task: Optional[str],
        report_progress: Optional[Callable[[str], None]] = None,
    ) -> Optional[InstrumentedCell]:
        """Open a session, or return ``None`` if this cell has no hardware.

        ``config`` is the cell's environment block, passed through untouched
        from its configuration file.
        """

name instance-attribute 

name: str

tasks 

tasks() -> Mapping[str, TaskSpec]

Every task this environment can run, keyed by slug.

Source code in core/src/armnet_core/environment.py 
185
186
def tasks(self) -> Mapping[str, TaskSpec]:
    """Every task this environment can run, keyed by slug."""

connect 

connect(config: Mapping[str, Any], *, task: Optional[str], report_progress: Optional[Callable[[str], None]] = None) -> Optional[InstrumentedCell]

Open a session, or return None if this cell has no hardware.

config is the cell's environment block, passed through untouched from its configuration file.

Source code in core/src/armnet_core/environment.py 
188
189
190
191
192
193
194
195
196
197
198
199
def connect(
    self,
    config: Mapping[str, Any],
    *,
    task: Optional[str],
    report_progress: Optional[Callable[[str], None]] = None,
) -> Optional[InstrumentedCell]:
    """Open a session, or return ``None`` if this cell has no hardware.

    ``config`` is the cell's environment block, passed through untouched
    from its configuration file.
    """

EnvironmentNotFound

Bases: LookupError

Raised when no installed package provides the named environment.

Source code in core/src/armnet_core/environment.py 
202
203
class EnvironmentNotFound(LookupError):
    """Raised when no installed package provides the named environment."""

available_environments 

available_environments() -> list[str]

Names of every environment provided by an installed package.

Source code in core/src/armnet_core/environment.py 
212
213
214
215
def available_environments() -> list[str]:
    """Names of every environment provided by an installed package."""

    return sorted({entry.name for entry in _entry_points()})

environment_for 

environment_for(name: str) -> Environment

Load the environment registered under name.

Loading is deferred to here so that importing this module costs nothing: an environment's drivers are only imported by a process that actually intends to talk to one.

Source code in core/src/armnet_core/environment.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def environment_for(name: str) -> Environment:
    """Load the environment registered under ``name``.

    Loading is deferred to here so that importing this module costs nothing:
    an environment's drivers are only imported by a process that actually
    intends to talk to one.
    """

    for entry in _entry_points():
        if entry.name == name:
            return entry.load()()
    installed = available_environments()
    raise EnvironmentNotFound(
        f"no environment named {name!r} is installed"
        + (f" (installed: {', '.join(installed)})" if installed else "")
    )

task_environment 

task_environment(task: Optional[str]) -> Optional[str]

Name the environment that owns task, or None if unowned.

Used where the mapping is needed without a database — chiefly tests and local runs. The orchestrator resolves through the tasks table instead, so routing never depends on which environment packages it has installed.

Source code in core/src/armnet_core/environment.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def task_environment(task: Optional[str]) -> Optional[str]:
    """Name the environment that owns ``task``, or ``None`` if unowned.

    Used where the mapping is needed without a database — chiefly tests and
    local runs. The orchestrator resolves through the ``tasks`` table instead,
    so routing never depends on which environment packages it has installed.
    """

    if not task:
        return None
    for name in available_environments():
        if task in environment_for(name).tasks():
            return name
    return None

armnet_core.manual_reset

Task-driven manual-reset models and deterministic layout geometry.

All collision and clearance calculations happen in source-frame pixels so a shirt-size square remains square on a non-square image. The resulting geometry is normalized for direct use by image/SVG renderers.

MAX_MANUAL_RESET_ELEMENTS module-attribute 

MAX_MANUAL_RESET_ELEMENTS = 12

POINT_RADIUS_SCALE module-attribute 

POINT_RADIUS_SCALE = 0.006

POINT_ARROW_LENGTH_SCALE module-attribute 

POINT_ARROW_LENGTH_SCALE = 0.04

DEFAULT_LAYOUT_ATTEMPTS module-attribute 

DEFAULT_LAYOUT_ATTEMPTS = 256

EDGE_DISTANCE_SCALES module-attribute 

EDGE_DISTANCE_SCALES = MappingProxyType({ShirtSize.XS: 0.015, ShirtSize.S: 0.035, ShirtSize.M: 0.07, ShirtSize.L: 0.14, ShirtSize.XL: 0.28})

BOX_LENGTH_SCALES module-attribute 

BOX_LENGTH_SCALES = MappingProxyType({ShirtSize.XS: 0.035, ShirtSize.S: 0.055, ShirtSize.M: 0.085, ShirtSize.L: 0.13, ShirtSize.XL: 0.2})

ManualResetElement module-attribute 

ManualResetElement = Annotated[ManualResetPoint | ManualResetBoundingBox, Field(discriminator='type')]

ManualResetElementGeometry module-attribute 

ManualResetElementGeometry = Annotated[ManualResetPointGeometry | ManualResetBoundingBoxGeometry, Field(discriminator='type')]

PixelPoint module-attribute 

PixelPoint = tuple[float, float]

PixelPolygon module-attribute 

PixelPolygon = tuple[PixelPoint, ...]

PixelFootprint module-attribute 

PixelFootprint = _PixelCircle | _PixelBox

ShirtSize

Bases: str, Enum

Discrete physical scale used by manual-reset instructions.

Source code in core/src/armnet_core/manual_reset.py 
44
45
46
47
48
49
50
51
class ShirtSize(str, Enum):
    """Discrete physical scale used by manual-reset instructions."""

    XS = "XS"
    S = "S"
    M = "M"
    L = "L"
    XL = "XL"

XS class-attribute instance-attribute 

XS = 'XS'

S class-attribute instance-attribute 

S = 'S'

M class-attribute instance-attribute 

M = 'M'

L class-attribute instance-attribute 

L = 'L'

XL class-attribute instance-attribute 

XL = 'XL'

ManualResetPoint

Bases: _ManualResetElement

A small point-like object with an orientation indicator.

Source code in core/src/armnet_core/manual_reset.py 
92
93
94
95
class ManualResetPoint(_ManualResetElement):
    """A small point-like object with an orientation indicator."""

    type: Literal["point"] = "point"

type class-attribute instance-attribute 

type: Literal['point'] = 'point'

ManualResetBoundingBox

Bases: _ManualResetElement

A rectangular object whose dimensions use the box shirt-size scale.

Source code in core/src/armnet_core/manual_reset.py 
 98
 99
100
101
102
103
class ManualResetBoundingBox(_ManualResetElement):
    """A rectangular object whose dimensions use the box shirt-size scale."""

    type: Literal["bounding_box"] = "bounding_box"
    horizontal_length: ShirtSize
    vertical_length: ShirtSize

type class-attribute instance-attribute 

type: Literal['bounding_box'] = 'bounding_box'

horizontal_length instance-attribute 

horizontal_length: ShirtSize

vertical_length instance-attribute 

vertical_length: ShirtSize

ManualResetConfig

Bases: _FrozenModel

Versioned task reset instructions in stable display/sampling order.

Source code in core/src/armnet_core/manual_reset.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class ManualResetConfig(_FrozenModel):
    """Versioned task reset instructions in stable display/sampling order."""

    schema_version: Literal[1] = 1
    elements: tuple[ManualResetElement, ...] = Field(
        default=(),
        max_length=MAX_MANUAL_RESET_ELEMENTS,
    )

    @model_validator(mode="after")
    def _unique_element_names(self) -> ManualResetConfig:
        names = [element.name.casefold() for element in self.elements]
        if len(set(names)) != len(names):
            raise ValueError("manual-reset element names must be unique")
        return self

schema_version class-attribute instance-attribute 

schema_version: Literal[1] = 1

elements class-attribute instance-attribute 

elements: tuple[ManualResetElement, ...] = Field(default=(), max_length=MAX_MANUAL_RESET_ELEMENTS)

NormalizedLine

Bases: _FrozenModel

A normalized line segment.

Source code in core/src/armnet_core/manual_reset.py 
129
130
131
132
133
class NormalizedLine(_FrozenModel):
    """A normalized line segment."""

    start: NormalizedPoint
    end: NormalizedPoint

start instance-attribute 

start: NormalizedPoint

end instance-attribute 

end: NormalizedPoint

ManualResetPointGeometry

Bases: _FrozenModel

Normalized rendering geometry for a sampled point element.

Source code in core/src/armnet_core/manual_reset.py 
136
137
138
139
140
141
142
143
144
145
146
class ManualResetPointGeometry(_FrozenModel):
    """Normalized rendering geometry for a sampled point element."""

    type: Literal["point"] = "point"
    element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)
    name: str
    center: NormalizedPoint
    radius_x: float = Field(gt=0)
    radius_y: float = Field(gt=0)
    rotation_degrees: float
    arrow: NormalizedLine

type class-attribute instance-attribute 

type: Literal['point'] = 'point'

element_index class-attribute instance-attribute 

element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)

name instance-attribute 

name: str

center instance-attribute 

center: NormalizedPoint

radius_x class-attribute instance-attribute 

radius_x: float = Field(gt=0)

radius_y class-attribute instance-attribute 

radius_y: float = Field(gt=0)

rotation_degrees instance-attribute 

rotation_degrees: float

arrow instance-attribute 

arrow: NormalizedLine

ManualResetBoundingBoxGeometry

Bases: _FrozenModel

Normalized rendering geometry for a sampled rotated rectangle.

Source code in core/src/armnet_core/manual_reset.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class ManualResetBoundingBoxGeometry(_FrozenModel):
    """Normalized rendering geometry for a sampled rotated rectangle."""

    type: Literal["bounding_box"] = "bounding_box"
    element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)
    name: str
    center: NormalizedPoint
    corners: tuple[
        NormalizedPoint,
        NormalizedPoint,
        NormalizedPoint,
        NormalizedPoint,
    ]
    rotation_degrees: float
    arrow: NormalizedLine

type class-attribute instance-attribute 

type: Literal['bounding_box'] = 'bounding_box'

element_index class-attribute instance-attribute 

element_index: int = Field(ge=0, lt=MAX_MANUAL_RESET_ELEMENTS)

name instance-attribute 

name: str

center instance-attribute 

center: NormalizedPoint

corners instance-attribute 

corners: tuple[NormalizedPoint, NormalizedPoint, NormalizedPoint, NormalizedPoint]

rotation_degrees instance-attribute 

rotation_degrees: float

arrow instance-attribute 

arrow: NormalizedLine

ManualResetLayout

Bases: _FrozenModel

A deterministic layout ready for normalized image/SVG rendering.

Source code in core/src/armnet_core/manual_reset.py 
172
173
174
175
176
177
class ManualResetLayout(_FrozenModel):
    """A deterministic layout ready for normalized image/SVG rendering."""

    source_frame_width: int = Field(gt=0)
    source_frame_height: int = Field(gt=0)
    elements: tuple[ManualResetElementGeometry, ...]

source_frame_width class-attribute instance-attribute 

source_frame_width: int = Field(gt=0)

source_frame_height class-attribute instance-attribute 

source_frame_height: int = Field(gt=0)

elements instance-attribute 

elements: tuple[ManualResetElementGeometry, ...]

ManualResetLayoutError

Bases: ValueError

Raised when a configured layout cannot fit after bounded retries.

Source code in core/src/armnet_core/manual_reset.py 
180
181
class ManualResetLayoutError(ValueError):
    """Raised when a configured layout cannot fit after bounded retries."""

sample_manual_reset_layout 

sample_manual_reset_layout(config: ManualResetConfig, calibration: SceneCalibration | None, *, source_frame_width: int | None = None, source_frame_height: int | None = None, seed: int, reset_index: int, cell_id: str, task_slug: str, max_attempts: int = DEFAULT_LAYOUT_ATTEMPTS) -> ManualResetLayout

Sample a stable, non-overlapping task reset layout.

source_frame_width and source_frame_height may be omitted when a calibration is supplied, in which case its source dimensions are used. Every element gets at most max_attempts deterministic candidates.

Source code in core/src/armnet_core/manual_reset.py 
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
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
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def sample_manual_reset_layout(
    config: ManualResetConfig,
    calibration: SceneCalibration | None,
    *,
    source_frame_width: int | None = None,
    source_frame_height: int | None = None,
    seed: int,
    reset_index: int,
    cell_id: str,
    task_slug: str,
    max_attempts: int = DEFAULT_LAYOUT_ATTEMPTS,
) -> ManualResetLayout:
    """Sample a stable, non-overlapping task reset layout.

    ``source_frame_width`` and ``source_frame_height`` may be omitted when a
    calibration is supplied, in which case its source dimensions are used.
    Every element gets at most ``max_attempts`` deterministic candidates.
    """

    config = ManualResetConfig.model_validate(config)
    if reset_index < 0:
        raise ValueError("reset_index must be non-negative")
    if not cell_id.strip():
        raise ValueError("cell_id must be nonempty")
    if not task_slug.strip():
        raise ValueError("task_slug must be nonempty")
    if max_attempts <= 0:
        raise ValueError("max_attempts must be positive")

    if source_frame_width is None:
        source_frame_width = (
            calibration.source_frame_width if calibration is not None else None
        )
    if source_frame_height is None:
        source_frame_height = (
            calibration.source_frame_height if calibration is not None else None
        )
    if (
        source_frame_width is None
        or source_frame_height is None
        or source_frame_width <= 0
        or source_frame_height <= 0
    ):
        raise ValueError("positive source frame width and height are required")

    normalized_scene = (
        calibration.vertices if calibration is not None else FULL_FRAME_VERTICES
    )
    pixel_scene = tuple(
        normalized_to_pixel(
            vertex,
            width=source_frame_width,
            height=source_frame_height,
        )
        for vertex in normalized_scene
    )
    # Shirt sizes are camera-frame references, matching the rendered size guide.
    # The manipulation polygon still controls where objects may be placed and
    # which edges clearance is measured from, but changing that polygon must not
    # silently change what XS/S/M/L/XL mean.
    short_axis = min(source_frame_width, source_frame_height)

    triangles = tuple(
        tuple(
            normalized_to_pixel(
                point,
                width=source_frame_width,
                height=source_frame_height,
            )
            for point in triangle
        )
        for triangle in _triangulate(normalized_scene)
    )
    weighted_triangles = tuple(
        (triangle, polygon_area(triangle)) for triangle in triangles
    )
    total_area = sum(area for _, area in weighted_triangles)
    calibration_id = (
        calibration.id if calibration is not None else "full-normalized-frame"
    )

    render_elements: list[ManualResetElementGeometry] = []
    footprints: list[PixelFootprint] = []
    for element_index, element in enumerate(config.elements):
        clearance = EDGE_DISTANCE_SCALES[element.min_edge_distance] * short_axis
        for attempt in range(max_attempts):
            random = _StableRandom(
                [
                    int(seed),
                    reset_index,
                    cell_id,
                    calibration_id,
                    task_slug,
                    element_index,
                    element.name,
                    attempt,
                ]
            )
            rotation = (
                (2.0 * random.unit() - 1.0)
                * element.rotation_range_degrees
            )
            center = _sample_weighted_triangle(
                weighted_triangles,
                total_area,
                random,
            )
            footprint = _make_footprint(
                element,
                center=center,
                rotation_degrees=rotation,
                short_axis=short_axis,
            )
            if not _footprint_inside_scene(
                footprint,
                pixel_scene,
                clearance=clearance,
            ):
                continue
            if any(_footprints_overlap(footprint, other) for other in footprints):
                continue

            footprints.append(footprint)
            render_elements.append(
                _render_geometry(
                    element,
                    element_index=element_index,
                    footprint=footprint,
                    rotation_degrees=rotation,
                    short_axis=short_axis,
                    width=source_frame_width,
                    height=source_frame_height,
                )
            )
            break
        else:
            raise ManualResetLayoutError(
                "unable to place manual-reset element "
                f"{element_index + 1} ({element.name!r}) after "
                f"{max_attempts} deterministic attempts; its footprint and "
                "edge clearance do not fit without overlap"
            )

    return ManualResetLayout(
        source_frame_width=source_frame_width,
        source_frame_height=source_frame_height,
        elements=tuple(render_elements),
    )

armnet_core.scene_calibration

Scene-calibration wire model and dependency-free polygon geometry.

NormalizedPoint module-attribute 

NormalizedPoint = tuple[float, float]

FULL_FRAME_VERTICES module-attribute 

FULL_FRAME_VERTICES: tuple[NormalizedPoint, ...] = ((0.0, 0.0), (1.0, 0.0), (1.0, 1.0), (0.0, 1.0))

MIN_NORMALIZED_POLYGON_AREA module-attribute 

MIN_NORMALIZED_POLYGON_AREA = 1e-06

SceneCalibration

Bases: BaseModel

One immutable manipulation-area calibration snapshot.

Source code in core/src/armnet_core/scene_calibration.py 
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
class SceneCalibration(BaseModel):
    """One immutable manipulation-area calibration snapshot."""

    model_config = ConfigDict(
        extra="forbid",
        frozen=True,
        allow_inf_nan=False,
    )

    id: str = Field(min_length=1)
    cell_id: str = Field(min_length=1)
    camera_name: str = Field(min_length=1)
    vertices: tuple[NormalizedPoint, ...]
    source_frame_width: int = Field(gt=0)
    source_frame_height: int = Field(gt=0)
    schema_version: int = Field(default=1, gt=0)
    note: str | None = None
    created_at: datetime

    @field_validator("id", "cell_id", "camera_name")
    @classmethod
    def _nonempty_identifier(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("identifier must be nonempty")
        return value

    @field_validator("vertices", mode="before")
    @classmethod
    def _valid_polygon(cls, value: Any) -> tuple[NormalizedPoint, ...]:
        return validate_normalized_polygon(value)

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', frozen=True, allow_inf_nan=False)

id class-attribute instance-attribute 

id: str = Field(min_length=1)

cell_id class-attribute instance-attribute 

cell_id: str = Field(min_length=1)

camera_name class-attribute instance-attribute 

camera_name: str = Field(min_length=1)

vertices instance-attribute 

vertices: tuple[NormalizedPoint, ...]

source_frame_width class-attribute instance-attribute 

source_frame_width: int = Field(gt=0)

source_frame_height class-attribute instance-attribute 

source_frame_height: int = Field(gt=0)

schema_version class-attribute instance-attribute 

schema_version: int = Field(default=1, gt=0)

note class-attribute instance-attribute 

note: str | None = None

created_at instance-attribute 

created_at: datetime

polygon_area 

polygon_area(vertices: Iterable[NormalizedPoint]) -> float

Return the unsigned area of an ordered polygon.

Source code in core/src/armnet_core/scene_calibration.py 
25
26
27
28
def polygon_area(vertices: Iterable[NormalizedPoint]) -> float:
    """Return the unsigned area of an ordered polygon."""
    points = tuple(vertices)
    return abs(_signed_area(points))

validate_normalized_polygon 

validate_normalized_polygon(vertices: Any, *, minimum_area: float = MIN_NORMALIZED_POLYGON_AREA) -> tuple[NormalizedPoint, ...]

Validate and normalize an ordered simple polygon in the unit frame.

Source code in core/src/armnet_core/scene_calibration.py 
31
32
33
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
def validate_normalized_polygon(
    vertices: Any,
    *,
    minimum_area: float = MIN_NORMALIZED_POLYGON_AREA,
) -> tuple[NormalizedPoint, ...]:
    """Validate and normalize an ordered simple polygon in the unit frame."""
    if not isinstance(vertices, (list, tuple)) or len(vertices) < 3:
        raise ValueError("polygon must contain at least 3 ordered vertices")

    points: list[NormalizedPoint] = []
    for index, vertex in enumerate(vertices):
        if not isinstance(vertex, (list, tuple)) or len(vertex) != 2:
            raise ValueError(f"vertex {index} must be an [x, y] pair")
        x, y = vertex
        if (
            isinstance(x, bool)
            or isinstance(y, bool)
            or not isinstance(x, (int, float))
            or not isinstance(y, (int, float))
        ):
            raise ValueError(  # noqa: TRY004 - public validator contract
                f"vertex {index} coordinates must be numbers"
            )
        point = (float(x), float(y))
        if not math.isfinite(point[0]) or not math.isfinite(point[1]):
            raise ValueError(f"vertex {index} coordinates must be finite")
        if not 0.0 <= point[0] <= 1.0 or not 0.0 <= point[1] <= 1.0:
            raise ValueError(f"vertex {index} coordinates must be in [0, 1]")
        points.append(point)

    if len(set(points)) != len(points):
        raise ValueError("polygon vertices must be distinct")
    if _has_self_intersection(points):
        raise ValueError("polygon must not self-intersect")
    if polygon_area(points) < minimum_area:
        raise ValueError(f"polygon area must be at least {minimum_area:g}")
    return tuple(points)

point_in_polygon 

point_in_polygon(point: NormalizedPoint, vertices: Iterable[NormalizedPoint]) -> bool

Return whether a point is inside or on the boundary of a polygon.

Source code in core/src/armnet_core/scene_calibration.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def point_in_polygon(
    point: NormalizedPoint,
    vertices: Iterable[NormalizedPoint],
) -> bool:
    """Return whether a point is inside or on the boundary of a polygon."""
    points = tuple(vertices)
    if len(points) < 3:
        return False
    x, y = point
    inside = False
    for index, start in enumerate(points):
        end = points[(index + 1) % len(points)]
        if _point_on_segment(point, start, end):
            return True
        if (start[1] > y) != (end[1] > y):
            crossing_x = start[0] + (y - start[1]) * (
                end[0] - start[0]
            ) / (end[1] - start[1])
            if x < crossing_x:
                inside = not inside
    return inside

normalized_to_pixel 

normalized_to_pixel(point: NormalizedPoint, *, width: int, height: int) -> NormalizedPoint

Scale a normalized point to source-frame pixel coordinates.

Source code in core/src/armnet_core/scene_calibration.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
def normalized_to_pixel(
    point: NormalizedPoint,
    *,
    width: int,
    height: int,
) -> NormalizedPoint:
    """Scale a normalized point to source-frame pixel coordinates."""
    if width <= 0 or height <= 0:
        raise ValueError("frame width and height must be positive")
    return point[0] * width, point[1] * height

pixel_to_normalized 

pixel_to_normalized(point: NormalizedPoint, *, width: int, height: int) -> NormalizedPoint

Scale source-frame pixel coordinates into the unit frame.

Source code in core/src/armnet_core/scene_calibration.py 
105
106
107
108
109
110
111
112
113
114
def pixel_to_normalized(
    point: NormalizedPoint,
    *,
    width: int,
    height: int,
) -> NormalizedPoint:
    """Scale source-frame pixel coordinates into the unit frame."""
    if width <= 0 or height <= 0:
        raise ValueError("frame width and height must be positive")
    return point[0] / width, point[1] / height

sample_scene_target 

sample_scene_target(calibration: SceneCalibration | None, *, seed: int, reset_index: int, cell_id: str, edge_clearance: float = 0.01) -> NormalizedPoint

Choose a stable reset target inside the active calibrated polygon.

With no active calibration, the whole normalized frame is used. Candidate points are drawn from an ear-clipped triangulation, then the point furthest from an edge is selected. This makes a small edge clearance best-effort without ever moving a point outside a narrow polygon.

Source code in core/src/armnet_core/scene_calibration.py 
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
def sample_scene_target(
    calibration: SceneCalibration | None,
    *,
    seed: int,
    reset_index: int,
    cell_id: str,
    edge_clearance: float = 0.01,
) -> NormalizedPoint:
    """Choose a stable reset target inside the active calibrated polygon.

    With no active calibration, the whole normalized frame is used. Candidate
    points are drawn from an ear-clipped triangulation, then the point furthest
    from an edge is selected. This makes a small edge clearance best-effort
    without ever moving a point outside a narrow polygon.
    """
    if reset_index < 0:
        raise ValueError("reset_index must be non-negative")
    if not cell_id:
        raise ValueError("cell_id must be nonempty")
    if not math.isfinite(edge_clearance) or edge_clearance < 0:
        raise ValueError("edge_clearance must be finite and non-negative")

    vertices = calibration.vertices if calibration else FULL_FRAME_VERTICES
    calibration_id = calibration.id if calibration else "full-normalized-frame"
    material = json.dumps(
        [int(seed), reset_index, cell_id, calibration_id],
        separators=(",", ":"),
        ensure_ascii=True,
    ).encode("utf-8")
    random = _StableRandom(material)
    triangles = _triangulate(vertices)
    weighted = tuple((triangle, polygon_area(triangle)) for triangle in triangles)
    total_area = sum(area for _, area in weighted)

    best: NormalizedPoint | None = None
    best_clearance = -1.0
    # Multiple deterministic candidates make the requested clearance likely
    # for ordinary work areas while preserving support for very narrow shapes.
    for _ in range(64):
        selection = random.unit() * total_area
        triangle = weighted[-1][0]
        for candidate_triangle, area in weighted:
            selection -= area
            if selection <= 0:
                triangle = candidate_triangle
                break
        candidate = _sample_triangle(triangle, random.unit(), random.unit())
        if not point_in_polygon(candidate, vertices):
            continue
        clearance = min(
            _distance_to_segment(
                candidate,
                vertices[index],
                vertices[(index + 1) % len(vertices)],
            )
            for index in range(len(vertices))
        )
        if clearance > best_clearance:
            best = candidate
            best_clearance = clearance
        if best_clearance >= edge_clearance:
            break

    if best is None:  # Defensive fallback for floating-point edge cases.
        best = _triangle_centroid(weighted[0][0])
    if not point_in_polygon(best, vertices):
        raise RuntimeError("failed to sample a point inside the calibration polygon")
    return best