feat(manipulation): add heuristic grasp provider

[ruthwikdasyam]

Contribution path

• Small, focused change extracted from the pick-and-place exploration work.

Problem

Pick-and-place needs a lightweight grasp provider that does not require the optional GraspGenX runtime.

Solution

• Add HeuristicGraspModule, implementing the existing GraspGenSpec.
• Produce one deterministic top-down parallel-jaw proposal from a segmented object point cloud.
• Validate point-cloud metadata and geometry.
• Register the module and add focused unit coverage.

How to Test

can test from PRs above this.

AI assistance

OpenCode with gpt-5.6-terra assisted with implementation and validation. The author will review the public method descriptions before they are added.

Checklist

• I have read and approved the CLA.

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
4869	2	4867	87

View the full list of 2 ❄️ flaky test(s)

dimos.e2e_tests.test_manipulation_planning_groups::test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator

Flake rate in main: 100.00% (Passed 0 times, Failed 2 times)

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fda1e7e60>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7b5fd8c463e0>

    def test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan one generated plan over both arms and dispatch through one trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fda1e7e60>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7b5fd8c463e0>

dimos/e2e_tests/test_manipulation_planning_groups.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fda1e7e60>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7b5fd8c463e0>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7b5fd8c46980>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fda1e7e60>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7b5fd8c46980>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4354686.799883064
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7b5fd8c46980>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

dimos.e2e_tests.test_manipulation_planning_groups::test_single_arm_plans_and_executes_through_control_coordinator

Flake rate in main: 20.00% (Passed 12 times, Failed 3 times)

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fd897bf80>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7b5fd8c45e40>

    def test_single_arm_plans_and_executes_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan with one arm and execute through its trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fd897bf80>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7b5fd8c45e40>

dimos/e2e_tests/test_manipulation_planning_groups.py:159: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fd897bf80>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7b5fd8c45e40>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7b5fd8c460c0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7b5fd897bf80>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7b5fd8c460c0>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4354566.602233287
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7b5fd8c460c0>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

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

[greptile-apps]

Greptile Summary

Adds a deterministic heuristic grasp provider for segmented point clouds and registers it as a runnable robot module. The provider validates cloud metadata and geometry, generates a top-down grasp centered on the object, and aligns the jaw axis with the narrow planar dimension. Runtime checks confirmed the input-validation, pose/message, orientation, and registry paths.

Confidence Score: 5/5

Safe to merge: no blocking failure remains.

No blocking failure remains. Runtime validation exercised invalid point-cloud rejection, narrow-axis alignment for axis-aligned and oblique objects, pose serialization, and generated module resolution successfully.

T-Rex Logs

What T-Rex did

• The initial runtime capture showed that numpy was absent in the unusable recreated .venv.
• After installing only the focused runtime dependencies into the local .venv, the exact artifact script exited 0 and produced eight PASS observations.
• The source artifact is the exact executed script, and the after log captures the executed command, its working directory, and exit code.

_{Ran code and verified through T-Rex}

_{Reviews (2): Last reviewed commit: "test(manipulation): assert heuristic gra..." | Re-trigger Greptile}