feat(teleop): add PICO WebXR body tracking

dimos/robot/all_blueprints.py

@@ -52,6 +52,7 @@
     "demo-mcp-stress-test": "dimos.core.demos.stress_test_blueprint:demo_mcp_stress_test",
     "demo-object-scene-registration": "dimos.perception.experimental.demo_object_scene_registration:demo_object_scene_registration",
     "demo-osm": "dimos.mapping.osm.demo_osm:demo_osm",
+    "demo-pico-body-tracking": "dimos.teleop.webxr.blueprints:demo_pico_body_tracking",
     "demo-skill": "dimos.agents.skills.demo_skill:demo_skill",
     "demo-virtual-mid360-fastlio": "dimos.hardware.sensors.lidar.virtual_mid360.blueprints:demo_virtual_mid360_fastlio",
     "demo-virtual-mid360-pointlio": "dimos.hardware.sensors.lidar.virtual_mid360.blueprints:demo_virtual_mid360_pointlio",
@@ -170,6 +171,7 @@
     "b1-connection-module": "dimos.robot.unitree.b1.connection.B1ConnectionModule",
     "basic-path-follower": "dimos.navigation.basic_path_follower.module.BasicPathFollower",
     "benchmarker": "dimos.control.benchmarking.benchmark.Benchmarker",
+    "body-tracking-monitor": "dimos.teleop.webxr.body_tracking_monitor.BodyTrackingMonitor",
     "camera-module": "dimos.hardware.sensors.camera.module.CameraModule",
     "camera-mux-module": "dimos.teleop.hosted.camera_mux.CameraMuxModule",
     "cartesian-motion-controller": "dimos.manipulation.control.servo_control.cartesian_motion_controller.CartesianMotionController",

dimos/teleop/webxr/README.md

@@ -97,7 +97,16 @@ entire session; both hands must engage again before commands resume.
 
 **Axes**: thumbstick X, thumbstick Y, trigger (analog), grip (analog)
 
-**Buttons**: trigger, grip, touchpad, thumbstick, X/A, Y/B, menu
+**Buttons**: trigger, grip, touchpad, thumbstick, X/A, Y/B, optional menu. WebXR
+omits a platform-reserved menu button on devices such as PICO controllers.
+
+## Body Tracking Messages
+
+The WebSocket carries two frame formats. Controller poses and joystick state use
+binary LCM messages. When body tracking is enabled, the browser sends JSON text
+frames containing every joint resolved by the headset. A `null` joint map means
+the body source is unavailable; an empty map means no joints resolved for that
+frame.
 
 ## File Structure
 

dimos/teleop/webxr/blueprints.py

@@ -33,12 +33,14 @@
     coordinator_teleop_xarm7,
 )
 from dimos.robot.unitree.go2.connection import GO2Connection
+from dimos.teleop.webxr.body_tracking_monitor import BodyTrackingMonitor
 from dimos.teleop.webxr.extensions import (
     ArmTeleopModule,
     Go2TeleopModule,
     HandTeleopModule,
     VideoArmTeleopModule,
 )
+from dimos.teleop.webxr.module import WebXRTeleopModule
 from dimos.visualization.vis_module import vis_module
 
 # Arm teleop with press-and-hold engage (has rerun viz)
@@ -163,3 +165,10 @@
     )
     .global_config(robot_model="unitree_go2")
 )
+
+
+# PICO 4 Ultra WebXR API test: require body tracking and report every usable joint.
+demo_pico_body_tracking = autoconnect(
+    WebXRTeleopModule.blueprint(body_tracking_mode="required"),
+    BodyTrackingMonitor.blueprint(),
+)

dimos/teleop/webxr/body_tracking.py

@@ -0,0 +1,50 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Body-joint snapshots received from a WebXR client."""
+
+from typing import Annotated, Literal, TypeAlias
+
+from pydantic import BaseModel, ConfigDict, Field, StringConstraints
+
+BodyTrackingMode: TypeAlias = Literal["off", "optional", "required"]
+_FiniteFloat: TypeAlias = Annotated[float, Field(strict=True, allow_inf_nan=False)]
+_NonEmptyString: TypeAlias = Annotated[
+    str,
+    StringConstraints(min_length=1, pattern=r".*\S.*"),
+]
+
+
+class BodyJointPose(BaseModel):
+    """One body joint's pose in the snapshot's WebXR reference space."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    position: tuple[_FiniteFloat, _FiniteFloat, _FiniteFloat]
+    orientation: tuple[_FiniteFloat, _FiniteFloat, _FiniteFloat, _FiniteFloat]
+
+
+class BodyTrackingSnapshot(BaseModel):
+    """Named body-joint poses captured in one WebXR reference space.
+
+    ``joints=None`` means the body source is unavailable. An empty mapping
+    means the source is available but did not resolve any joints.
+    """
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    type: Literal["body_tracking_snapshot"]
+    capture_time_s: _FiniteFloat
+    frame_id: _NonEmptyString
+    joints: dict[_NonEmptyString, BodyJointPose] | None

dimos/teleop/webxr/body_tracking_monitor.py

@@ -0,0 +1,104 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Live health reporting for the PICO WebXR body-tracking demo."""
+
+from time import monotonic
+from typing import Any
+
+from reactivex.disposable import Disposable
+
+from dimos.core.core import rpc
+from dimos.core.module import Module
+from dimos.core.stream import In
+from dimos.teleop.webxr.body_tracking import BodyTrackingSnapshot
+from dimos.utils.logging_config import setup_logger
+
+logger = setup_logger()
+
+REPORT_INTERVAL_S = 5.0
+
+
+def body_tracking_summary(
+    snapshot: BodyTrackingSnapshot,
+    *,
+    snapshot_rate_hz: float,
+    resolved_joint_ever_seen: bool,
+) -> dict[str, Any]:
+    """Build one compact body-tracking health summary."""
+    joints = snapshot.joints
+    state = "unavailable" if joints is None else "empty" if not joints else "tracking"
+    positions: dict[str, tuple[float, float, float]] = {}
+    if joints:
+        positions = {
+            name: (
+                round(pose.position[0], 3),
+                round(pose.position[1], 3),
+                round(pose.position[2], 3),
+            )
+            for name, pose in joints.items()
+        }
+
+    return {
+        "snapshot_rate_hz": round(snapshot_rate_hz, 1),
+        "state": state,
+        "reference_space": snapshot.frame_id,
+        "resolved_joint_count": 0 if joints is None else len(joints),
+        "resolved_joint_ever_seen": resolved_joint_ever_seen,
+        "joint_positions": positions,
+    }
+
+
+class BodyTrackingMonitor(Module):
+    """Report live PICO body-tracking availability, rate, and joint poses."""
+
+    body_tracking: In[BodyTrackingSnapshot]
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self._report_started_at = monotonic()
+        self._snapshots_since_report = 0
+        self._resolved_joint_ever_seen = False
+
+    @rpc
+    def start(self) -> None:
+        super().start()
+        self.register_disposable(Disposable(self.body_tracking.subscribe(self._on_body_tracking)))
+
+    def _on_body_tracking(self, snapshot: BodyTrackingSnapshot) -> None:
+        self._snapshots_since_report += 1
+        if snapshot.joints and not self._resolved_joint_ever_seen:
+            self._resolved_joint_ever_seen = True
+            logger.info(
+                "WebXR body tracking acquired",
+                reference_space=snapshot.frame_id,
+                resolved_joint_count=len(snapshot.joints),
+            )
+
+        now = monotonic()
+        elapsed = now - self._report_started_at
+        if elapsed < REPORT_INTERVAL_S:
+            return
+
+        summary = body_tracking_summary(
+            snapshot,
+            snapshot_rate_hz=self._snapshots_since_report / elapsed,
+            resolved_joint_ever_seen=self._resolved_joint_ever_seen,
+        )
+        if snapshot.joints:
+            logger.info("WebXR body tracking health", **summary)
+        else:
+            logger.warning("WebXR body tracking has no resolved joints", **summary)
+        self._report_started_at = now
+        self._snapshots_since_report = 0

dimos/teleop/webxr/controller_types.py

@@ -49,11 +49,11 @@ class WebXRControllerState:
         0: thumbstick X, 1: thumbstick Y, 2: trigger (analog), 3: grip (analog)
     Button indices (digital, 0 or 1):
         0: trigger, 1: grip, 2: touchpad, 3: thumbstick,
-        4: X/A, 5: Y/B, 6: menu
+        4: X/A, 5: Y/B, 6: menu (optional)
     """
 
     EXPECTED_AXES: ClassVar[int] = 4
-    EXPECTED_BUTTONS: ClassVar[int] = 7
+    REQUIRED_BUTTONS: ClassVar[int] = 6
 
     is_left: bool = True
     # Analog values (0.0-1.0)
@@ -72,17 +72,17 @@ class WebXRControllerState:
     def from_joy(cls, joy: Joy, is_left: bool = True) -> "WebXRControllerState":
         """Create WebXRControllerState from Joy message.
         Expected axes: [thumbstick_x, thumbstick_y, trigger_analog, grip_analog]
-        Expected buttons: [trigger, grip, touchpad, thumbstick, X/A, Y/B, menu]
+        Expected buttons: [trigger, grip, touchpad, thumbstick, X/A, Y/B, optional menu]
         Raises:
             ValueError: If Joy message doesn't have expected WebXR controller format.
         """
         buttons = joy.buttons or []
         axes = joy.axes or []
 
-        if len(buttons) < cls.EXPECTED_BUTTONS:
-            raise ValueError(f"Expected {cls.EXPECTED_BUTTONS} buttons, got {len(buttons)}")
         if len(axes) < cls.EXPECTED_AXES:
             raise ValueError(f"Expected {cls.EXPECTED_AXES} axes, got {len(axes)}")
+        if len(buttons) < cls.REQUIRED_BUTTONS:
+            raise ValueError(f"Expected {cls.REQUIRED_BUTTONS} buttons, got {len(buttons)}")
 
         return cls(
             is_left=is_left,
@@ -92,7 +92,7 @@ def from_joy(cls, joy: Joy, is_left: bool = True) -> "WebXRControllerState":
             thumbstick_press=buttons[3] > 0.5,
             primary=buttons[4] > 0.5,
             secondary=buttons[5] > 0.5,
-            menu=buttons[6] > 0.5,
+            menu=len(buttons) > 6 and buttons[6] > 0.5,
             thumbstick=ThumbstickState(x=float(axes[0]), y=float(axes[1])),
         )
 

dimos/teleop/webxr/module.py

@@ -35,7 +35,7 @@
 from fastapi import WebSocket, WebSocketDisconnect
 from fastapi.responses import HTMLResponse
 from fastapi.staticfiles import StaticFiles
-from pydantic import Field
+from pydantic import Field, ValidationError
 from reactivex.disposable import Disposable
 
 from dimos.constants import DIMOS_PROJECT_ROOT
@@ -46,6 +46,10 @@
 from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
 from dimos.msgs.sensor_msgs.Joy import Joy
 from dimos.teleop.utils.teleop_transforms import webxr_to_robot
+from dimos.teleop.webxr.body_tracking import (
+    BodyTrackingMode,
+    BodyTrackingSnapshot,
+)
 
 # Hand is re-exported for callers; it lives in controller_types.
 from dimos.teleop.webxr.controller_types import Buttons, Hand, WebXRControllerState
@@ -83,6 +87,7 @@ class WebXRTeleopConfig(ModuleConfig):
     control_loop_hz: float = 50.0
     server_port: int = 8443
     input_timeout_s: float = Field(default=1.0, gt=0)
+    body_tracking_mode: BodyTrackingMode = "off"
 
 
 _Config = TypeVar("_Config", bound=WebXRTeleopConfig)
@@ -99,6 +104,7 @@ class WebXRTeleopModule(Module):
         - left_controller_output: PoseStamped (output pose for left hand)
         - right_controller_output: PoseStamped (output pose for right hand)
         - teleop_buttons: Buttons (button states for both controllers)
+        - body_tracking: named body-joint poses in their WebXR reference space
     """
 
     config: WebXRTeleopConfig
@@ -108,6 +114,7 @@ class WebXRTeleopModule(Module):
     right_controller_output: Out[PoseStamped]
     teleop_buttons: Out[Buttons]
     status: In[EpisodeStatus]
+    body_tracking: Out[BodyTrackingSnapshot]
 
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(**kwargs)
@@ -159,6 +166,10 @@ async def teleop_index() -> HTMLResponse:
             index_path = STATIC_DIR / "index.html"
             return HTMLResponse(content=index_path.read_text())
 
+        @self._web_server.app.get("/teleop/config")
+        async def teleop_config() -> dict[str, Any]:
+            return self._webxr_client_config()
+
         if STATIC_DIR.is_dir():
             self._web_server.app.mount(
                 "/static", StaticFiles(directory=str(STATIC_DIR)), name="teleop_static"
@@ -175,20 +186,63 @@ async def websocket_endpoint(ws: WebSocket) -> None:
             logger.info("WebXR client connected")
             try:
                 while True:
-                    data = await ws.receive_bytes()
-                    fingerprint = data[:8]
-                    decoder = self._decoders.get(fingerprint)
-                    if decoder:
-                        decoder(data)
-                    else:
-                        logger.warning(f"Unknown message fingerprint: {fingerprint.hex()}")
+                    message = await ws.receive()
+                    if message["type"] == "websocket.disconnect":
+                        logger.info("WebXR client disconnected")
+                        break
+                    data = message.get("bytes")
+                    text = message.get("text")
+                    if data is not None:
+                        self._dispatch_binary_message(data)
+                    elif text is not None:
+                        self._dispatch_text_message(text)
             except WebSocketDisconnect:
                 logger.info("WebXR client disconnected")
             except Exception:
                 logger.exception("WebSocket error")
             finally:
                 self._client_disconnected(ws)
 
+    def _webxr_client_config(self) -> dict[str, Any]:
+        required_features = ["local-floor"]
+        optional_features = ["hand-tracking"]
+        session_modes = ["immersive-ar", "immersive-vr"]
+
+        if self.config.body_tracking_mode != "off":
+            optional_features.append("bounded-floor")
+        if self.config.body_tracking_mode == "optional":
+            optional_features.append("body-tracking")
+        elif self.config.body_tracking_mode == "required":
+            required_features.append("body-tracking")
+            session_modes = ["immersive-ar"]
+
+        return {
+            "body_tracking_mode": self.config.body_tracking_mode,
+            "session_modes": session_modes,
+            "session_options": {
+                "requiredFeatures": required_features,
+                "optionalFeatures": optional_features,
+            },
+        }
+
+    def _dispatch_binary_message(self, data: bytes) -> bool:
+        fingerprint = data[:8]
+        decoder = self._decoders.get(fingerprint)
+        if decoder is None:
+            logger.warning("Unknown WebXR message fingerprint", fingerprint=fingerprint.hex())
+            return False
+        decoder(data)
+        return True
+
+    def _dispatch_text_message(self, payload: str) -> bool:
+        try:
+            snapshot = BodyTrackingSnapshot.model_validate_json(payload)
+        except ValidationError as exc:
+            logger.warning("Dropping malformed WebXR body snapshot", error=str(exc))
+            return False
+        self.body_tracking.publish(snapshot)
+        return True
+
     def _client_connected(self, ws: WebSocket) -> bool:
         with self._clients_lock:
             if self._connected_clients: