feat: new openarm adapter

CONTEXT.md

@@ -0,0 +1,21 @@
+# DimOS Robotics Context
+
+DimOS describes robots, actuators, and control surfaces using precise robotics terminology. This glossary records domain language only, not implementation details.
+
+## Language
+
+**Damiao-based Robot**:
+A robot whose joints are actuated by one or more Damiao motors, possibly spread across multiple CAN buses and physical limbs.
+_Avoid_: Damiao arm when the robot may contain multiple motor groups
+
+**Damiao Joint Group**:
+An ordered set of Damiao-driven joints that forms a meaningful physical group such as an arm, torso, or other controllable body section.
+_Avoid_: Arm when the group is not necessarily an arm
+
+**Damiao Bus**:
+A named communication channel used by a Damiao-based Robot to reach one or more Damiao motors.
+_Avoid_: Treating a bus as owned by a single joint group when multiple groups may share a channel
+
+**OpenArm**:
+An OpenArm robot configuration built from Damiao motors, with OpenArm-specific joints, side naming, limits, and robot description.
+_Avoid_: Damiao robot when referring to OpenArm-specific geometry or naming

dimos/control/hardware_interface.py

@@ -281,16 +281,17 @@ def read_state(self) -> dict[JointName, JointState]:
             for i, name in enumerate(self._joint_names)
         }
 
-    def write_command(self, commands: dict[str, float], _mode: ControlMode) -> bool:
+    def write_command(self, commands: dict[str, float], mode: ControlMode) -> bool:
         """Write velocity commands — always sends velocities regardless of mode.
 
         Args:
             commands: {joint_name: velocity} - can be partial
-            _mode: Control mode (ignored — twist bases always use velocity)
+            mode: Control mode (ignored — twist bases always use velocity)
 
         Returns:
             True if command was sent successfully
         """
+        del mode
         # Update last commanded for joints we received
         for joint_name, value in commands.items():
             if joint_name in self._last_commanded:
@@ -390,47 +391,62 @@ def read_state(self) -> dict[JointName, JointState]:
         }
 
     def write_command(self, commands: dict[str, float], mode: ControlMode) -> bool:
-        """Write position commands — converts to MotorCommand with per-joint PD gains.
-
-        Only POSITION / SERVO_POSITION are supported; other modes are warned
-        and dropped (matches ConnectedHardware's warn-and-skip pattern).
-        Per-joint kp/kd come from ``component.wb_config`` (resolved in
-        ``__init__``); fall back to ``_DEFAULT_KP``/``_DEFAULT_KD`` when
-        the blueprint didn't supply gains.
-        """
-        from dimos.hardware.whole_body.spec import MotorCommand
+        """Dispatch coordinator joint-position commands by control mode."""
 
         if mode not in (ControlMode.POSITION, ControlMode.SERVO_POSITION):
             logger.warning(
                 f"WholeBody {self.hardware_id} only supports POSITION/SERVO_POSITION; "
                 f"got {mode.name} — skipping"
             )
             return False
+        return self.write_position(commands)
+
+    def write_position(self, commands: dict[str, float]) -> bool:
+        """Write named position commands using native adapter position IO when available.
+
+        Unknown joints are warned once and ignored. Partial commands hold the
+        previous commanded value for omitted joints. The hold-last cache is
+        committed only after the underlying adapter accepts the full frame.
+        """
 
         if not self._initialized and not self._try_initialize_last_commanded():
             return False
 
+        candidate = dict(self._last_commanded)
         for joint_name, value in commands.items():
             if joint_name in self._joint_names:
-                self._last_commanded[joint_name] = value
+                candidate[joint_name] = value
             elif joint_name not in self._warned_unknown_joints:
                 logger.warning(
                     f"WholeBody {self.hardware_id} received command for unknown joint "
                     f"{joint_name}. Valid joints: {self._joint_names}"
                 )
                 self._warned_unknown_joints.add(joint_name)
 
