feat(manipulation): add xArm pick-and-place workflow

[ruthwikdasyam]

Contribution path

• Small, safe change that does not need a tracking issue

Problem

Provide an operator workflow for xArm6 perception-assisted pick and place, including safe table-aware execution and agent-driven box filling.

Solution

• Add xArm6 pick-and-place and agent blueprints with the explicit PnP console.
• Add prompted Moondream plus EdgeTAM object discovery, table estimation, collision setup, grasp verification, and above-rim box drops.
• Add GraspGenX proposal display and manual selection support.
• Preserve Rerun/Viser overlays for detected objects, table geometry, selected grasps, and open boxes.

How to Test

uv run --no-sync dimos run picknplace --daemon
uv run --no-sync python -m dimos.manipulation.pnpconsole

Validation run locally:

uv run --no-sync pytest dimos/manipulation/grasping/test_grasp_gen_x.py dimos/manipulation/test_picknplace.py dimos/manipulation/test_pnpconsole.py dimos/manipulation/test_roboplan.py dimos/manipulation/test_roboplan_integration.py dimos/robot/test_all_blueprints_generation.py
uv run --no-sync pre-commit run --all-files

AI assistance

OpenCode with gpt-5.6-terra assisted with implementation, merge-conflict resolution, and test execution. The author reviewed the resulting changes.

Checklist

• I have read and approved the CLA.

[codecov]

❌ 4 Tests Failed:

Tests completed	Failed	Passed	Skipped
3842	4	3838	175

View the top 3 failed test(s) by shortest run time

dimos.codebase_checks.test_inline_heavy_imports::test_heavy_imports_are_inline

Stack Traces | 1.81s run time

def test_heavy_imports_are_inline() -> None:
        """Fail if any file imports cv2/open3d/rerun at module level."""
        hits = find_eager_heavy_imports()
        if hits:
            listing = "\n".join(
                f"  - dimos/{f}:{line}: `{module}`"
                for f, lines in sorted(hits.items())
                for line, module in lines
            )
>           raise AssertionError(
                f"Found module-level import(s) of {'/'.join(HEAVY_MODULES)}:\n{listing}\n\n"
                "These libraries load large native extensions into every process that "
                "transitively imports the module. Import them inside the function or "
                "method that uses them; imports needed only for type annotations go "
                "under `if TYPE_CHECKING:`."
            )
E           AssertionError: Found module-level import(s) of cv2/open3d/rerun:
E             - .../manipulation/visualization/pose_overlay.py:17: `cv2`
E             - .../manipulation/visualization/rerun.py:20: `rerun.blueprint`
E           
E           These libraries load large native extensions into every process that transitively imports the module. Import them inside the function or method that uses them; imports needed only for type annotations go under `if TYPE_CHECKING:`.

hits       = {'manipulation/visualization/pose_overlay.py': [(17, 'cv2')], 'manipulation/visualization/rerun.py': [(20, 'rerun.blueprint')]}
listing    = '  - .../manipulation/visualization/pose_overlay.py:17: `cv2`\n  - .../manipulation/visualization/rerun.py:20: `rerun.blueprint`'

dimos/codebase_checks/test_inline_heavy_imports.py:92: AssertionError

dimos.codebase_checks.test_no_dunder_new::test_no_dunder_new

Stack Traces | 2.43s run time

def test_no_dunder_new() -> None:
        """Fail if any test file calls `__new__` to bypass `__init__`."""
        dimos_dir = DIMOS_PROJECT_ROOT / "dimos"
        hits = find_dunder_new_calls()
        if hits:
            listing = "\n".join(
                f"  - {p.relative_to(dimos_dir)}:{lineno}: {line.strip()}" for p, lineno, line in hits
            )
>           raise AssertionError(
                f"Found __new__ call(s) in test files:\n{listing}\n\n"
                "Tests must construct objects with the real constructor: __init__ is "
                "code under test too, and an object assembled by hand silently rots "
                "when the constructor changes. If __init__ does heavy work, mock the "
                "collaborators it needs instead of skipping it. Only if that is truly "
                "impossible, add the call to the WHITELIST in "
                "dimos/codebase_checks/test_no_dunder_new.py."
            )
