Trazio by CerebrixOS

← All guides

Robotics telemetry

The situation. A night shift of warehouse robots. Robot 17 stops suddenly in the loading bay. You want to know when, why, and what else happened around it, without opening a log viewer.

Set the five values first. This guide covers robot state, faults, battery and motor data, and completed ROS 2 / MCAP recordings. For camera footage see Robot and camera video.

1. What the pipeline needs

Your administrator sets these up. Ask for them if a step below fails.

• Accepted sources matching robot.*, ros2.* (or ros1.*, otel.*) as your robots really emit them.
• artifact_storage.enabled=true, to upload MCAP or bag files.
• telemetry_processing.mode=index_only for topics and structure, or decode for supported typed messages.
• vector_search.enabled=true, only if you want semantic search of recordings.

2. Send a live event

The same emergency stop, three ways.

curl

export ROBOT_RUN_ID="robot-check-$(date +%s)"

curl --fail-with-body -sS -X POST \
  "$PLATFORM_BASE/v1/ingest/$TENANT_ID/$PIPELINE_ID" \
  -H "Authorization: Bearer $TENANT_JWT" \
  -H "Idempotency-Key: $ROBOT_RUN_ID" \
  -H 'Content-Type: application/json' \
  -d "{\"source_type\":\"robot.safety\",\"event_type\":\"emergency_stop\",\"severity\":\"warning\",\"device_id\":\"robot-17\",\"run_id\":\"$ROBOT_RUN_ID\",\"message\":\"Emergency stop on robot 17\"}" \
  | tee ./robot-event.json | jq

Python

import os
from trazio_telemetry import Client

client = Client(
    base_url=os.environ["PLATFORM_BASE"],
    tenant_id=os.environ["TENANT_ID"],
    pipeline_id=os.environ["PIPELINE_ID"],
    tenant_jwt=os.environ["TENANT_JWT"],
)

result = client.record_event({
    "source_type": "robot.safety",
    "event_type": "emergency_stop",
    "severity": "warning",
    "device_id": "robot-17",
    "message": "Emergency stop on robot 17",
}, idempotency_key="robot-17-estop-001")
print(result)

Node.js

import { Client } from "trazio-telemetry";

const client = new Client({
  baseUrl: process.env.PLATFORM_BASE,
  tenantId: process.env.TENANT_ID,
  pipelineId: process.env.PIPELINE_ID,
  tenantJwt: process.env.TENANT_JWT,
});

const result = await client.recordEvent({
  source_type: "robot.safety",
  event_type: "emergency_stop",
  severity: "warning",
  device_id: "robot-17",
  message: "Emergency stop on robot 17",
}, { idempotencyKey: "robot-17-estop-001" });
console.log(result);

Also send a hot motor, which the rules below treat differently. The fields battery_voltage, motor_temperature_c, speed_mps and fault_code are accepted:

curl --fail-with-body -sS -X POST \
  "$PLATFORM_BASE/v1/ingest/$TENANT_ID/$PIPELINE_ID" \
  -H "Authorization: Bearer $TENANT_JWT" \
  -H "Idempotency-Key: robot-17-motor-001" \
  -H 'Content-Type: application/json' \
  -d '{"source_type":"robot.telemetry","event_type":"motor_status","device_id":"robot-17","motor_temperature_c":97,"battery_voltage":23.4,"message":"Motor 2 running hot"}' | jq

Continuous numeric data (battery, joints, diagnostics). Keep your existing OpenTelemetry instrumentation and point its exporter at the platform. The collector config is in Application observability.

3. Upload a recording

Keep your recorder (rosbag2, an MCAP writer, a CAN logger). Upload the completed file.

export ROBOT_FILE='./recording.mcap'

# 1. open an upload session
curl --fail-with-body -sS -X POST \
  "$PLATFORM_BASE/v1/artifacts/$TENANT_ID/$PIPELINE_ID/sessions" \
  -H "Authorization: Bearer $TENANT_JWT" \
  -H 'Content-Type: application/json' \
  -d "{\"format\":\"mcap\",\"filename\":\"$(basename "$ROBOT_FILE")\",\"device_id\":\"robot-17\",\"run_id\":\"$ROBOT_RUN_ID\",\"goal_id\":\"inspect-loading-bay\"}" \
  | tee ./robot-session.json | jq
export SESSION_ID="$(jq -r '.session_id' ./robot-session.json)"

# 2. send the file
curl --fail-with-body -sS -X PUT \
  "$PLATFORM_BASE/v1/artifacts/$TENANT_ID/$PIPELINE_ID/sessions/$SESSION_ID/content" \
  -H "Authorization: Bearer $TENANT_JWT" \
  -H 'Content-Type: application/octet-stream' \
  --data-binary @"$ROBOT_FILE" | jq

# 3. finish
curl --fail-with-body -sS -X POST \
  "$PLATFORM_BASE/v1/artifacts/$TENANT_ID/$PIPELINE_ID/sessions/$SESSION_ID/complete" \
  -H "Authorization: Bearer $TENANT_JWT" | jq

The same upload with the SDKs (reads the whole file into memory, so use it for bounded files):

session = client.upload_artifact("./recording.mcap", format="mcap",
                                 device_id="robot-17", goal_id="inspect-loading-bay")

const session = await client.uploadArtifact("./recording.mcap", {
  format: "mcap", deviceId: "robot-17", goalId: "inspect-loading-bay",
});

For large or unattended transfers, point the platform-edge uploader at your recorder's completed-file folder. It ignores .partial and .active files, resumes interrupted uploads and removes a file only after the server confirms it.

4. Check it landed

A 200 or 202 only means the request was accepted. Always confirm.

sleep 5
curl --fail-with-body -sS \
  "$PLATFORM_BASE/v1/query/$TENANT_ID/$PIPELINE_ID?limit=100&include_ai=true" \
  -H "Authorization: Bearer $TENANT_JWT" \
  | jq '.results[]? | select(.metadata.device_id == "robot-17")'

For a recording, poll until processing finishes:

for attempt in $(seq 1 24); do
  curl --fail-with-body -sS \
    "$PLATFORM_BASE/v1/artifacts/$TENANT_ID/$PIPELINE_ID/sessions/$SESSION_ID" \
    -H "Authorization: Bearer $TENANT_JWT" | tee ./robot-session-status.json >/dev/null
  jq '{status,index_status,index_error,telemetry_processing}' ./robot-session-status.json
  test "$(jq -r '.index_status // "pending"' ./robot-session-status.json)" = complete && break
  sleep 5
done

Success means the original file is complete and readable and indexing reached the state you asked for. It does not mean every byte was decoded: check decoded_messages, skipped_messages and decoder_errors.

5. Rules that fire

The standard physical-ai pack watches robot.* and vehicle.* events.

Both events above trigger one. Each result appears on the event as rule_evaluation. To add your own rule, validate and dry-run it before publishing:

client.validate_rule_pack(pack)
client.dry_run_rule_pack(pack, [{"source_type": "robot.telemetry", "motor_temperature_c": 97}])

6. Ask about it

Assistant. Connect https://ca.mcp.trazio.cerebrixos.com/mcp (how) and ask in plain language:

• "What happened with robot 17?"
• "Show critical safety events from the last hour."
• "Which robots had a motor above 90 degrees?"

Sample answer. Searching for "robot 17 emergency stop" in the demo pipeline returned, as the top match:

{
  "score": 1,
  "event": {
    "source_type": "robot.safety",
    "event_type": "emergency_stop",
    "severity": "critical",
    "message": "Emergency stop triggered on robot 17",
    "device_id": "robot-17",
    "rule_evaluation": { "severity": "critical", "matched_rules": [{ "name": "Emergency stop", "severity": "critical" }] }
  }
}

The next two results were the person detected 1.2 m from robot 17 beside a forklift lane (high) and the mission resuming after the safety check, so the three events tell the story in order.

Python.

client.semantic_query("Show critical robot faults from the last 2 hours")
client.search_events("robot 17 emergency stop")

curl. GET /v1/query/$TENANT_ID/$PIPELINE_ID?limit=100 as in step 4, filtered with jq.

To see how events connect (same run, same device), ask for the evidence graph around one event:

curl --fail-with-body -sS \
  "$PLATFORM_BASE/v1/query/$TENANT_ID/$PIPELINE_ID/graph?event_id=$EVENT_ID&hops=2&limit=100" \
  -H "Authorization: Bearer $TENANT_JWT" | jq

An edge shows recorded evidence and its confidence. It does not prove causation.

Troubleshooting