refactor(manipulation): localize device connection configuration

dimos/control/connection.py

@@ -0,0 +1,115 @@
+# 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.
+
+"""Typed connection backends; importing schemas never loads robot implementations."""
+
+from typing import Annotated, Literal, Self
+
+from pydantic import AfterValidator, Field, model_validator
+
+from dimos.protocol.service.spec import BaseConfig
+
+
+def validate_address(value: str | None) -> str | None:
+    if value is not None and (not value.strip() or value != value.strip()):
+        raise ValueError("Device address must be nonempty and have no surrounding whitespace")
+    return value
+
+
+OptionalDeviceAddress = Annotated[str | None, AfterValidator(validate_address)]
+
+
+class SingleAddressConfig(BaseConfig):
+    address: OptionalDeviceAddress = None
+
+
+class XArmConnectionConfig(SingleAddressConfig):
+    # The hardware-only variant preserves blueprints without simulator wiring.
+    backend: Literal["xarm", "xarm_hardware"] = "xarm"
+    dof: Literal[6, 7] = 7
+
+
+class PiperConnectionConfig(SingleAddressConfig):
+    backend: Literal["piper", "piper_hardware"] = "piper"
+
+
+class A1ZConnectionConfig(SingleAddressConfig):
+    backend: Literal["a1z"] = "a1z"
+
+
+class A750ConnectionConfig(SingleAddressConfig):
+    backend: Literal["a750"] = "a750"
+
+
+class OpenYamConnectionConfig(SingleAddressConfig):
+    backend: Literal["openyam"] = "openyam"
+
+
+class DualXArmConnectionConfig(BaseConfig):
+    backend: Literal["dual_xarm"] = "dual_xarm"
+    left_address: OptionalDeviceAddress = None
+    right_address: OptionalDeviceAddress = None
+
+    @model_validator(mode="after")
+    def complete_pair(self) -> Self:
+        if (self.left_address is None) != (self.right_address is None):
+            raise ValueError("Supply both left and right addresses, or neither for mock hardware")
+        return self
+
+
+class PairedCanConfig(BaseConfig):
+    left_can_port: OptionalDeviceAddress = None
+    right_can_port: OptionalDeviceAddress = None
+
+    @model_validator(mode="after")
+    def complete_pair(self) -> Self:
+        if (self.left_can_port is None) != (self.right_can_port is None):
+            raise ValueError("Supply both left and right CAN ports, or neither for mock hardware")
+        if self.left_can_port is not None and self.left_can_port == self.right_can_port:
+            raise ValueError("Left and right CAN ports must be distinct")
+        return self
+
+
+class OpenArmConnectionConfig(PairedCanConfig):
+    backend: Literal["openarm"] = "openarm"
+
+
+class DualOpenYamConnectionConfig(PairedCanConfig):
+    backend: Literal["dual_openyam"] = "dual_openyam"
+
+
+class MixedArmConnectionConfig(BaseConfig):
+    backend: Literal["xarm_piper"] = "xarm_piper"
+    xarm_address: OptionalDeviceAddress = None
+    piper_address: OptionalDeviceAddress = None
+
+    @model_validator(mode="after")
+    def complete_pair(self) -> Self:
+        if (self.xarm_address is None) != (self.piper_address is None):
+            raise ValueError("Supply both xArm and Piper addresses, or neither for mock hardware")
+        return self
+
+
+HardwareConnectionConfig = Annotated[
+    XArmConnectionConfig
+    | PiperConnectionConfig
+    | A1ZConnectionConfig
+    | A750ConnectionConfig
+    | OpenYamConnectionConfig
+    | DualXArmConnectionConfig
+    | OpenArmConnectionConfig
+    | DualOpenYamConnectionConfig
+    | MixedArmConnectionConfig,
+    Field(discriminator="backend"),
+]

dimos/control/connection_factory.py