E           AssertionError: Found __new__ call(s) in test files:
E             - manipulation/test_table_collision.py:23: module = object.__new__(ManipulationModule)
E           
E           Tests must construct objects with the real constructor: __init__ is code under test too, and an object assembled by hand silently rots when the constructor changes. If __init__ does heavy work, mock the collaborators it needs instead of skipping it. Only if that is truly impossible, add the call to the WHITELIST in dimos/codebase_checks/test_no_dunder_new.py.

dimos_dir  = PosixPath('.../dimos/dimos/dimos')
hits       = [(PosixPath('.../dimos/dimos/dimos/manipulation/test_table_collision.py'), 23, '    module = object.__new__(ManipulationModule)')]
listing    = '  - manipulation/test_table_collision.py:23: module = object.__new__(ManipulationModule)'

dimos/codebase_checks/test_no_dunder_new.py:67: AssertionError

dimos.codebase_checks.test_import_from_source::test_import_from_source

Stack Traces | 8.46s run time

def test_import_from_source() -> None:
        """Fail if any name is imported from a module that only re-imported it."""
        violations = find_reexport_imports()
        if violations:
            listing = "\n".join(
                f"  - {p.relative_to(DIMOS_PROJECT_ROOT)}:{line}: `{name}` imported from "
                f"{src}, but defined in {origin}"
                for p, line, name, src, origin in sorted(violations)
            )
>           raise AssertionError(
                f"Found import(s) that pull a name from a re-exporter:\n{listing}\n\n"
                "Import each name straight from the module that defines it (shown above). "
                "If a module re-exports a name on purpose, mark its import with "
                "`# noqa: F401` or the `from x import Y as Y` form, and that re-export "
                "will be allowed."
            )
E           AssertionError: Found import(s) that pull a name from a re-exporter:
E             - dimos/manipulation/test_picknplace.py:30: `IKStatus` imported from dimos.manipulation.planning.spec.models, but defined in dimos.manipulation.planning.spec.enums
E           
E           Import each name straight from the module that defines it (shown above). If a module re-exports a name on purpose, mark its import with `# noqa: F401` or the `from x import Y as Y` form, and that re-export will be allowed.

listing    = '  - dimos/manipulation/test_picknplace.py:30: `IKStatus` imported from dimos.manipulation.planning.spec.models, but defined in dimos.manipulation.planning.spec.enums'
violations = [(PosixPath('.../dimos/manipulation/test_picknplace.py'), 30, 'IKStatus', 'dimos.manipulation.planning.spec.models', 'dimos.manipulation.planning.spec.enums')]

dimos/codebase_checks/test_import_from_source.py:148: AssertionError

dimos.manipulation.test_picknplace::test_picknplace_blueprint_accepts_short_backend_and_grasp_options

Stack Traces | 15.7s run time

def test_picknplace_blueprint_accepts_short_backend_and_grasp_options() -> None:
        config = BlueprintConfigParser(picknplace)
    
>       options = config.parse(
            overrides={"osr": {"det": "moondream", "seg": "edgetam"}, "pnp": {"grasp": "graspgenx"}}
        )

config     = <dimos.core.coordination.blueprint_config.parser.BlueprintConfigParser object at 0xff7bfd9e49b0>

dimos/manipulation/test_picknplace.py:221: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../coordination/blueprint_config/parser.py:168: in parse
    key: plain(value)
        cli_tokens = ()
        config_path = None
        environ    = None
        global_overrides = None
        overrides  = {'osr': {'det': 'moondream', 'seg': 'edgetam'}, 'pnp': {'grasp': 'graspgenx'}}
        schema     = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] ParserSchema object at 0xff7c0d9f38c0>
        self       = <dimos.core.coordination.blueprint_config.parser.BlueprintConfigParser object at 0xff7bfd9e49b0>
