feat: in-house robot asset manager

dimos/control/blueprints/_hardware.py

@@ -23,9 +23,13 @@
     make_twist_base_joints,
 )
 from dimos.core.global_config import global_config
+from dimos.robot.assets.source import RobotDescriptionSource
 from dimos.utils.data import LfsPath
 
-PIPER_FK_MODEL = LfsPath("piper_description/mujoco_model/piper_no_gripper_description.xml")
+PIPER_DESCRIPTION_REPO = "https://github.com/agilexrobotics/agx_arm_urdf"
+_PIPER_REPO = RobotDescriptionSource(url=PIPER_DESCRIPTION_REPO, ref="main")
+
+PIPER_FK_MODEL = _PIPER_REPO / "piper" / "urdf" / "piper_description.urdf"
 XARM6_FK_MODEL = LfsPath("xarm_description/urdf/xarm6/xarm6.urdf")
 XARM7_FK_MODEL = LfsPath("xarm_description/urdf/xarm7/xarm7.urdf")
 A750_FK_MODEL = LfsPath("a750_description/urdf/a750_rev1_no_gripper.urdf")

dimos/control/examples/cartesian_ik_jogger.py

@@ -162,11 +162,10 @@ def clamp(value: float, min_val: float, max_val: float) -> float:
 
 
 def _get_piper_model_path() -> str:
-    """Get path to Piper MJCF model."""
-    from dimos.utils.data import get_data
+    """Get path to Piper FK model."""
+    from dimos.control.blueprints._hardware import PIPER_FK_MODEL
 
-    piper_path = get_data("piper_description")
-    return str(piper_path / "mujoco_model" / "piper_no_gripper_description.xml")
+    return str(PIPER_FK_MODEL)
 
 
 def run_jogger_ui(model_path: str | None = None, ee_joint_id: int = 6) -> None:

dimos/control/tasks/cartesian_ik_task/cartesian_ik_task.py

@@ -86,13 +86,12 @@ class CartesianIKTask(BaseControlTask):
     outputs JointCommandOutput and participates in joint-level arbitration.
 
     Example:
