feat(imitation): add native OpenYAM collection

.github/workflows/ci.yml

@@ -329,10 +329,10 @@ jobs:
         with:
           enable-cache: true
           prune-cache: true
-      - name: Install dependency for pyaudio
+      - name: Install Python audio and image dependencies
         run: |
           sudo apt-get update
-          sudo apt-get install -y portaudio19-dev
+          sudo apt-get install -y portaudio19-dev libturbojpeg
       - name: Build and test PyO3 bindings
         # PyO3 extension modules (cdylib + `extension-module`) are the one case
         # that still needs a plain build: pytest imports the compiled .so.
@@ -352,7 +352,10 @@ jobs:
       - name: Native module e2e tests
         run: |
           cargo build --release --locked -p dimos-livox -p dimos-virtual-mid360
-          uv run pytest -m native_e2e dimos/hardware/sensors/lidar/livox/test_e2e.py --no-cov
+          uv run pytest -m native_e2e \
+            dimos/hardware/sensors/lidar/livox/test_e2e.py \
+            dimos/robot/manipulators/openyam/blueprints/test_learning_collection_e2e.py \
+            dimos/imitation/test_datacollection_e2e.py --no-cov
 
   native:
     name: Native builds (C++ and Rust)

dimos/agents/demo_agent.py

@@ -17,25 +17,21 @@
 from dimos.agents.skills.observe_skill import ObserveSkill
 from dimos.core.coordination.blueprints import autoconnect
 from dimos.hardware.sensors.camera.module import CameraModule
-from dimos.hardware.sensors.camera.webcam import Webcam
+from dimos.hardware.sensors.camera.webcam import WebcamConfig
 from dimos.hardware.sensors.camera.zed import compat as zed
 
 demo_agent = autoconnect(McpServer.blueprint(), McpClient.blueprint())
 
 
-def _create_webcam() -> Webcam:
-    return Webcam(
-        camera_index=0,
-        fps=15,
-        camera_info=zed.CameraInfo.SingleWebcam,
-    )
-
-
 demo_agent_camera = autoconnect(
     McpServer.blueprint(),
     McpClient.blueprint(),
     ObserveSkill.blueprint(),
     CameraModule.blueprint(
-        hardware=_create_webcam,
+        hardware=WebcamConfig(
+            camera_index=0,
+            fps=15,
+            camera_info=zed.CameraInfo.SingleWebcam,
+        ),
     ),
 )

dimos/cli/dimos.py

@@ -53,7 +53,6 @@
 from dimos.cli.commands.bake import bake
 from dimos.cli.commands.cameracalibrate import cameracalibrate
 from dimos.cli.commands.data import data_app
-from dimos.cli.commands.dataprep import dataprep_app
 from dimos.cli.commands.docs import docs
 from dimos.cli.commands.global_options import create_dynamic_callback
 from dimos.cli.commands.graph import graph
@@ -143,7 +142,6 @@ def cli_main() -> None:
 from dimos.navigation.nav_3d.evaluator.cli import app as nav_eval_app
 
 main.add_typer(nav_eval_app, name="nav-eval")
-main.add_typer(dataprep_app, name="dataprep")
 
 from dimos.memory.cli.app import mem_app
 

dimos/control/coordinator.py

@@ -150,6 +150,8 @@ class ControlCoordinator(Module):
 
     # Output: Aggregated joint state for external consumers
     coordinator_joint_state: Out[JointState]
+    # Sparse position updates accepted by hardware, for diagnostics and recording.
+    applied_joint_position_command: Out[JointState]
 
     # Input: Streaming joint commands for real-time control
     joint_command: In[JointState]
@@ -903,6 +905,7 @@ def start(self) -> None:
             task_lock=self._task_lock,
             joint_to_hardware=self._joint_to_hardware,
             publish_callback=publish_cb,
+            publish_command_callback=self.applied_joint_position_command.publish,
             publish_robot_callback=publish_robot_cb,
             frame_id=self.config.joint_state_frame_id,
             log_ticks=self.config.log_ticks,

dimos/control/test_control.py

@@ -1132,7 +1132,9 @@ def test_unready_whole_body_is_excluded_from_read_and_write(self, mocker):
 
         state, per_hardware = tick_loop._read_all_hardware()
         imu = tick_loop._read_all_imu()
-        tick_loop._write_all_hardware({"g1": ({"g1/joint1": 0.25}, ControlMode.SERVO_POSITION)})
+        tick_loop._write_all_hardware(
+            {"g1": ({"g1/joint1": 0.25}, ControlMode.SERVO_POSITION)}, timestamp=123.0
+        )
 
         assert state.joint_positions == {}
         assert per_hardware == {}
@@ -1163,7 +1165,9 @@ def test_ready_whole_body_reads_and_writes(self):
         )
 
         state, _per_hardware = tick_loop._read_all_hardware()
