Ruthwik/go2sdk: Prev. Go2 low-level adapter

dimos/control/components.py

@@ -203,6 +203,41 @@ def make_humanoid_joints(hardware_id: HardwareId) -> list[JointName]:
     return [f"{hardware_id}/{j}" for j in _HUMANOID_29DOF_JOINTS]
 
 
+# Index order matches Unitree's canonical LowCmd_.motor_cmd[0..11] layout:
+# FR_hip=0, FR_thigh=1, FR_calf=2, FL_hip=3, FL_thigh=4, FL_calf=5,
+# RR_hip=6, RR_thigh=7, RR_calf=8, RL_hip=9, RL_thigh=10, RL_calf=11.
+_QUADRUPED_12DOF_JOINTS = [
+    "FR_hip",
+    "FR_thigh",
+    "FR_calf",
+    "FL_hip",
+    "FL_thigh",
+    "FL_calf",
+    "RR_hip",
+    "RR_thigh",
+    "RR_calf",
+    "RL_hip",
+    "RL_thigh",
+    "RL_calf",
+]
+
+
+def make_quadruped_joints(hardware_id: HardwareId) -> list[JointName]:
+    """Create joint names for a 12-DOF Unitree-layout quadruped.
+
+    Order matches Unitree's LowCmd_.motor_cmd[0..11] indexing (FR, FL, RR, RL,
+    each hip→thigh→calf). Single-source-of-truth so wholebody Modules and the
+    coordinator-side adapter agree on the wire-level name → motor-index mapping.
+
+    Args:
+        hardware_id: The hardware identifier (e.g., "go2")
+
+    Returns:
+        List of 12 joint names like ["go2/FR_hip", ..., "go2/RL_calf"]
+    """
+    return [f"{hardware_id}/{j}" for j in _QUADRUPED_12DOF_JOINTS]
+
+
 __all__ = [
     "TWIST_SUFFIX_MAP",
     "HardwareComponent",
@@ -214,6 +249,7 @@ def make_humanoid_joints(hardware_id: HardwareId) -> list[JointName]:
     "make_gripper_joints",
     "make_humanoid_joints",
     "make_joints",
+    "make_quadruped_joints",
     "make_twist_base_joints",
     "split_joint_name",
 ]

dimos/robot/all_blueprints.py

@@ -107,6 +107,8 @@
     "unitree-go2-relocalization": "dimos.robot.unitree.go2.blueprints.smart.unitree_go2:unitree_go2_relocalization",
     "unitree-go2-ros": "dimos.robot.unitree.go2.blueprints.smart.unitree_go2_ros:unitree_go2_ros",
     "unitree-go2-security": "dimos.robot.unitree.go2.blueprints.agentic.unitree_go2_security:unitree_go2_security",
+    "unitree-go2-wholebody-commander": "dimos.robot.unitree.go2.blueprints.basic.unitree_go2_wholebody_commander:unitree_go2_wholebody_commander",
+    "unitree-go2-wholebody-coordinator": "dimos.robot.unitree.go2.blueprints.basic.unitree_go2_wholebody_coordinator:unitree_go2_wholebody_coordinator",
     "unitree-go2-spatial": "dimos.robot.unitree.go2.blueprints.smart.unitree_go2_spatial:unitree_go2_spatial",
     "unitree-go2-temporal-memory": "dimos.robot.unitree.go2.blueprints.agentic.unitree_go2_temporal_memory:unitree_go2_temporal_memory",
     "unitree-go2-vlm-stream-test": "dimos.robot.unitree.go2.blueprints.smart.unitree_go2_vlm_stream_test:unitree_go2_vlm_stream_test",