-        >>> from dimos.utils.data import get_data
-        >>> piper_path = get_data("piper_description")
+        >>> from dimos.control.blueprints._hardware import PIPER_FK_MODEL
         >>> task = CartesianIKTask(
         ...     name="cartesian_arm",
         ...     config=CartesianIKTaskConfig(
         ...         joint_names=["joint1", "joint2", "joint3", "joint4", "joint5", "joint6"],
-        ...         model_path=piper_path / "mujoco_model" / "piper_no_gripper_description.xml",
+        ...         model_path=PIPER_FK_MODEL,
         ...         ee_joint_id=6,
         ...         priority=10,
         ...         timeout=0.5,

dimos/control/tasks/teleop_task/teleop_task.py

@@ -96,13 +96,12 @@ class TeleopIKTask(BaseControlTask):
     Outputs JointCommandOutput and participates in joint-level arbitration.
 
     Example:
-        >>> from dimos.utils.data import get_data
-        >>> piper_path = get_data("piper_description")
+        >>> from dimos.control.blueprints._hardware import PIPER_FK_MODEL
         >>> task = TeleopIKTask(
         ...     name="teleop_arm",
         ...     config=TeleopIKTaskConfig(
         ...         joint_names=["joint1", "joint2", "joint3", "joint4", "joint5", "joint6"],
-        ...         model_path=piper_path / "mujoco_model" / "piper_no_gripper_description.xml",
+        ...         model_path=PIPER_FK_MODEL,
         ...         ee_joint_id=6,
         ...         priority=10,
         ...         timeout=0.5,

dimos/manipulation/blueprints.py

@@ -46,8 +46,8 @@
 from dimos.msgs.geometry_msgs.Transform import Transform
 from dimos.msgs.geometry_msgs.Vector3 import Vector3
 from dimos.perception.object_scene_registration import ObjectSceneRegistrationModule
+from dimos.robot.assets.source import RobotDescriptionSource
 from dimos.simulation.engines.mujoco_sim_module import MujocoSimModule
-from dimos.utils.data import LfsPath
 from dimos.visualization.rerun.bridge import RerunBridgeModule
 
 XARM_GRIPPER_COLLISION_EXCLUSIONS: list[tuple[str, str]] = [
@@ -69,8 +69,10 @@
     ("link6", "right_outer_knuckle"),
 ]
 
-_XARM_MODEL_PATH = LfsPath("xarm_description") / "urdf/xarm_device.urdf.xacro"
-_XARM_PACKAGE_PATHS: dict[str, Path] = {"xarm_description": LfsPath("xarm_description")}
+XARM_ROS2_REPO = "https://github.com/xArm-Developer/xarm_ros2"
+_XARM_REPO = RobotDescriptionSource(url=XARM_ROS2_REPO, ref="humble")
+_XARM_MODEL_PATH = _XARM_REPO / "xarm_description" / "urdf" / "xarm_device.urdf.xacro"
+_XARM_PACKAGE_PATHS: dict[str, Path] = {"xarm_description": _XARM_REPO / "xarm_description"}
 
 
 def _base_pose(x: float = 0.0, y: float = 0.0, z: float = 0.0) -> PoseStamped:

dimos/manipulation/planning/utils/mesh_utils.py

@@ -35,9 +35,9 @@
 from pathlib import Path
 import re
 import shutil
-import tempfile
 from typing import TYPE_CHECKING
 
+from dimos.robot.assets.processing import DERIVED_ASSET_CACHE_ROOT, render_urdf
 from dimos.utils.logging_config import setup_logger
 
 if TYPE_CHECKING:
@@ -46,8 +46,8 @@
 
 logger = setup_logger()
 
-# Cache directory for processed URDFs
-_CACHE_DIR = Path(tempfile.gettempdir()) / "dimos_urdf_cache"
+# Cache directory for Drake-specific URDFs derived from rendered robot assets.
+_CACHE_DIR = DERIVED_ASSET_CACHE_ROOT / "drake_urdfs"
 
 
 def prepare_urdf_for_drake(
@@ -72,33 +72,31 @@ def prepare_urdf_for_drake(
     Returns:
         Path to the prepared URDF file (may be cached)
     """
-    urdf_path = Path(urdf_path)
     package_paths = package_paths or {}
     xacro_args = xacro_args or {}
+    rendered_urdf = render_urdf(
+        urdf_path,
+        package_paths,
+        xacro_args,
+        package_uri_mode="absolute",
+    )
 
-    # Generate cache key
-    cache_key = _generate_cache_key(urdf_path, package_paths, xacro_args, convert_meshes)
-    cache_path = _CACHE_DIR / cache_key / urdf_path.stem
+    # Generate cache key for Drake-specific processing.
+    cache_key = _generate_cache_key(rendered_urdf, convert_meshes)
+    cache_path = _CACHE_DIR / cache_key / rendered_urdf.stem
     cache_path.mkdir(parents=True, exist_ok=True)
-    cached_urdf = cache_path / f"{urdf_path.stem}.urdf"
+    cached_urdf = cache_path / f"{rendered_urdf.stem}.urdf"
 
     # Check cache
     if cached_urdf.exists():
         logger.debug(f"Using cached URDF: {cached_urdf}")
         return str(cached_urdf)
 
-    # Process xacro if needed
-    if urdf_path.suffix in (".xacro", ".urdf.xacro"):
-        urdf_content = _process_xacro(urdf_path, package_paths, xacro_args)
-    else:
-        urdf_content = urdf_path.read_text()
+    urdf_content = rendered_urdf.read_text()
 
     # Strip transmission blocks (Drake doesn't need them, and they can cause issues)
     urdf_content = _strip_transmission_blocks(urdf_content)
 
-    # Resolve package:// URIs
-    urdf_content = _resolve_package_uris(urdf_content, package_paths, cache_path)
-
     # Convert meshes if requested
     if convert_meshes:
         urdf_content = _convert_meshes(urdf_content, cache_path)
@@ -112,11 +110,9 @@ def prepare_urdf_for_drake(
 
 def _generate_cache_key(
     urdf_path: Path,
-    package_paths: dict[str, Path],
-    xacro_args: dict[str, str],
     convert_meshes: bool,
 ) -> str:
-    """Generate a cache key for the URDF configuration.
+    """Generate a cache key for Drake-specific URDF processing.
 
     Includes a version number to invalidate cache when processing logic changes.
     """
@@ -125,29 +121,11 @@ def _generate_cache_key(
 
     # Version number to invalidate cache when processing logic changes
     # Increment this when adding new processing steps (e.g., stripping transmission blocks)
-    processing_version = "v2"
-
-    key_data = f"{processing_version}:{urdf_path}:{mtime}:{sorted(package_paths.items())}:{sorted(xacro_args.items())}:{convert_meshes}"
+    processing_version = "drake-urdf-v1"
+    key_data = f"{processing_version}:{urdf_path}:{mtime}:{convert_meshes}"
     return hashlib.md5(key_data.encode()).hexdigest()[:16]
 
 
-def _process_xacro(
-    xacro_path: Path,
-    package_paths: dict[str, Path],
-    xacro_args: dict[str, str],
-) -> str:
-    """Process xacro file to URDF."""
-    try:
-        from dimos.utils.ament_prefix import process_xacro
-    except ImportError:
-        raise ImportError(
-            "xacro is required for processing .xacro files. "
-            "Install the manipulation extra: pip install dimos[manipulation]"
-        )
-
-    return process_xacro(xacro_path, package_paths, xacro_args)
-
-
 def _strip_transmission_blocks(urdf_content: str) -> str:
     """Remove transmission blocks from URDF content.
 
@@ -175,36 +153,6 @@ def _strip_transmission_blocks(urdf_content: str) -> str:
     return result
 
 
-def _resolve_package_uris(
-    urdf_content: str,
-    package_paths: dict[str, Path],
-    output_dir: Path,
-) -> str:
-    """Resolve package:// URIs to filesystem paths."""
-    # Pattern for package:// URIs (handles both single and double quotes)
-    # Note: Use triple quotes so \s is correctly interpreted as whitespace, not literal 's'
-    pattern = r"""package://([^/]+)/(.+?)(["'<>\s])"""
-
-    def replace_uri(match: re.Match[str]) -> str:
-        pkg_name = match.group(1)
-        rel_path = match.group(2)
-        suffix = match.group(3)
-
-        if pkg_name in package_paths:
-            # Ensure absolute path for proper resolution
-            pkg_path = Path(package_paths[pkg_name]).resolve()
-            full_path = pkg_path / rel_path
-            if full_path.exists():
-                return f"{full_path}{suffix}"
-            else:
-                logger.warning(f"File not found: {full_path}")
-
-        # Return original if not found
-        return match.group(0)
-
-    return re.sub(pattern, replace_uri, urdf_content)
-
-
 def _convert_meshes(urdf_content: str, output_dir: Path) -> str:
     """Convert DAE/STL meshes to OBJ format for Drake compatibility."""
     try:

dimos/manipulation/planning/utils/test_mesh_utils.py

@@ -0,0 +1,52 @@
+# 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 pathlib import Path
+
+from dimos.manipulation.planning.utils import mesh_utils
+import dimos.robot.assets.processing as processing
+
+
+def test_prepare_urdf_for_drake_uses_rendered_urdf_and_keeps_drake_cleanup(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    monkeypatch.setattr(processing, "_RENDERED_URDF_CACHE_ROOT", tmp_path / "rendered")
+    monkeypatch.setattr(mesh_utils, "_CACHE_DIR", tmp_path / "drake")
+    package_root = tmp_path / "pkg"
+    mesh = package_root / "meshes" / "link.stl"
+    mesh.parent.mkdir(parents=True)
+    mesh.write_text("solid link\nendsolid link\n")
+    urdf = tmp_path / "robot.urdf"
+    urdf.write_text(
+        "<robot name='r'>"
+        "<link name='base'><visual><geometry>"
+        "<mesh filename='package://pkg/meshes/link.stl'/>"
+        "</geometry></visual></link>"
+        "<transmission name='ignore_me'><type>bad</type></transmission>"
+        "</robot>"
+    )
+
+    prepared = Path(
+        mesh_utils.prepare_urdf_for_drake(
+            urdf,
+            {"pkg": package_root},
+        )
+    )
+
+    prepared_text = prepared.read_text()
+    assert prepared.is_relative_to(tmp_path / "drake")
+    assert "package://" not in prepared_text
+    assert str(mesh) in prepared_text
+    assert "<transmission" not in prepared_text

dimos/manipulation/planning/world/drake_world.py

@@ -261,7 +261,7 @@ def _load_model(self, config: RobotModelConfig) -> Any:
             # Register package paths (not applicable to MJCF)
             if config.package_paths:
                 for pkg_name, pkg_path in config.package_paths.items():
-                    self._parser.package_map().Add(pkg_name, Path(pkg_path))
+                    self._parser.package_map().Add(pkg_name, pkg_path)
             else:
                 self._parser.package_map().Add(
                     f"{config.name}_description", prepared_path_obj.parent

dimos/manipulation/test_manipulation_module.py

@@ -26,6 +26,7 @@
 
 import pytest
 
+from dimos.manipulation.blueprints import _XARM_MODEL_PATH, _XARM_PACKAGE_PATHS
 from dimos.manipulation.manipulation_module import (
     ManipulationModule,
     ManipulationState,
@@ -36,7 +37,6 @@
 from dimos.msgs.geometry_msgs.Quaternion import Quaternion
 from dimos.msgs.geometry_msgs.Vector3 import Vector3
 from dimos.msgs.sensor_msgs.JointState import JointState
-from dimos.utils.data import get_data
 
 pytestmark = pytest.mark.self_hosted
 
@@ -47,24 +47,22 @@ def _drake_available() -> bool:
 
 def _xarm_urdf_available() -> bool:
     try:
-        desc_path = get_data("xarm_description")
-        model_path = desc_path / "urdf/xarm_device.urdf.xacro"
+        model_path = _XARM_MODEL_PATH
         return model_path.exists()
     except Exception:
         return False
 
 
 def _get_xarm7_config() -> RobotModelConfig:
     """Create XArm7 robot config for testing."""
-    desc_path = get_data("xarm_description")
     return RobotModelConfig(
         name="test_arm",
-        model_path=desc_path / "urdf/xarm_device.urdf.xacro",
+        model_path=_XARM_MODEL_PATH,
         base_pose=PoseStamped(position=Vector3(), orientation=Quaternion()),
         joint_names=["joint1", "joint2", "joint3", "joint4", "joint5", "joint6", "joint7"],
         end_effector_link="link7",
         base_link="link_base",
-        package_paths={"xarm_description": desc_path},
+        package_paths=_XARM_PACKAGE_PATHS,
         xacro_args={"dof": "7", "limited": "true"},
         auto_convert_meshes=True,
         max_velocity=1.0,