@@ -0,0 +1,147 @@
+# 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.
+
+"""Resolve connection descriptions without opening hardware.
+
+Robot factories are imported on demand, just like manipulation backends.
+"""
+
+from dataclasses import replace
+from importlib import import_module
+from typing import Any
+
+from dimos.control.components import HardwareComponent
+from dimos.control.connection import (
+    A1ZConnectionConfig,
+    A750ConnectionConfig,
+    DualOpenYamConnectionConfig,
+    DualXArmConnectionConfig,
+    HardwareConnectionConfig,
+    MixedArmConnectionConfig,
+    OpenArmConnectionConfig,
+    OpenYamConnectionConfig,
+    PiperConnectionConfig,
+    XArmConnectionConfig,
+)
+from dimos.hardware.whole_body.damiao.config import DamiaoRuntimeConfig
+
+
+def _hardware(robot: str, factory: str, **kwargs: Any) -> HardwareComponent:
+    module = import_module(f"dimos.robot.manipulators.{robot}.config")
+    result: HardwareComponent = getattr(module, factory)(**kwargs)
+    return result
+
+
+def _resolved_component(
+    component: HardwareComponent, resolved: HardwareComponent
+) -> HardwareComponent:
+    adapter_kwargs = {**resolved.adapter_kwargs, **component.adapter_kwargs}
+    runtime = resolved.adapter_kwargs.get("runtime_config")
+    if isinstance(runtime, DamiaoRuntimeConfig):
+        configured = component.adapter_kwargs.get("runtime_config", runtime)
+        adapter_kwargs["runtime_config"] = replace(configured, bus_devices=runtime.bus_devices)
+    return replace(
+        component,
+        adapter_type=resolved.adapter_type,
+        address=resolved.address,
+        limits=component.limits if resolved.adapter_type in ("mock", "mock_whole_body") else None,
+        adapter_kwargs=adapter_kwargs,
+    )
+
+
+def _xarm(
+    component: HardwareComponent, dof: int, address: str | None, simulation: str
+) -> HardwareComponent:
+    resolved = _hardware("xarm", f"xarm{dof}_hardware", address=address, simulation=simulation)
+    # Keep the blueprint's gripper and other adapter options.
+    return replace(
+        _resolved_component(component, resolved),
+        adapter_kwargs={**component.adapter_kwargs, "arm_dof": dof},
+    )
+
+
+def resolve_connection(
+    config: HardwareConnectionConfig,
+    hardware: list[HardwareComponent],
+    simulation: str,
+) -> list[HardwareComponent]:
+    """Select concrete hardware while preserving blueprint-owned descriptions."""
+    if isinstance(config, (OpenArmConnectionConfig, DualOpenYamConnectionConfig)):
+        if len(hardware) > 1:
+            raise ValueError("Coupled connection requires at most one hardware description")
+        robot = config.backend
+        resolved = _hardware(
+            robot,
+            f"{robot}_hardware",
+            left_can_port=config.left_can_port,
+            right_can_port=config.right_can_port,
+        )
+        return [_resolved_component(hardware[0], resolved)] if hardware else [resolved]
+
+    if isinstance(config, (DualXArmConnectionConfig, MixedArmConnectionConfig)):
+        if len(hardware) != 2:
+            raise ValueError("Dual-arm connection requires exactly two hardware descriptions")
+        left, right = hardware
+        if isinstance(config, DualXArmConnectionConfig):
+            return [
+                _xarm(left, 7, config.left_address, simulation),
+                _xarm(right, 6, config.right_address, simulation),
+            ]
+        return [
+            _xarm(left, 6, config.xarm_address, ""),
+            _resolved_component(
+                right, _hardware("piper", "piper_hardware", address=config.piper_address)
+            ),
+        ]
+
+    if len(hardware) != 1:
+        raise ValueError("Single-arm connection requires exactly one hardware description")
+    component = hardware[0]
+    if isinstance(config, XArmConnectionConfig):
+        return [
+            _xarm(
+                component,
+                config.dof,
+                config.address,
+                simulation if config.backend == "xarm" else "",
+            )
+        ]
+    if isinstance(config, PiperConnectionConfig):
+        resolved = _hardware(
+            "piper",
+            "piper_hardware",
+            address=config.address,
+            simulation=simulation if config.backend == "piper" else "",
+        )
+    elif isinstance(config, A1ZConnectionConfig):
+        # A1Z's blueprint already owns the gripper/dynamics configuration.
+        return [
+            replace(
+                component,
+                adapter_type="galaxea_a1z"
+                if config.address is not None and not simulation
+                else "mock",
+                address=config.address if not simulation else None,
+                limits=component.limits if simulation or config.address is None else None,
+            )
+        ]
+    elif isinstance(config, A750ConnectionConfig):
+        resolved = _hardware("a750", "a750_hardware", address=config.address)
+    elif isinstance(config, OpenYamConnectionConfig):
+        resolved = _hardware(
+            "openyam", "openyam_hardware", address=config.address, simulation=simulation
+        )
+    else:
+        raise TypeError(f"Unsupported connection config: {type(config).__name__}")
+    return [_resolved_component(component, resolved)]

dimos/control/coordinator.py

@@ -41,6 +41,8 @@
     TaskName,
     split_joint_name,
 )
+from dimos.control.connection import HardwareConnectionConfig
+from dimos.control.connection_factory import resolve_connection
 from dimos.control.hardware_interface import (
     ConnectedHardware,
     ConnectedTwistBase,
@@ -99,6 +101,7 @@ class ControlCoordinatorConfig(ModuleConfig):
     joint_state_frame_id: str = "coordinator"
     log_ticks: bool = False
     hardware: list[HardwareComponent] = field(default_factory=lambda: [])
+    connection: HardwareConnectionConfig | None = None
     tasks: list[TaskConfig] = field(default_factory=lambda: [])
 
 
@@ -204,6 +207,10 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
 
     def _setup_from_config(self) -> None:
         """Create hardware and tasks from config (called on start)."""
+        if self.config.connection is not None:
+            self.config.hardware = resolve_connection(
+                self.config.connection, self.config.hardware, self.config.g.simulation
+            )
         hardware_added: list[str] = []
         tasks_added: list[TaskName] = []
 
@@ -240,10 +247,16 @@ def _setup_hardware(self, component: HardwareComponent) -> None:
         else:
             adapter = self._create_adapter(component)
 
-        if not adapter.connect():
-            raise RuntimeError(f"Failed to connect to {component.adapter_type} adapter")
-
         try:
+            logger.info(
+                "Connecting hardware",
+                hardware_id=component.hardware_id,
+                adapter_type=component.adapter_type,
+                address=component.address,
+            )
+            if not adapter.connect():
+                raise RuntimeError(f"Failed to connect to {component.adapter_type} adapter")
+
             if component.auto_enable:
                 activate = getattr(adapter, "activate", None)
                 if callable(activate):
@@ -254,7 +267,10 @@ def _setup_hardware(self, component: HardwareComponent) -> None:
 
             self.add_hardware(adapter, component)
         except Exception:
-            adapter.disconnect()
+            try:
+                adapter.disconnect()
+            except Exception:
+                logger.exception("Hardware cleanup failed", hardware_id=component.hardware_id)
             raise
 
     def _create_adapter(self, component: HardwareComponent) -> ManipulatorAdapter: