RoboTrace SDK

OTel-native Python SDK for robot observability. Mission tracing, sensor telemetry, evaluation scoring -- think Langfuse, but for robots.

Install

pip install robotrace-sdk

Optional extras:

pip install robotrace-sdk[mcap]      # MCAP recording

Quick Start

from robotrace import RoboTrace, Scalar, Pose3D, Battery

rt = RoboTrace(
    host="https://your-robotrace-server.com",
    public_key="rt-pk-xxx",
    secret_key="rt-sk-xxx",
    device_id="amr-001",
)

# Log sensor telemetry
rt.log("robot/pose", Pose3D(x=1.0, y=2.0, z=0.0))
rt.log("sensors/battery", Battery(voltage=12.6, soc=0.875, current=-2.1))

# Track missions
with rt.mission("deliver_pallet", tags=["zone-B"]) as m:
    with m.phase("navigate"):
        rt.log("robot/pose", Pose3D(x=5.0, y=3.0, z=0.0))
    m.score("success", True)

rt.shutdown()

ROS2 Integration

Connect to any ROS2 robot with zero code changes (bridge) or 3 lines (converters):

Converters (in your code)

from robotrace.ros2 import from_odometry, from_laser_scan, from_battery_state

# In your ROS2 callback:
rt.log("robot/pose", from_odometry(odom_msg))
rt.log("sensors/lidar", from_laser_scan(scan_msg))
rt.log("sensors/battery", from_battery_state(battery_msg))

Bridge (zero code changes)

# bridge.yaml
robotrace:
  host: https://your-server.com
  public_key: rt-pk-xxx
  secret_key: rt-sk-xxx
  device_id: my-robot
topics:
  /scan: {stream: sensors/lidar, type: LaserScan}
  /odom: {stream: robot/pose, type: Odometry}
  /battery_state: {stream: sensors/battery, type: BatteryState}

python -m robotrace.ros2 --config bridge.yaml

16 converters included: LaserScan, Imu, BatteryState, Odometry, Twist, PoseStamped, NavSatFix, JointState, Image, CompressedImage, PointCloud2, Path, DiagnosticArray, Temperature, Range.

Stream Configuration

Tell the platform what your data means (units, ranges, thresholds):

rt.configure_stream("sensors/battery", {
    "display_name": "Main Battery",
    "data_type": "battery",
    "fields": {
        "soc": {"unit": "ratio", "scale": 100, "display_unit": "%"},
        "voltage": {"unit": "V", "range": [10.0, 14.8]},
        "temperature": {"unit": "C", "warning_above": 60},
    },
    "rate_hz": 1,
})

Decorator Pattern

from robotrace import mission, phase

@mission(name="deliver_pallet")
def deliver(destination):
    navigate_to(destination)
    pick_up()
    return {"status": "delivered"}

@phase()
def navigate_to(destination):
    rt.log("robot/pose", Pose3D(x=1.0, y=2.0, z=0.0))

# Works with async too
@mission()
async def async_delivery(destination):
    await navigate_async(destination)

Context Manager Pattern

with rt.mission("pick_and_place", tags=["priority-high"]) as m:
    with m.phase("navigate") as nav:
        nav.log("robot/pose", Pose3D(x=1.0, y=2.0, z=0.0))
        nav.event("obstacle_detected", {"distance_m": 0.5})
    with m.phase("pick") as pick:
        pick.decision("grasp_planner", model="graspnet-v2",
                      input={"target": [0, 0, 0]}, output={"confidence": 0.95})
    m.score("cycle_time_s", 14.2)
    m.score("success", True)

Data Types

20 typed sensor data types:

