Real Robot Backend¶
Control the physical PIB robot via rosbridge websocket connection.
Overview¶
RealRobotBackend connects to a PIB robot's ROS system via rosbridge and provides joint control and trajectory execution.
RealRobotBackend Class¶
RealRobotBackend
¶
RealRobotBackend(
host="172.26.34.149",
port=9090,
timeout=5.0,
motor_mode="direct",
estop_keys=True,
stop_button=True,
ai_owner=None,
)
Bases: RobotBackend
Control the real PIB robot.
By default, motor commands are sent directly to Tinkerforge servo bricklets for low latency (~5-20 ms). Servo bricklets are auto-discovered on connect — no manual configuration needed.
ROS/rosbridge is still connected for audio, camera, and AI
subsystems. To use ROS for motor control instead, pass
motor_mode="ros".
Note
Uses joint_limits_robot.yaml for percentage <-> radians conversion. Calibrate with: python -m pib3.tools.calibrate_joints
Example
from pib3 import Robot, Joint with Robot(host="172.26.34.149") as robot: ... # Motor commands go directly to Tinkerforge (default) ... robot.set_joint(Joint.ELBOW_LEFT, 50.0) ... angle = robot.get_joint(Joint.ELBOW_LEFT) ... ... # Save and restore pose ... saved_pose = robot.get_joints() ... robot.set_joints(saved_pose)
Use ROS for motor control instead:¶
with Robot(host="172.26.34.149", motor_mode="ros") as robot: ... robot.set_joint(Joint.ELBOW_LEFT, 50.0)
Initialize real robot backend.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
host
|
str
|
Robot IP address. |
'172.26.34.149'
|
port
|
int
|
Rosbridge websocket port. |
9090
|
timeout
|
float
|
Connection timeout in seconds. |
5.0
|
motor_mode
|
str
|
Motor control mode (default: |
'direct'
|
estop_keys
|
Union[bool, str, Sequence[str]]
|
Emergency-stop keys armed on connect. True (default) = Space, Esc, Numpad-0 and Pause; a key name or list for your own; False for none. Ctrl+C always stops the robot. |
True
|
stop_button
|
Union[bool, str]
|
Show the on-screen STOP button while the stop is
armed. True (default): it opens with the first motion command
and is the visible sign that the stop is armed; |
True
|
ai_owner
|
Optional[str]
|
Name this client uses when it starts AI models. A model
runs as long as any owner holds it, so two clients with the
same name share (and stop) each other's models. Default:
|
None
|
Quick Start¶
from pib3 import Robot, Joint
with Robot(host="172.26.34.149") as robot:
robot.set_joint(Joint.ELBOW_LEFT, 50.0) # IDE tab completion
pos = robot.get_joint(Joint.ELBOW_LEFT)
robot.run_trajectory("trajectory.json")
Constructor¶
Robot(
host: str = "172.26.34.149",
port: int = 9090,
timeout: float = 5.0,
motor_mode: str = "direct",
estop_keys=True,
stop_button="auto",
)
Parameters:
| Parameter | Type | Default | Description |
|---|---|---|---|
host |
str |
"172.26.34.149" |
IP address of the robot. |
port |
int |
9090 |
Rosbridge websocket port. |
timeout |
float |
5.0 |
Connection timeout in seconds. |
motor_mode |
str |
"direct" |
Motor control mode: "direct" (Tinkerforge) or "ros" (via rosbridge). Anything else raises ValueError. |
estop_keys |
bool, str or list |
True |
Emergency-stop keys, armed by the program's first motion command: True = Space, Esc, Numpad-0, Pause; a name or list for others; False for none. Ctrl+C stops too. A stop latches every pib3 program on this robot. See Safety. |
stop_button |
bool or "auto" |
True |
On-screen STOP window while the stop is armed (the visible "armed" sign, lists the working triggers); "auto" only where the keys cannot work; False never. |
Example:
from pib3 import Robot
# Default connection (direct Tinkerforge control, auto-discovered)
robot = Robot()
# Custom host
robot = Robot(host="192.168.1.100")
# All parameters
robot = Robot(
host="192.168.1.100",
port=9090,
timeout=10.0,
)
# Use ROS for motor control instead of Tinkerforge
robot = Robot(host="192.168.1.100", motor_mode="ros")
ROS motor mode
The robot's motor_control node expects one trajectory point per
joint. pib3 0.2 sends all joints of a command in one such request
(earlier versions moved only the first joint of a trajectory waypoint).
The node ignores the velocity, so speed= has no effect in ROS mode;
joints move with the speed set in Cerebra.
Subsystem Properties¶
The robot provides high-level subsystem APIs for simplified access to AI and camera features.
robot.ai¶
Access the AI subsystem for AI inference with auto-managed subscriptions.
with Robot(host="172.26.34.149") as robot:
robot.ai.set_model(AIModel.HAND)
for hand in robot.ai.get_hand_landmarks():
print(f"{hand.handedness}: index={hand.finger_angles.index:.0f}°")
robot.camera¶
Access the camera subsystem for camera streaming with auto-managed subscriptions.
with Robot(host="172.26.34.149") as robot:
frame = robot.camera.get_frame()
if frame:
img = frame.to_numpy() # Requires OpenCV
Subsystems vs Raw Subscriptions
For most use cases, use robot.ai and robot.camera instead of the low-level
subscribe_ai_detections() and subscribe_camera_image() methods. The subsystems
automatically manage subscription lifecycle and provide typed return values.
Connection¶
connect()¶
Establish websocket connection to the robot's rosbridge server.
Raises:
ConnectionError: If unable to connect within timeout
Example:
from pib3 import Robot
robot = Robot(host="172.26.34.149")
try:
robot.connect()
print(f"Connected: {robot.is_connected}")
# ... use robot ...
finally:
robot.disconnect()
Using Context Manager¶
from pib3 import Robot, Joint
with Robot(host="172.26.34.149") as robot:
robot.set_joint(Joint.ELBOW_LEFT, 50.0)
# Automatically disconnected
Joint Control¶
The robot backend inherits all methods from RobotBackend. Key methods:
set_joint()¶
Set a single joint position.
def set_joint(
self,
motor_name: Union[str, Joint],
position: float,
unit: Literal["percent", "rad", "deg"] = "percent",
async_: bool = False,
timeout: float = 2.0,
tolerance: Optional[float] = None,
speed: Optional[float] = None,
) -> bool
| Parameter | Type | Default | Description |
|---|---|---|---|
motor_name |
str or Joint |
required | Motor name or Joint enum (e.g., Joint.ELBOW_LEFT). |
position |
float |
required | Target position in specified unit. |
unit |
"percent", "rad", "deg" |
"percent" |
Position unit. |
async_ |
bool |
False |
If False (default), poll until joint reaches target. |
timeout |
float |
2.0 |
Max wait time when async_=False. |
tolerance |
float |
None |
Acceptable error (default: 2%, 3°, or 0.05 rad). |
speed |
float or None |
None |
Movement speed in deg/s. None uses the servo channel's current motion config (default ≈ 150°/s on the direct path — see DEFAULT_MOTION_VELOCITY). |
Example:
from pib3 import Robot, Joint
with Robot(host="172.26.34.149") as robot:
robot.set_joint(Joint.TURN_HEAD, 50.0) # Percentage
robot.set_joint(Joint.ELBOW_LEFT, 1.25, unit="rad") # Radians
robot.set_joint(Joint.ELBOW_LEFT, -30.0, unit="deg") # Degrees
# Fire and forget (default waits for completion)
robot.set_joint(Joint.ELBOW_LEFT, 50.0, async_=True)
set_joints()¶
Set multiple joint positions simultaneously.
def set_joints(
self,
positions: Dict[Union[str, Joint], float],
unit: Literal["percent", "rad", "deg"] = "percent",
async_: bool = False,
timeout: float = 2.0,
tolerance: Optional[float] = None,
speed: Optional[float] = None,
) -> bool
| Parameter | Type | Default | Description |
|---|---|---|---|
positions |
Dict[str\|Joint, float] |
required | Target positions. A HandPose or sequence raises TypeError — use set_joints_pose() / set_joints_sequence() instead. |
unit |
"percent", "rad", "deg" |
"percent" |
Position unit. |
async_ |
bool |
False |
If False (default), poll until joints reach targets. |
timeout |
float |
2.0 |
Max wait time when async_=False. |
tolerance |
float |
None |
Acceptable error. |
speed |
float or None |
None |
Movement speed in deg/s. On the direct path this rewrites the servo's shared motion config — see base class docs. |
Example:
from pib3 import Robot, Joint
with Robot(host="172.26.34.149") as robot:
robot.set_joints({
Joint.SHOULDER_VERTICAL_LEFT: 30.0,
Joint.ELBOW_LEFT: 60.0,
})
get_joint()¶
Read a single joint position.
def get_joint(
self,
motor_name: Union[str, Joint],
unit: Literal["percent", "rad", "deg"] = "percent",
timeout: Optional[float] = None,
) -> Optional[float]
| Parameter | Type | Default | Description |
|---|---|---|---|
motor_name |
str or Joint |
required | Motor name or Joint enum. |
unit |
"percent", "rad", "deg" |
"percent" |
Return unit. |
timeout |
float |
5.0 |
Max wait time for ROS messages (seconds). |
Returns: float or None - Current position.
from pib3 import Robot, Joint
with Robot(host="172.26.34.149") as robot:
pos = robot.get_joint(Joint.ELBOW_LEFT)
pos_rad = robot.get_joint(Joint.ELBOW_LEFT, unit="rad")
get_joints()¶
Read multiple joint positions.
def get_joints(
self,
motor_names: Optional[List[Union[str, Joint]]] = None,
unit: Literal["percent", "rad", "deg"] = "percent",
timeout: Optional[float] = None,
) -> Dict[str, float]
| Parameter | Type | Default | Description |
|---|---|---|---|
motor_names |
List[str\|Joint] or None |
None |
Motors to query. None returns all. |
unit |
"percent", "rad", "deg" |
"percent" |
Return unit. |
timeout |
float |
5.0 |
Max wait time for ROS messages (seconds). |
Returns: Dict[str, float] - Motor names mapped to positions.
from pib3 import Robot, Joint
with Robot(host="172.26.34.149") as robot:
arm = robot.get_joints([Joint.ELBOW_LEFT, Joint.WRIST_LEFT])
all_joints = robot.get_joints()
Trajectory Execution¶
run_trajectory()¶
def run_trajectory(
self,
trajectory: Union[str, Path, Trajectory],
rate_hz: float = 20.0,
progress_callback: Optional[Callable[[int, int], None]] = None,
) -> bool
| Parameter | Type | Default | Description |
|---|---|---|---|
trajectory |
str, Path, Trajectory |
required | Trajectory file or object. |
rate_hz |
float |
20.0 |
Waypoints per second. |
progress_callback |
Callable[[int, int], None] |
None |
Progress callback (current, total). |
from pib3 import Robot, Trajectory
with Robot(host="172.26.34.149") as robot:
robot.run_trajectory("trajectory.json")
# With progress
robot.run_trajectory(
trajectory,
rate_hz=20.0,
progress_callback=lambda c, t: print(f"\r{c}/{t}", end=""),
)
Save and Restore Poses¶
import json
from pib3 import Robot
with Robot(host="172.26.34.149") as robot:
pose = robot.get_joints() # Save
with open("pose.json", "w") as f:
json.dump(pose, f)
with open("pose.json") as f:
robot.set_joints(json.load(f)) # Restore
Camera Streaming¶
subscribe_camera_image()¶
Subscribe to camera images (raw JPEG bytes). The camera node publishes them as base64 text on /camera_topic and only encodes frames while someone is subscribed. The frame is 1280×720 by default.
Call .unsubscribe() on the returned topic to stop streaming.
import cv2, numpy as np
def on_frame(jpeg_bytes):
frame = cv2.imdecode(np.frombuffer(jpeg_bytes, np.uint8), cv2.IMREAD_COLOR)
cv2.imshow("Camera", frame)
cv2.waitKey(1)
with Robot(host="192.168.178.71") as robot:
sub = robot.subscribe_camera_image(on_frame)
time.sleep(10)
sub.unsubscribe()
set_camera_config()¶
def set_camera_config(
fps: Optional[int] = None,
quality: Optional[int] = None, # JPEG quality 1-100
resolution: Optional[tuple] = None, # (width, height)
) -> None
Each setting goes to its own topic of the camera node (quality_factor_topic, timer_period_topic, size_topic); there are matching set_camera_quality(), set_camera_timer_period() and set_camera_preview_size().
Note
Changing the resolution restarts the camera pipeline. Keep it 16:9: the AI models read the same frame, and the camera refuses to feed them one of another aspect.
AI Detection¶
The camera runs the models of the robot's model store. A client asks for a model with /start_model and names itself as an owner; the model runs while any owner holds it. The robot publishes each model's results as datatypes/DetectionArray on detections/<model>. robot.ai wraps all of this; the methods below are the layer underneath.
ai_owner¶
The name this client uses for /start_model and /stop_model: pib3-<user>@<computer> unless set with Robot(ai_owner="group-3"). Clients with the same name share, and stop, each other's models. The default is stable on purpose: a script that crashed leaves its model held under that name, and the next run releases it.
get_available_ai_models()¶
models = robot.get_available_ai_models() # dict: model id -> info
# {"yolo26s_coco_512x288": {"task": "object_detection", "licence": "...",
# "shaves": 4, "size_bytes": 5472216,
# "available": True, "active": False}, ...}
Calls /list_models. Only models with available: True can be started. Empty if the service does not answer.
start_ai_model() / stop_ai_model()¶
def start_ai_model(model, shaves: int = 0, timeout: float = 90.0) -> Tuple[bool, str]
def stop_ai_model(model, timeout: float = 90.0) -> Tuple[bool, str]
Call /start_model and /stop_model as ai_owner and return (success, message). The message says why a start failed ("Unknown model", "Model is unavailable", ...). shaves=0 takes the number the model was compiled for.
ok, message = robot.start_ai_model(AIModel.HAND)
if not ok:
print(message)
robot.stop_ai_model(AIModel.HAND)
Note
A start rebuilds the camera pipeline, which takes longer than the 5 s rosbridge waits by default, so pib3 asks rosbridge to wait timeout seconds (90 by default). When the call still gets no answer, start_ai_model asks /list_models whether the model runs and counts it as started if it does. A stop that gets no answer is reported as failed although the robot may still stop the model; /models_status shows the truth.
set_ai_model()¶
Same as robot.ai.set_model(model): run this model and none of the others this client started. Each start or stop rebuilds the camera pipeline.
subscribe_ai_detections()¶
Subscribes to the model's topic. The model must be running; the topic is silent until then. The callback receives one DetectionArray per frame:
{"header": {"stamp": {"sec": 1790000000, "nanosec": 123000000}},
"model_id": "yolo26s_coco_512x288",
"frame_width": 1280, "frame_height": 720,
"detections": [{"label": "person", "score": 0.92,
"x_min": 353, "y_min": 257, "x_max": 547, "y_max": 570, # pixels
"keypoint_names": [], "keypoint_x": [], "keypoint_y": [], "keypoint_z": [],
"keypoint_score": [], "scalar_names": [], "scalar_values": []}]}
Pixels refer to frame_width × frame_height. keypoint_z is millimetres, 0 meaning none. keypoint_score is each keypoint's confidence 0..1, empty when the model reports none (and absent on backends without it). The hand models publish their hand-relative depth unitless instead (see z_source in scalar_names).
def on_detection(message):
for det in message['detections']:
print(f"{det['label']} @ {det['score']:.0%}")
sub = robot.subscribe_ai_detections(AIModel.YOLO26S, on_detection)
time.sleep(10)
sub.unsubscribe()
For typed results use AIDetectionReceiver or parse_detection_message() from pib3.backends.
get_ai_detections()¶
Calls /get_detections once. frame_width is 0 while the camera has no frame yet.
subscribe_ai_status()¶
The callback receives {"models": [{"model_id", "state", "message", "active", "fps", "shaves"}, ...]}; state is idle, starting, running or failed, and message says why a model failed. fps is averaged over at least ten seconds.
IMU Sensor¶
The BMI270 IMU of the OAK-D Lite streams at a fixed 100 Hz on /imu (sensor_msgs/Imu) whenever the camera node runs. Axes follow ROS (REP-103): x forward, y left, z up, so a robot at rest reads about +9.8 m/s² on z. The BMI270 cannot fuse an orientation: orientation stays the identity and orientation_covariance[0] is -1.
subscribe_imu()¶
def subscribe_imu(
callback: Callable[[dict], None],
data_type: str = "full", # "full", "accelerometer", or "gyroscope"
) -> roslibpy.Topic
data_type |
Callback receives |
|---|---|
"full" |
the whole message: header, linear_acceleration, angular_velocity, orientation (identity), covariances |
"accelerometer" |
{"header", "vector": {x,y,z}} |
"gyroscope" |
{"header", "vector": {x,y,z}} |
def on_imu(data):
accel = data['linear_acceleration']
print(f"Accel: {accel['x']:.2f}, {accel['y']:.2f}, {accel['z']:.2f}")
sub = robot.subscribe_imu(on_imu)
time.sleep(5)
sub.unsubscribe()
subscribe_imu_raw()¶
set_imu_frequency()¶
Raises NotImplementedError: the camera node publishes at a fixed 100 Hz. Skip samples in your callback for a lower rate.
Helper Functions¶
rle_decode()¶
Decode RLE-encoded segmentation masks (det.mask_rle). The robot has no segmentation model; the simulation produces them for "segmentation".
from pib3.backends import rle_decode
mask = rle_decode(det.mask_rle) # Returns np.ndarray (height, width)
Text-to-Speech¶
RealRobotBackend.speak() overrides the base implementation and prefers the robot's on-board TTS service /play_audio_from_speech — speech is synthesized directly on the robot, avoiding Piper inference on the client and streaming audio over rosbridge. Falls back to base-class Piper synthesis for AudioOutput.LOCAL or when the robot service is unavailable.
def speak(
text: str,
output: Optional[AudioOutput] = None,
voice: Optional[str] = None, # Piper voice (local fallback only)
block: bool = True,
use_robot_tts: Optional[bool] = None, # None = auto (prefer robot when output includes ROBOT)
language: str = "de", # Robot TTS language code
) -> bool
See play_audio_from_speech() for the underlying ROS service (supports wait=False for fire-and-forget).
Low-Latency Mode¶
Bypass ROS for direct Tinkerforge motor control with ~5-20ms latency (vs ~100-200ms via ROS). When enabled, both get_joint()/get_joints() and set_joint()/set_joints() use direct Tinkerforge communication.
Quick Setup¶
Direct Tinkerforge control is the default mode. On connect pib3 reads the
robot's own motor table from pib-api (http://<host>:5000/motor, the table
Cerebra edits). It gives the exact bricklet UID and pin of every motor, plus
each motor's invert flag and rotation range, so direct control moves every
joint exactly like the ROS path. Without pib-api, pib3 falls back to
enumerating the servo bricklets (LowLatencyConfig.use_robot_motor_config).
What get_joint() reads
pib's hobby servos report nothing back. In direct mode get_joint()
returns the position of the bricklet's motion ramp, i.e. where the servo
is being told to be right now. A blocked joint therefore still reads
as "arrived".
from pib3 import Robot
with Robot(host="172.26.34.149") as robot:
robot.set_joint("elbow_left", 0.5, unit="rad") # Direct write
pos = robot.get_joint("elbow_left", unit="rad") # Direct read
# To use ROS for motor control instead:
with Robot(host="172.26.34.149", motor_mode="ros") as robot:
robot.set_joint("elbow_left", 0.5, unit="rad") # Via ROS
Properties¶
| Property | Type | Description |
|---|---|---|
low_latency_available |
bool |
True if connected and configured |
low_latency_enabled |
bool |
Get/set enabled state. The setter raises RuntimeError when the backend is not connected or when Tinkerforge discovery fails (the flag is rolled back). |
low_latency_sync_to_ros |
bool |
Get/set cache sync setting |
discovered_servo_uids |
List[str] |
UIDs from last discovery |
Default motion config¶
Every motion command sets its velocity: speed=..., else robot.default_speed
(150 deg/s). Acceleration and deceleration come from these constants; change
them with configure_all_servo_channels().
| Constant | Default | Unit | Notes |
|---|---|---|---|
RealRobotBackend.DEFAULT_MOTION_VELOCITY |
15000 |
0.01 °/s | ≈ 150°/s |
RealRobotBackend.DEFAULT_MOTION_ACCELERATION |
15000 |
0.01 °/s² | |
RealRobotBackend.DEFAULT_MOTION_DECELERATION |
15000 |
0.01 °/s² |
Methods¶
discover_servo_bricklets()¶
Returns list of Servo Bricklet V2 UIDs.
configure_motor_mapping()¶
def configure_motor_mapping(
mapping: Dict[str, Tuple[str, int]],
reinitialize: bool = True,
) -> None
Configure motor-to-bricklet mapping at runtime. Auto-configures servo channels.
configure_servo_channel()¶
def configure_servo_channel(
motor_name: str,
pulse_width_min: int = 700,
pulse_width_max: int = 2500,
velocity: int = 9000,
acceleration: int = 9000,
deceleration: int = 9000,
) -> bool
Configure individual servo channel settings.
Helper Functions¶
from pib3 import build_motor_mapping, PIB_SERVO_CHANNELS
# Build complete mapping from 3 UIDs
mapping = build_motor_mapping("UID1", "UID2", "UID3")
# Reference standard channel assignments: (bricklet_number, channel)
print(PIB_SERVO_CHANNELS["elbow_left"]) # (3, 8)
See the Low-Latency Tutorial for complete examples.
ROS Integration¶
| Topic/Service | Purpose |
|---|---|
/joint_trajectory |
Joint position feedback |
/apply_joint_trajectory |
Motor commands |
Commands use ROS2 JointTrajectory format with positions in centidegrees. Every topic and service pib3 uses, with message layouts, is in Robot ROS Interface.
Troubleshooting¶
| Problem | Cause | Solution |
|---|---|---|
| Connection refused | Rosbridge not running | ros2 launch rosbridge_server rosbridge_websocket_launch.xml |
| Connection timeout | Wrong IP or network issue | ping <ip>, increase timeout parameter |
| Joint not moving | Limits not calibrated | Calibrate or use unit="rad" |
get_joint returns None |
ROS message timeout | Increase timeout, verify the /joint_trajectory topic |