Device interfaces#
Public imports are lazy with respect to hardware. Constructing a selected backend may connect a device; importing these contracts does not.
Interface |
Selector |
Implementation |
|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
Robot server |
|
Backend |
Only the camera selector has a fixed backend-name list. Robot/gripper type names must match lowercase package-name rules. Missing implementations, dependencies, and incomplete abstract contracts produce explicit errors. They do not silently fall back to another device.
Camera#
"""Minimal camera contract, independent of ROS and camera hardware."""
from abc import ABC, abstractmethod
class BaseCamera(ABC):
"""Synchronous camera operations used by agents and robot utilities.
Implementations retain their existing image formats, point-cloud units and
ordering. This interface deliberately does not convert or normalize data.
"""
@abstractmethod
def get_rgb_image(self, use_new_frame: bool = True):
"""Return an RGB NumPy array, or a copy of the cached frame if requested."""
raise NotImplementedError
@abstractmethod
def get_point_cloud(self, save_points: bool = False):
"""Return XYZ data, or None for an empty cloud.
RGB-only backends raise RuntimeError. Supported backends retain their
existing point-cloud representation and optional file-saving behavior.
"""
raise NotImplementedError
def close(self):
"""Release backend-owned resources, never a shared ROS runtime.
Backends without owned resources inherit this no-op.
"""
The file backend supplies an RGB image but no point cloud. ROS backends own their subscriptions and caching resources. See adding a camera for the additional capture call sites required by a new backend.
Robot state and helpers#
"""Robot-independent contracts. Importing them never connects to hardware."""
from abc import ABC, abstractmethod
class RobotStateError(RuntimeError):
"""Raised when the robot state stream cannot provide a fresh TCP pose."""
class BaseRobotState(ABC):
"""Read robot state without commanding motion or applying tool geometry.
Readers must use the robot-base coordinates used by GTA's world-frame
keypoints. The pose origin/orientation must match the frame in which the
deployment defines its local tool keypoints and axes. The UR5e reader
preserves the configured TCP pose returned by RTDE; it does not substitute
a flange pose, set the controller TCP, or apply GTA's additional tool offset.
A backend owns its receive connection and its connection-sharing policy.
Importing a backend must not connect to hardware.
"""
@abstractmethod
def get_tcp_pose(self):
"""Return six finite floats as a NumPy array: [x, y, z, rx, ry, rz].
Position is in metres; orientation is a rotation vector in radians,
not Euler angles. Raise RobotStateError if usable state is unavailable.
"""
raise NotImplementedError
@abstractmethod
def close(self):
"""Release this reader's resources, not motion-control or shared ROS resources."""
raise NotImplementedError
class BaseRobotHelpers(ABC):
"""Supporting robot operations, independent of GUI events and grippers."""
@abstractmethod
def reset_joints(self, *, check_cancel, on_started, errors, wait):
"""Move to this deployment's reviewed reset pose.
Check cancellation before connecting and throughout motion. Call
on_started only when ready to move. Always stop/release owned hardware,
appending cleanup failures to errors even when the operation raises.
wait(seconds) is an interruptible pause supplied by the caller.
"""
raise NotImplementedError
The UR5e reader shares a receive connection per process and checks freshness.
Do not assume close() on one shared reader leaves another reference
connected. No tool offset or physical TCP change is performed by the reader.
Helpers receive cancellation/start/wait callbacks and an error list. They must clean up on both success and exception; they do not own GUI event names. See adding a robot.
Gripper#
"""Gripper contract. ROS lifecycle belongs to the caller, not the device."""
from abc import ABC, abstractmethod
class BaseGripper(ABC):
"""Operations may raise; cancellation/cleanup return lists of error strings.
Implementations own only their action/SDK client and its outstanding goals.
They must not initialize ROS, subscribe to GTA commands, or connect the arm.
Shutdown releases client resources; close physically closes the jaws.
"""
@abstractmethod
def wait_for_server(self, timeout_s=None):
raise NotImplementedError
@abstractmethod
def open(self, timeout_s=None, **kwargs):
raise NotImplementedError
@abstractmethod
def close(self, timeout_s=None, **kwargs):
raise NotImplementedError
def half_open(self, timeout_s=None, **kwargs):
raise NotImplementedError('This gripper does not support half_open')
@abstractmethod
def cancel(self, *, ack_timeout_s=0.5, timeout_s=0.5):
"""Cancel this client's outstanding goal, ignoring operation cancellation."""
raise NotImplementedError
@abstractmethod
def shutdown(self):
"""Release the client only, after cancel(); return cleanup errors."""
raise NotImplementedError
ROS 2 callers supply a node and future-spinning callback. Optional wait-policy
arguments are implementation-specific; the current Robotiq client accepts
check_cancel, poll_interval_s, and cancel_on_timeout.
ROS 1 operations support caller-selected timeout and success-check policies.
The client owns goals, not the shared ROS lifecycle.
Launcher boundaries#
python scene/robot/server.py or python -m scene.robot.server launches
the selected server. This can initialize physical hardware.
--help parses without loading hardware implementations.
--operation reset_joints, open_gripper, and close_gripper are
supervisor-only launch modes. They require the existing control channel and
child role; they are not general terminal commands. Reset dispatches to the
robot helper, while gripper operations use a gripper-only ROS node and no RTDE
connection. See execution for supported operator controls.