Type	Description
Scalar(value)	Single numeric value
Vector3(x, y, z)	3D vector
VectorN(values)	N-dimensional vector
Pose3D(x, y, z, qx, qy, qz, qw)	6DOF pose with quaternion
Transform3D(translation, rotation)	Coordinate transform
PointCloud(points, colors)	Nx3 point cloud
Image(data, format, meta)	Compressed image (JPEG/PNG); meta travels with it — a geotag places it on the map (below)
DepthImage(data)	Depth map
Video(data, format)	Video clip
LaserScan(ranges, angle_min, angle_max, angle_increment)	2D LiDAR scan
JointState(names, positions, velocities, efforts)	Multi-DOF joint state
NumericSet(values)	Named numeric values
GeoLocation(latitude, longitude, altitude, heading, speed)	GPS/GNSS position
Path(points, frame_id)	Trajectory waypoints
Twist(linear_x/y/z, angular_x/y/z)	6DOF velocity
BoundingBox2D(x, y, w, h, label)	2D detection
BoundingBox3D(center, size, label)	3D detection
Log(message, level)	Text log message
Battery(voltage, current, soc, health, temperature)	Battery state
Bitset(value, width, labels)	Hardware status register

Footage: geotagged images and recordings

Images are uploaded to object storage; their telemetry row keeps the capture time and the meta you give. With a geotag, RoboTrace's operation Footage tab draws the ground each image shows and adds up the area covered:

rt.log("sensors/camera", Image(jpeg, meta={
    "pose": {"x": 3.9, "y": 7.4, "z": 8.0, "yaw_deg": 1.6, "frame": "map"},   # yaw from +x, counter-clockwise
    "geo": {"lat": 47.3980, "lon": 8.5462, "alt": 8.0},
    "camera": {"hfov_deg": 99.7, "vfov_deg": 83.3, "pitch_deg": 60, "width": 640, "height": 480},  # pitch: below the horizon
}))

VideoRecorder records a clip per task (mission_id=); it is stamped with when it began, so it lands inside its mission. rt.log(..., timestamp=) does the same for anything logged after the fact.

Fleet Gateway

Manage telemetry for multiple robots from one process:

from robotrace import RoboTraceFleet

fleet = RoboTraceFleet(
    host="https://your-server.com",
    public_key="rt-pk-xxx",
    secret_key="rt-sk-xxx",
    device_ids=["amr-001", "amr-002", "amr-003"],
)

with fleet.mission("amr-001", "deliver_pallet") as m:
    fleet.log("amr-001", "sensors/battery", Battery(voltage=12.4))
    m.score("success", True)

fleet.broadcast_log("fleet/heartbeat", Scalar(1.0))
fleet.shutdown_all()

Commands

RoboTrace records the commands your robots execute (with m.phase("movej", command="movej", input={...})) but doesn't send commands to robots; commanding a fleet is RoboSwarm's job. rt.command_channel() and the [commands] extra are deprecated and will be removed in 3.0.

MCAP Recording

Dual-output: telemetry goes to both the server AND a local MCAP file (requires pip install robotrace-sdk[mcap]):

rt.start_recording("mission.mcap")
# All subsequent log() calls also write to the MCAP file
rt.stop_recording()

Pipeline Health

health = rt.health()
# {'buffered_count': 5, 'offline_count': 0, 'stats': {...}, 'is_recording': False}

rt = RoboTrace(..., on_error=lambda ctx, exc: print(f"Error in {ctx}: {exc}"))

Environment Variables

Variable	Default	Description
ROBOTRACE_HOST	(none)	Server URL (fallback if host not set)
ROBOTRACE_DEBUG	false	Enable debug logging to stderr
ROBOTRACE_LOG_LEVEL	WARNING	Log level: DEBUG, INFO, WARNING, ERROR
ROBOTRACE_LOG_FILE	(none)	Write SDK logs to this file
ROBOTRACE_DATA_DIR	~/.robotrace	Directory for offline queue database
ROBOTRACE_OFFLINE_MAX_MB	100	Max offline queue size in MB
ROBOTRACE_OFFLINE_MAX_HOURS	72	Max offline queue age in hours

Development

pip install -e ".[dev]"
pytest                        # 242 tests
pytest tests/test_ros2_converters.py  # ROS2 converter tests only

License

MIT