.../coordination/blueprint_config/values.py:50: in plain
    return [plain(item) for item in value]
        value      = [<[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attem..._description.tar.gz']' returned non-zero exit status 1.") raised in repr()] RobotModelConfig object at 0xff7c5e7df340>]
.../coordination/blueprint_config/values.py:46: in plain
    return {key: plain(item) for key, item in value.model_dump(mode="python").items()}
        value      = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp...m_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] RobotModelConfig object at 0xff7c5e7df340>
.../coordination/blueprint_config/values.py:45: in plain
    if isinstance(value, BaseModel):
        value      = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff7c5e8525d0>
dimos/utils/data.py:364: in __getattribute__
    resolved = object.__getattribute__(self, "_ensure_downloaded")()
        name       = '__class__'
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff7c5e8525d0>
dimos/utils/data.py:347: in _ensure_downloaded
    cache = get_data(filename)
        cache      = None
        filename   = 'xarm_description/urdf/xarm_device.urdf.xacro'
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff7c5e8525d0>
dimos/utils/data.py:304: in get_data
    archive_path = _decompress_archive(_pull_lfs_archive(archive_name))
        archive_name = 'xarm_description'
        data_dir   = PosixPath('.../dimos/dimos/data')
        file_path  = PosixPath('.../dimos/dimos/data/xarm_description/urdf/xarm_device.urdf.xacro')
        name       = 'xarm_description/urdf/xarm_device.urdf.xacro'
        nested_path = PosixPath('urdf/xarm_device.urdf.xacro')
        path_parts = ('xarm_description', 'urdf', 'xarm_device.urdf.xacro')
dimos/utils/data.py:248: in _pull_lfs_archive
    _lfs_pull(file_path, repo_root)
        file_path  = PosixPath('.../dimos/data/.lfs/xarm_description.tar.gz')
        filename   = 'xarm_description'
        repo_root  = PosixPath('.../work/dimos/dimos')
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

file_path = PosixPath('.../dimos/data/.lfs/xarm_description.tar.gz')
repo_root = PosixPath('.../work/dimos/dimos')

    def _lfs_pull(file_path: Path, repo_root: Path, *, retries: int = 2) -> None:
        relative_path = file_path.relative_to(repo_root)
    
        env = os.environ.copy()
        env["GIT_LFS_FORCE_PROGRESS"] = "1"
    
        last_err: subprocess.CalledProcessError | None = None
        for attempt in range(1, retries + 2):  # retries + 1 total attempts
            try:
                subprocess.run(
                    ["git", "lfs", "pull", "--include", str(relative_path)],
                    cwd=repo_root,
                    check=True,
                    env=env,
                )
                return
            except subprocess.CalledProcessError as e:
                last_err = e
                if attempt <= retries:
                    time.sleep(attempt)  # 1s, 2s backoff
    
>       raise RuntimeError(
            f"Failed to pull LFS file {file_path} after {retries + 1} attempts: {last_err}"
        )
E       RuntimeError: Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attempts: Command '['git', 'lfs', 'pull', '--include', 'data/.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.

attempt    = 3
env        = {'ACCEPT_EULA': 'Y', 'ACTIONS_ID_TOKEN_REQUEST_TOKEN': 'eyJhbGciOiJSUzI1NiIsImtpZCI6IjM4ODI2YjE3LTZhMzAtNWY5Yi1iMTY5LT...-version=2.0', 'ACTIONS_ORCHESTRATION_ID': 'af57cd1c-8f06-4476-8d9f-53cffdaf46b9.tests.ubuntu-24_04-arm_3_14_fal', ...}
file_path  = PosixPath('.../dimos/data/.lfs/xarm_description.tar.gz')
last_err   = CalledProcessError(1, ['git', 'lfs', 'pull', '--include', 'data/.lfs/xarm_description.tar.gz'])
relative_path = PosixPath('data/.lfs/xarm_description.tar.gz')
repo_root  = PosixPath('.../work/dimos/dimos')
retries    = 2

dimos/utils/data.py:216: RuntimeError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}