+        positions = [candidate[name] for name in self._joint_names]
+        write_joint_positions = getattr(self._wb_adapter, "write_joint_positions", None)
+        if callable(write_joint_positions):
+            ok = bool(write_joint_positions(positions))
+            if ok:
+                self._last_commanded = candidate
+            return ok
+
+        from dimos.hardware.whole_body.spec import MotorCommand
+
         motor_cmds = [
             MotorCommand(
-                q=self._last_commanded[name],
+                q=candidate[name],
                 dq=0.0,
                 kp=self._kp_by_name[name],
                 kd=self._kd_by_name[name],
                 tau=0.0,
             )
             for name in self._joint_names
         ]
-        return self._wb_adapter.write_motor_commands(motor_cmds)
+        ok = self._wb_adapter.write_motor_commands(motor_cmds)
+        if ok:
+            self._last_commanded = candidate
+        return ok
 
     def write_motor_commands(self, commands: list[MotorCommand]) -> bool:
         """Direct pass-through to adapter for full MotorCommand control."""
@@ -441,6 +457,12 @@ def _try_initialize_last_commanded(self) -> bool:
         if not self._wb_adapter.has_motor_states():
             return False
         states = self._wb_adapter.read_motor_states()
+        if len(states) != len(self._joint_names):
+            logger.warning(
+                f"WholeBody {self.hardware_id} read {len(states)} motor states for "
+                f"{len(self._joint_names)} joints; skipping command initialization"
+            )
+            return False
         for i, name in enumerate(self._joint_names):
             self._last_commanded[name] = states[i].q
         self._initialized = True

dimos/control/test_control.py

@@ -18,15 +18,17 @@
 
 import threading
 import time
+from typing import cast
 from unittest.mock import MagicMock
 
 import pytest
 
-from dimos.control.components import HardwareComponent, HardwareType, make_joints
+from dimos.control.components import HardwareComponent, HardwareType, TaskName, make_joints
 from dimos.control.coordinator import ControlCoordinator
 from dimos.control.hardware_interface import ConnectedHardware
 from dimos.control.task import (
     ControlMode,
+    ControlTask,
     CoordinatorState,
     JointCommandOutput,
     JointStateSnapshot,
@@ -469,7 +471,7 @@ def test_tick_loop_starts_and_stops(self, mock_adapter):
         )
         hw = ConnectedHardware(mock_adapter, component)
         hardware = {"arm": hw}
-        tasks: dict = {}
+        tasks: dict[TaskName, ControlTask] = {}
         joint_to_hardware = {f"arm/joint{i + 1}": "arm" for i in range(6)}
 
         tick_loop = TickLoop(
@@ -512,7 +514,7 @@ def test_tick_loop_calls_compute(self, mock_adapter):
             mode=ControlMode.POSITION,
         )
 
-        tasks = {"test_task": mock_task}
+        tasks: dict[TaskName, ControlTask] = {"test_task": cast("ControlTask", mock_task)}
         joint_to_hardware = {f"arm/joint{i + 1}": "arm" for i in range(6)}
 
         tick_loop = TickLoop(
@@ -546,7 +548,7 @@ def test_full_trajectory_execution(self, mock_adapter):
             priority=10,
         )
         traj_task = JointTrajectoryTask(name="traj_arm", config=config)
-        tasks = {"traj_arm": traj_task}
+        tasks: dict[TaskName, ControlTask] = {"traj_arm": traj_task}
 
         joint_to_hardware = {f"arm/joint{i + 1}": "arm" for i in range(6)}
 

dimos/control/test_hardware_interface.py

@@ -0,0 +1,107 @@
+# Copyright 2025-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.
+
+from __future__ import annotations
+
+from dimos.control.components import HardwareComponent, HardwareType
+from dimos.control.hardware_interface import ConnectedWholeBody
+from dimos.hardware.whole_body.spec import IMUState, MotorCommand, MotorState
+
+
+class _NativePositionWholeBodyAdapter:
+    def __init__(self) -> None:
+        self.accept = False
+        self.position_writes: list[list[float]] = []
+
+    def connect(self) -> bool:
+        return True
+
+    def disconnect(self) -> None:
+        return None
+
+    def is_connected(self) -> bool:
+        return True
+
+    def read_motor_states(self) -> list[MotorState]:
+        return [MotorState(q=0.0), MotorState(q=0.0)]
+
+    def has_motor_states(self) -> bool:
+        return True
+
+    def read_imu(self) -> IMUState:
+        return IMUState()
+
+    def write_motor_commands(self, commands: list[MotorCommand]) -> bool:
+        raise AssertionError("native position path should not call write_motor_commands")
+
+    def write_joint_positions(self, positions: list[float]) -> bool:
+        self.position_writes.append(list(positions))
+        return self.accept
+
+
+class _MotorCommandWholeBodyAdapter:
+    def __init__(self) -> None:
+        self.motor_writes: list[list[MotorCommand]] = []
+
+    def connect(self) -> bool:
+        return True
+
+    def disconnect(self) -> None:
+        return None
+
+    def is_connected(self) -> bool:
+        return True
+
+    def read_motor_states(self) -> list[MotorState]:
+        return [MotorState(q=0.0), MotorState(q=0.0)]
+
+    def has_motor_states(self) -> bool:
+        return True
+
+    def read_imu(self) -> IMUState:
+        return IMUState()
+
+    def write_motor_commands(self, commands: list[MotorCommand]) -> bool:
+        self.motor_writes.append(commands)
+        return True
+
+
+def _component() -> HardwareComponent:
+    return HardwareComponent(
+        hardware_id="body",
+        hardware_type=HardwareType.WHOLE_BODY,
+        joints=["body/j1", "body/j2"],
+    )
+
+
+def test_whole_body_write_position_uses_native_adapter_and_commits_only_on_success() -> None:
+    adapter = _NativePositionWholeBodyAdapter()
+    connected = ConnectedWholeBody(adapter=adapter, component=_component())
+
+    assert connected.write_position({"body/j1": 2.0}) is False
+    adapter.accept = True
+    assert connected.write_position({"body/j2": 3.0}) is True
+
+    assert adapter.position_writes == [[2.0, 0.0], [0.0, 3.0]]
+
+
+def test_whole_body_write_position_falls_back_to_motor_commands() -> None:
+    adapter = _MotorCommandWholeBodyAdapter()
+    connected = ConnectedWholeBody(adapter=adapter, component=_component())
+
+    assert connected.write_position({"body/j1": 1.0}) is True
+
+    commands = adapter.motor_writes[-1]
+    assert [command.q for command in commands] == [1.0, 0.0]
+    assert [command.dq for command in commands] == [0.0, 0.0]

dimos/hardware/damiao/__init__.py

@@ -0,0 +1,38 @@
+# Copyright 2025-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.
+
+"""Shared Damiao actuator/runtime adapters."""
+
+from dimos.hardware.damiao.arm_adapter import DamiaoArmAdapter
+from dimos.hardware.damiao.runtime import DamiaoBindingUnavailableError, DamiaoRobotRuntime
+from dimos.hardware.damiao.specs import (
+    DamiaoArmSpec,
+    DamiaoBusSpec,
+    DamiaoJointGroupSpec,
+    DamiaoMotorSpec,
+    DamiaoRobotSpec,
+)
+from dimos.hardware.damiao.whole_body_adapter import DamiaoWholeBodyAdapter
+
+__all__ = [
+    "DamiaoArmAdapter",
+    "DamiaoArmSpec",
+    "DamiaoBindingUnavailableError",
+    "DamiaoBusSpec",
+    "DamiaoJointGroupSpec",
+    "DamiaoMotorSpec",
+    "DamiaoRobotRuntime",
+    "DamiaoRobotSpec",
+    "DamiaoWholeBodyAdapter",
+]