-        tick_loop._write_all_hardware({"g1": ({"g1/joint1": 0.25}, ControlMode.SERVO_POSITION)})
+        tick_loop._write_all_hardware(
+            {"g1": ({"g1/joint1": 0.25}, ControlMode.SERVO_POSITION)}, timestamp=123.0
+        )
 
         assert state.joint_positions == {"g1/joint1": 0.5}
         adapter.write_motor_commands.assert_called_once()
@@ -1296,12 +1300,97 @@ def test_write_all_hardware_rejected_command_logs_error(self, mocker):
             joint_to_hardware={"arm/joint1": "arm"},
         )
 
-        tick_loop._write_all_hardware({"arm": ({"arm/joint1": 0.25}, ControlMode.SERVO_POSITION)})
+        tick_loop._write_all_hardware(
+            {"arm": ({"arm/joint1": 0.25}, ControlMode.SERVO_POSITION)}, timestamp=123.0
+        )
 
         log_error.assert_called_once_with(
             "Hardware arm rejected SERVO_POSITION command from control task"
         )
 
+    @pytest.mark.parametrize(
+        "mode,ready,accepted,error,expected",
+        [
+            (ControlMode.POSITION, True, True, None, [0.25]),
+            (ControlMode.SERVO_POSITION, True, True, None, [0.25]),
+            (ControlMode.VELOCITY, True, True, None, []),
+            (ControlMode.TORQUE, True, True, None, []),
+            (ControlMode.POSITION, False, True, None, []),
+            (ControlMode.POSITION, True, False, None, []),
+            (ControlMode.POSITION, True, True, RuntimeError("write failed"), []),
+        ],
+    )
+    def test_command_feedback_reports_only_successful_position_writes(
+        self, mocker, mode, ready, accepted, error, expected
+    ):
+        hardware = mocker.Mock(spec=ConnectedHardware)
+        hardware.ready_for_control.return_value = ready
+        hardware.write_command.return_value = accepted
+        hardware.write_command.side_effect = error
+        received = []
+        lock = threading.Lock()
+
+        def receive(message):
+            assert not lock.locked()
+            received.append(message)
+
+        tick_loop = TickLoop(
+            tick_rate=100.0,
+            hardware={"arm": hardware},
+            hardware_lock=lock,
+            tasks={},
+            task_lock=threading.Lock(),
+            joint_to_hardware={},
+            publish_command_callback=receive,
+            frame_id="robot",
+        )
+        tick_loop._write_all_hardware({"arm": ({"arm/joint1": 0.25}, mode)}, timestamp=123.0)
+
+        assert [value for msg in received for value in msg.position] == expected
+        assert [msg.name for msg in received] == ([["arm/joint1"]] if expected else [])
+        assert [(msg.ts, msg.frame_id) for msg in received] == (
+            [(123.0, "robot")] if expected else []
+        )
+        assert hardware.write_command.call_count == int(ready)
+
+    def test_command_feedback_combines_successful_writes_and_keeps_sparse_updates(self, mocker):
+        hardware = {
+            name: mocker.Mock(spec=ConnectedHardware) for name in ("left", "right", "failed")
+        }
+        for hw in hardware.values():
+            hw.ready_for_control.return_value = True
+            hw.write_command.return_value = True
+        hardware["failed"].write_command.return_value = False
+        received = []
+        tick_loop = TickLoop(
+            tick_rate=100.0,
+            hardware=hardware,
+            hardware_lock=threading.Lock(),
+            tasks={},
+            task_lock=threading.Lock(),
+            joint_to_hardware={},
+            publish_command_callback=received.append,
+        )
+        tick_loop._write_all_hardware(
+            {
+                "left": ({"left/joint1": 0.25}, ControlMode.POSITION),
+                "failed": ({"failed/joint1": 0.75}, ControlMode.POSITION),
+                "right": ({"right/gripper": 0.5}, ControlMode.SERVO_POSITION),
+            },
+            timestamp=123.0,
+        )
+        tick_loop._write_all_hardware(
+            {
+                "right": ({"right/gripper": 0.6}, ControlMode.SERVO_POSITION),
+            },
+            timestamp=124.0,
+        )
+
+        assert [(msg.name, msg.position) for msg in received] == [
+            (["left/joint1", "right/gripper"], [0.25, 0.5]),
+            (["right/gripper"], [0.6]),
+        ]
+
 
 class TestIntegration:
     def test_full_trajectory_execution(self, mock_adapter, wait_until):

dimos/control/test_per_robot_joint_states.py

@@ -169,7 +169,10 @@ def test_plain_coordinator_needs_no_per_robot_ports(self, make_coordinator):
         coordinator.start()
 
         assert sorted(coordinator.list_hardware()) == ["left_arm", "right_arm"]
-        assert sorted(coordinator.outputs) == ["coordinator_joint_state"]
+        assert sorted(coordinator.outputs) == [
+            "applied_joint_position_command",
+            "coordinator_joint_state",
+        ]
 
 
 class TestPerRobotPublishing: