refactor: split control blueprints + added env variables

dimos/control/blueprints/_hardware.py

@@ -0,0 +1,93 @@
+# 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.
+
+"""Hardware component factories for coordinator blueprints."""
+
+from __future__ import annotations
+
+from dimos.control.components import (
+    HardwareComponent,
+    HardwareType,
+    make_gripper_joints,
+    make_joints,
+    make_twist_base_joints,
+)
+from dimos.core.global_config import global_config
+from dimos.utils.data import LfsPath
+
+XARM7_IP = global_config.xarm7_ip
+XARM6_IP = global_config.xarm6_ip
+CAN_PORT = global_config.can_port
+
+PIPER_MODEL_PATH = LfsPath("piper_description/mujoco_model/piper_no_gripper_description.xml")
+XARM6_MODEL_PATH = LfsPath("xarm_description/urdf/xarm6/xarm6.urdf")
+XARM7_MODEL_PATH = LfsPath("xarm_description/urdf/xarm7/xarm7.urdf")
+
+
+def mock_arm(hw_id: str = "arm", n_joints: int = 7) -> HardwareComponent:
+    """Mock manipulator (no real hardware)."""
+    return HardwareComponent(
+        hardware_id=hw_id,
+        hardware_type=HardwareType.MANIPULATOR,
+        joints=make_joints(hw_id, n_joints),
+        adapter_type="mock",
+    )
+
+
+def xarm7(hw_id: str = "arm", *, gripper: bool = False) -> HardwareComponent:
+    """XArm7 real hardware (7-DOF)."""
+    return HardwareComponent(
+        hardware_id=hw_id,
+        hardware_type=HardwareType.MANIPULATOR,
+        joints=make_joints(hw_id, 7),
+        adapter_type="xarm",
+        address=XARM7_IP,
+        auto_enable=True,
+        gripper_joints=make_gripper_joints(hw_id) if gripper else [],
+    )
+
+
+def xarm6(hw_id: str = "arm", *, gripper: bool = False) -> HardwareComponent:
+    """XArm6 real hardware (6-DOF)."""
+    return HardwareComponent(
+        hardware_id=hw_id,
+        hardware_type=HardwareType.MANIPULATOR,
+        joints=make_joints(hw_id, 6),
+        adapter_type="xarm",
+        address=XARM6_IP,
+        auto_enable=True,
+        gripper_joints=make_gripper_joints(hw_id) if gripper else [],
+    )
+
+
+def piper(hw_id: str = "arm") -> HardwareComponent:
+    """Piper arm (6-DOF, CAN bus)."""
+    return HardwareComponent(
+        hardware_id=hw_id,
+        hardware_type=HardwareType.MANIPULATOR,
+        joints=make_joints(hw_id, 6),
+        adapter_type="piper",
+        address=CAN_PORT,
+        auto_enable=True,
+    )
+
+
+def mock_twist_base(hw_id: str = "base") -> HardwareComponent:
+    """Mock holonomic twist base (3-DOF: vx, vy, wz)."""
+    return HardwareComponent(
+        hardware_id=hw_id,
+        hardware_type=HardwareType.BASE,
+        joints=make_twist_base_joints(hw_id),
+        adapter_type="mock_twist_base",
+    )

dimos/control/blueprints/basic.py