dimos/robot/unitree/go2/blueprints/basic/unitree_go2_wholebody_commander.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Go2 wholebody coordinator + stdin joint commander, all in one process.
+
+Variant of `unitree_go2_wholebody_coordinator` that adds a stdin-driven
+Go2JointCommanderModule so you can type pose commands at the terminal:
+
+    ROBOT_INTERFACE=<nic> dimos run unitree-go2-wholebody-commander
+
+Then at the prompt:
+    go2> stand
+    go2> set 1 0.5
+    go2> lay
+    go2> quit
+
+You still need to arm the coordinator from a separate Python shell:
+    app = Dimos.connect()
+    app.ControlCoordinator.set_dry_run(False)
+    app.ControlCoordinator.set_activated(True)
+"""
+
+from __future__ import annotations
+
+import os
+
+from dimos.control.components import HardwareComponent, HardwareType, make_quadruped_joints
+from dimos.control.coordinator import ControlCoordinator, TaskConfig
+from dimos.core.coordination.blueprints import autoconnect
+from dimos.core.transport import LCMTransport
+from dimos.msgs.sensor_msgs.Imu import Imu
+from dimos.msgs.sensor_msgs.JointState import JointState
+from dimos.msgs.sensor_msgs.MotorCommandArray import MotorCommandArray
+from dimos.robot.unitree.go2.joint_commander_module import Go2JointCommanderModule
+from dimos.robot.unitree.go2.wholebody_connection import Go2WholeBodyConnection
+
+_go2_joints = make_quadruped_joints("go2")
+
+unitree_go2_wholebody_commander = (
+    autoconnect(
+        Go2JointCommanderModule.blueprint(initial_pose="zero"),
+        Go2WholeBodyConnection.blueprint(
+            release_sport_mode=True,
+            network_interface=os.getenv("ROBOT_INTERFACE", ""),
+        ),
+        ControlCoordinator.blueprint(
+            tick_rate=500,
+            hardware=[
+                HardwareComponent(
+                    hardware_id="go2",
+                    hardware_type=HardwareType.WHOLE_BODY,
+                    joints=_go2_joints,
+                    adapter_type="transport_lcm",
+                ),
+            ],
+            tasks=[
+                TaskConfig(
+                    name="servo_go2",
+                    type="servo",
+                    joint_names=_go2_joints,
+                    priority=10,
+                ),
+            ],
+        ),
+    )
+    # Commander's joint_command Out and ControlCoordinator's joint_command In
+    # share the same name → autoconnect wires them in-process for free, but we
+    # also LCM-bridge the topic so external publishers / observers can see it.
+    .transports(
+        {
+            ("motor_states", JointState): LCMTransport("/go2/motor_states", JointState),
+            ("imu", Imu): LCMTransport("/go2/imu", Imu),
+            ("motor_command", MotorCommandArray): LCMTransport(
+                "/go2/motor_command", MotorCommandArray
+            ),
+            ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+            ("joint_command", JointState): LCMTransport("/go2/joint_command", JointState),
+        }
+    )
+)
+
+
+__all__ = ["unitree_go2_wholebody_commander"]

dimos/robot/unitree/go2/blueprints/basic/unitree_go2_wholebody_coordinator.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Go2 ControlCoordinator: Go2WholeBodyConnection Module + servo task via LCM bridge.
+
+Mirrors `unitree_g1_coordinator.py`. Run with
+`ROBOT_INTERFACE=<nic> dimos run unitree-go2-wholebody-coordinator`.
+"""
+
+from __future__ import annotations
+
+import os
+
+from dimos.control.components import HardwareComponent, HardwareType, make_quadruped_joints
+from dimos.control.coordinator import ControlCoordinator, TaskConfig
+from dimos.core.coordination.blueprints import autoconnect
+from dimos.core.transport import LCMTransport
+from dimos.msgs.sensor_msgs.Imu import Imu
+from dimos.msgs.sensor_msgs.JointState import JointState
+from dimos.msgs.sensor_msgs.MotorCommandArray import MotorCommandArray
+from dimos.robot.unitree.go2.wholebody_connection import Go2WholeBodyConnection
+
+_go2_joints = make_quadruped_joints("go2")
+
+# ROBOT_INTERFACE pins cyclonedds to a NIC; required on multi-NIC hosts.
+unitree_go2_wholebody_coordinator = (
+    autoconnect(
+        Go2WholeBodyConnection.blueprint(
+            release_sport_mode=True,
+            network_interface=os.getenv("ROBOT_INTERFACE", ""),
+        ),
+        ControlCoordinator.blueprint(
+            tick_rate=500,
+            hardware=[
+                HardwareComponent(
+                    hardware_id="go2",
+                    hardware_type=HardwareType.WHOLE_BODY,
+                    joints=_go2_joints,
+                    adapter_type="transport_lcm",
+                ),
+            ],
+            tasks=[
+                TaskConfig(
+                    name="servo_go2",
+                    type="servo",
+                    joint_names=_go2_joints,
+                    priority=10,
+                ),
+            ],
+        ),
+    )
+    # No remappings: Module stream names (motor_states/imu/motor_command) don't
+    # collide with ControlCoordinator's (joint_state/joint_command/...).
+    .transports(
+        {
+            ("motor_states", JointState): LCMTransport("/go2/motor_states", JointState),
+            ("imu", Imu): LCMTransport("/go2/imu", Imu),
+            ("motor_command", MotorCommandArray): LCMTransport(
+                "/go2/motor_command", MotorCommandArray
+            ),
+            ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+            ("joint_command", JointState): LCMTransport("/go2/joint_command", JointState),
+        }
+    )
+)
+
+
+__all__ = ["unitree_go2_wholebody_coordinator"]