feat: integrate GraspGenX grasp proposals

dimos/manipulation/demo_graspgenx/__main__.py

@@ -0,0 +1,48 @@
+# 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.
+
+"""Run GraspGenX once and save an annotated point-cloud PNG."""
+
+import argparse
+from collections.abc import Sequence
+import os
+from pathlib import Path
+
+from .demo import run_contributor_demo
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=Path(os.environ.get("DIMOS_GRASPGENX_OUTPUT", "graspgenx-ycb-demo.png")),
+        help="Destination PNG path.",
+    )
+    return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    result = run_contributor_demo(output_path=args.output)
+    print(
+        f"graspgenx-ycb-demo complete candidates={result.candidate_count} "
+        f"image={result.image_path}",
+        flush=True,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

dimos/manipulation/demo_graspgenx/demo.py

@@ -0,0 +1,150 @@
+# 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.
+
+"""One-shot GraspGenX inference and static image generation."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+import torch
+
+from dimos.manipulation.grasping.grasp_gen_spec import GraspGenSpec
+from dimos.manipulation.grasping.grasp_gen_x import (
+    GRASPGENX_MODEL_REPO,
+    GRASPGENX_MODEL_REVISION,
+    GraspGenXConfig,
+    GraspGenXModule,
+    SweepVolumeGripperConfig,
+)
+from dimos.msgs.manipulation_msgs.GraspCandidateArray import GraspCandidateArray
+from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
+
+from .fixture import load_demo_clouds
+from .render import SweepVolumeLike, render_grasp_image
+
+
+@dataclass(frozen=True)
+class DemoResult:
+    image_path: Path
+    scene_points: int
+    object_points: int
+    candidate_count: int
+    best_score: float
+    frame: str
+
+
+def deployment_config() -> GraspGenXConfig:
+    """Build the fixed sweep-volume deployment without downloading the checkpoint."""
+    return GraspGenXConfig(
+        gripper=SweepVolumeGripperConfig(
+            extents_open=(0.08, 0.045, 0.04),
+            offset_open=(0.0, 0.0, 0.135),
+            extents_half_open=(0.04, 0.045, 0.035),
+            offset_half_open=(0.0, 0.0, 0.118),
+            fingertip_depth=0.15,
+            family="revolute_3f",
+        ),
+        max_candidates=100,
+    )
+
+
+def _cuda_context() -> dict[str, object]:
+    available = bool(torch.cuda.is_available())
+    return {"available": available, "device": torch.cuda.get_device_name(0) if available else "cpu"}
+
+
+def _validate(result: GraspCandidateArray, object_cloud: PointCloud2) -> None:
+    if not result.candidates:
+        raise ValueError("proposer returned no grasp candidates")
+    if result.header.frame_id != "world":
+        raise ValueError("proposer returned a result outside world frame")
+    if (
+        result.header.frame_id != object_cloud.frame_id
+        or result.header.timestamp != object_cloud.ts
+    ):
+        raise ValueError("proposer changed the object point-cloud frame or timestamp")
+
+    scores = np.asarray([candidate.score for candidate in result.candidates], dtype=float)
+    if not np.all(np.isfinite(scores)) or np.any(scores[:-1] < scores[1:]):
+        raise ValueError("grasp scores must be finite and descending")
+    for candidate in result.candidates:
+        p, q = candidate.pose.position, candidate.pose.orientation
+        values = np.asarray([p.x, p.y, p.z, q.x, q.y, q.z, q.w], dtype=float)
+        if not np.all(np.isfinite(values)):
+            raise ValueError("proposer returned a non-finite TCP pose")
+
+
+def run_demo(
+    proposer: GraspGenSpec,
+    output_path: Path,
+    *,
+    gripper: SweepVolumeLike,
+    renderer: Callable[
+        [Path, PointCloud2, PointCloud2, GraspCandidateArray, SweepVolumeLike], Path
+    ] = render_grasp_image,
+) -> DemoResult:
+    """Load one scene, run inference once, render one PNG, and return."""
+    if not callable(getattr(proposer, "propose_grasps", None)):
+        raise TypeError("proposer must implement GraspGenSpec.propose_grasps")
+
+    scene, object_cloud = load_demo_clouds()
+    result = proposer.propose_grasps(object_cloud)
+    _validate(result, object_cloud)
+    image_path = renderer(output_path, scene, object_cloud, result, gripper)
+    best_score = float(result.candidates[0].score)
+    print(
+        "graspgenx-ycb-demo "
+        f"scene_points={len(scene)} object_points={len(object_cloud)} "
+        f"candidates={len(result)} best_score={best_score:.6f} "
+        f"image={image_path}",
+        flush=True,
+    )
+    return DemoResult(
+        image_path=image_path,
+        scene_points=len(scene),
+        object_points=len(object_cloud),
+        candidate_count=len(result),
+        best_score=best_score,
+        frame=result.header.frame_id,
+    )
+
+
+def run_contributor_demo(
+    *,
+    output_path: Path,
+    config: GraspGenXConfig | None = None,
+) -> DemoResult:
+    """Run the real adapter once and save its annotated point-cloud image."""
+    active_config = config if config is not None else deployment_config()
+    cuda = _cuda_context()
+    print(
+        "graspgenx-ycb-demo "
+        f"checkpoint=hf://{GRASPGENX_MODEL_REPO}@{GRASPGENX_MODEL_REVISION} "
+        f"cuda={cuda['available']} device={cuda['device']}",
+        flush=True,
+    )
+
+    module_args = active_config.model_dump(
+        exclude={"rpc_transport", "tf_transport", "g"},
+    )
+    adapter = GraspGenXModule(**module_args)
+    try:
+        adapter.start()
+        return run_demo(adapter, output_path, gripper=active_config.gripper)
+    finally:
+        adapter.stop()

dimos/manipulation/demo_graspgenx/fixture.py

@@ -0,0 +1,124 @@
+# 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.
+
+"""Deterministic YCB scene input loaded through the repository data system."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+import re
+from typing import TypedDict
+
+import numpy as np
+
+from dimos.msgs.sensor_msgs.PointCloud2 import PointCloud2
+from dimos.utils.data import get_data
+
+DATASET = "graspgenx_ycb_banana_scene"
+SCENE_FILE = "scene.npz"
+METADATA_FILE = "scene.json"
+BANANA_LABEL = 0
+
+
+class SceneMetadata(TypedDict, total=False):
+    counts: dict[str, int]
+    labels: dict[str, dict[str, str]]
+    frame: str
+    timestamp: float
+    final_sha256: str
+    source: dict[str, object]
+
+
+def _dataset_dir() -> Path:
+    return get_data(DATASET)
+
+
+def _validate_record(
+    path: Path,
+    points: np.ndarray,
+    labels: np.ndarray,
+    timestamp: float,
+    metadata: SceneMetadata,
+) -> None:
+    expected_sha = metadata.get("final_sha256")
+    if not isinstance(expected_sha, str) or not re.fullmatch(r"[0-9a-f]{64}", expected_sha):
+        raise ValueError("scene fixture metadata must contain a valid final SHA-256")
+    if hashlib.sha256(path.read_bytes()).hexdigest() != expected_sha:
+        raise ValueError("scene fixture bytes do not match provenance SHA-256")
+
+    expected_counts = {"banana": 3500, "table": 256, "distractor": 48, "total": 3804}
+    if metadata.get("counts") != expected_counts or expected_counts["total"] != len(points):
+        raise ValueError("scene fixture counts do not match bytes")
+    if (metadata.get("labels") or {}).get("encoding") != {
+        "0": "banana",
+        "1": "table",
+        "2": "distractor",
+    }:
+        raise ValueError("scene fixture label encoding metadata is invalid")
+
+    actual_counts = dict(zip(*np.unique(labels, return_counts=True), strict=True))
+    if {str(label): int(count) for label, count in actual_counts.items()} != {
+        "0": 3500,
+        "1": 256,
+        "2": 48,
+    }:
+        raise ValueError("scene fixture label values do not match metadata")
+    if metadata.get("frame") != "world" or timestamp != float(
+        metadata.get("timestamp", float("nan"))
+    ):
+        raise ValueError("scene fixture frame/timestamp metadata is invalid")
+
+    source = metadata.get("source")
+    if not isinstance(source, dict) or source.get("path") != (
+        "assets/sample_data/object_mesh/banana.obj"
+    ):
+        raise ValueError("scene fixture source OBJ provenance is incomplete")
+    source_sha = source.get("source_obj_sha256")
+    if not isinstance(source_sha, str) or not re.fullmatch(r"[0-9a-f]{64}", source_sha):
+        raise ValueError("scene fixture source OBJ must have its 64-character SHA-256")
+
+
+def load_scene_record(
+    dataset_dir: Path | None = None,
+) -> tuple[np.ndarray, np.ndarray, SceneMetadata]:
+    """Load stored XYZ points, semantic labels, and provenance metadata."""
+    root = dataset_dir if dataset_dir is not None else _dataset_dir()
+    scene_path = root / SCENE_FILE
+    metadata_path = root / METADATA_FILE
+    with np.load(scene_path, allow_pickle=False) as data:
+        points = np.asarray(data["points"], dtype=np.float32)
+        labels = np.asarray(data["labels"], dtype=np.uint8)
+        timestamp = float(np.asarray(data["timestamp"]).item())
+    if points.shape != (len(labels), 3) or not np.all(np.isfinite(points)):
+        raise ValueError("scene fixture has invalid points or labels")
+    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+    _validate_record(scene_path, points, labels, timestamp, metadata)
+    return points, labels, metadata
+
+
+def load_demo_clouds(
+    dataset_dir: Path | None = None,
+) -> tuple[PointCloud2, PointCloud2]:
+    """Return the complete scene and its labeled banana Object Point Cloud."""
+    points, labels, metadata = load_scene_record(dataset_dir)
+    timestamp = float(metadata["timestamp"])
+    scene = PointCloud2.from_numpy(points, frame_id="world", timestamp=timestamp)
+    object_cloud = PointCloud2.from_numpy(
+        points[labels == BANANA_LABEL],
+        frame_id="world",
+        timestamp=timestamp,
+    )
+    return scene, object_cloud