@@ -0,0 +1,117 @@
+# 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.
+
+"""Single-arm coordinator blueprints with trajectory control.
+
+Usage:
+    dimos run coordinator-mock           # Mock 7-DOF arm
+    dimos run coordinator-xarm7          # XArm7 real hardware
+    dimos run coordinator-xarm6          # XArm6 real hardware
+    dimos run coordinator-piper          # Piper arm (CAN bus)
+"""
+
+from __future__ import annotations
+
+from dimos.control.blueprints._hardware import mock_arm, piper, xarm6, xarm7
+from dimos.control.coordinator import TaskConfig, control_coordinator
+from dimos.core.transport import LCMTransport
+from dimos.msgs.sensor_msgs.JointState import JointState
+
+# Minimal blueprint (no hardware, no tasks)
+coordinator_basic = control_coordinator(
+    tick_rate=100.0,
+    publish_joint_state=True,
+    joint_state_frame_id="coordinator",
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+# Mock 7-DOF arm (for testing)
+coordinator_mock = control_coordinator(
+    hardware=[mock_arm()],
+    tasks=[
+        TaskConfig(
+            name="traj_arm",
+            type="trajectory",
+            joint_names=[f"arm_joint{i + 1}" for i in range(7)],
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+# XArm7 real hardware
+coordinator_xarm7 = control_coordinator(
+    hardware=[xarm7()],
+    tasks=[
+        TaskConfig(
+            name="traj_arm",
+            type="trajectory",
+            joint_names=[f"arm_joint{i + 1}" for i in range(7)],
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+# XArm6 real hardware
+coordinator_xarm6 = control_coordinator(
+    hardware=[xarm6()],
+    tasks=[
+        TaskConfig(
+            name="traj_xarm",
+            type="trajectory",
+            joint_names=[f"arm_joint{i + 1}" for i in range(6)],
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+# Piper arm (6-DOF, CAN bus)
+coordinator_piper = control_coordinator(
+    hardware=[piper()],
+    tasks=[
+        TaskConfig(
+            name="traj_piper",
+            type="trajectory",
+            joint_names=[f"arm_joint{i + 1}" for i in range(6)],
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+
+__all__ = [
+    "coordinator_basic",
+    "coordinator_mock",
+    "coordinator_piper",
+    "coordinator_xarm6",
+    "coordinator_xarm7",
+]

dimos/control/blueprints/dual.py

@@ -0,0 +1,104 @@
+# 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.
+
+"""Dual-arm coordinator blueprints with trajectory control.
+
+Usage:
+    dimos run coordinator-dual-mock      # Mock 7+6 DOF arms
+    dimos run coordinator-dual-xarm      # XArm7 left + XArm6 right
+    dimos run coordinator-piper-xarm     # XArm6 + Piper
+"""
+
+from __future__ import annotations
+
+from dimos.control.blueprints._hardware import mock_arm, piper, xarm6, xarm7
+from dimos.control.coordinator import TaskConfig, control_coordinator
+from dimos.core.transport import LCMTransport
+from dimos.msgs.sensor_msgs.JointState import JointState
+
+# Dual mock arms (7-DOF left, 6-DOF right)
+coordinator_dual_mock = control_coordinator(
+    hardware=[mock_arm("left_arm", 7), mock_arm("right_arm", 6)],
+    tasks=[
+        TaskConfig(
+            name="traj_left",
+            type="trajectory",
+            joint_names=[f"left_arm_joint{i + 1}" for i in range(7)],
+            priority=10,
+        ),
+        TaskConfig(
+            name="traj_right",
+            type="trajectory",
+            joint_names=[f"right_arm_joint{i + 1}" for i in range(6)],
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+# Dual XArm (XArm7 left, XArm6 right)
+coordinator_dual_xarm = control_coordinator(
+    hardware=[xarm7("left_arm"), xarm6("right_arm")],
+    tasks=[
+        TaskConfig(
+            name="traj_left",
+            type="trajectory",
+            joint_names=[f"left_arm_joint{i + 1}" for i in range(7)],
+            priority=10,
+        ),
+        TaskConfig(
+            name="traj_right",
+            type="trajectory",
+            joint_names=[f"right_arm_joint{i + 1}" for i in range(6)],
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+# Dual arm (XArm6 + Piper)
+coordinator_piper_xarm = control_coordinator(
+    hardware=[xarm6("xarm_arm"), piper("piper_arm")],
+    tasks=[
+        TaskConfig(
+            name="traj_xarm",
+            type="trajectory",
+            joint_names=[f"xarm_arm_joint{i + 1}" for i in range(6)],
+            priority=10,
+        ),
+        TaskConfig(
+            name="traj_piper",
+            type="trajectory",
+            joint_names=[f"piper_arm_joint{i + 1}" for i in range(6)],
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+    }
+)
+
+
+__all__ = [
+    "coordinator_dual_mock",
+    "coordinator_dual_xarm",
+    "coordinator_piper_xarm",
+]

dimos/control/blueprints/mobile.py

@@ -0,0 +1,79 @@
+# 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.
+
+"""Mobile manipulation coordinator blueprints.
+
+Usage:
+    dimos run coordinator-mock-twist-base      # Mock holonomic base
+    dimos run coordinator-mobile-manip-mock    # Mock arm + base
+"""
+
+from __future__ import annotations
+
+from dimos.control.blueprints._hardware import mock_arm, mock_twist_base
+from dimos.control.components import make_twist_base_joints
+from dimos.control.coordinator import TaskConfig, control_coordinator
+from dimos.core.transport import LCMTransport
+from dimos.msgs.geometry_msgs.Twist import Twist
+from dimos.msgs.sensor_msgs.JointState import JointState
+
+_base_joints = make_twist_base_joints("base")
+
+# Mock holonomic twist base (3-DOF: vx, vy, wz)
+coordinator_mock_twist_base = control_coordinator(
+    hardware=[mock_twist_base()],
+    tasks=[
+        TaskConfig(
+            name="vel_base",
+            type="velocity",
+            joint_names=_base_joints,
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+        ("twist_command", Twist): LCMTransport("/cmd_vel", Twist),
+    }
+)
+
+# Mock arm (7-DOF) + mock holonomic base (3-DOF)
+coordinator_mobile_manip_mock = control_coordinator(
+    hardware=[mock_arm(), mock_twist_base()],
+    tasks=[
+        TaskConfig(
+            name="traj_arm",
+            type="trajectory",
+            joint_names=[f"arm_joint{i + 1}" for i in range(7)],
+            priority=10,
+        ),
+        TaskConfig(
+            name="vel_base",
+            type="velocity",
+            joint_names=_base_joints,
+            priority=10,
+        ),
+    ],
+).transports(
+    {
+        ("joint_state", JointState): LCMTransport("/coordinator/joint_state", JointState),
+        ("twist_command", Twist): LCMTransport("/cmd_vel", Twist),
+    }
+)
+
+
+__all__ = [
+    "coordinator_mobile_manip_mock",
+    "coordinator_mock_twist_base",
+]