From 869eb2ab0e18ee663d6cd95a5727ed6e03ab579f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BD=A0=E7=9A=84=E5=90=8D=E5=AD=97?= <你的邮箱>
Date: Sat, 18 Jul 2026 04:34:17 +0800
Subject: [PATCH 1/5] feat(input): add MANUS ROS 2 source adapter

---
 src/somehand/runtime/manus_source.py | 265 +++++++++++++++++++++++++++
 tests/ros2_manus_source_probe.py     | 141 ++++++++++++++
 tests/test_manus_source.py           | 114 ++++++++++++
 3 files changed, 520 insertions(+)
 create mode 100644 src/somehand/runtime/manus_source.py
 create mode 100644 tests/ros2_manus_source_probe.py
 create mode 100644 tests/test_manus_source.py

diff --git a/src/somehand/runtime/manus_source.py b/src/somehand/runtime/manus_source.py
new file mode 100644
index 0000000..2305b59
--- /dev/null
+++ b/src/somehand/runtime/manus_source.py
@@ -0,0 +1,265 @@
+"""MANUS ROS 2 input adapter for somehand."""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+import numpy as np
+
+from somehand.core import HandFrame, SourceFrame, normalize_hand_side
+
+
+# MANUS 25 节点 → MediaPipe 风格 21 点。
+#
+# 0: wrist
+# 1-4: thumb
+# 5-8: index
+# 9-12: middle
+# 13-16: ring
+# 17-20: pinky
+# MANUS 25-node skeleton -> MediaPipe-style 21 landmarks.
+#
+# MANUS semantics:
+#   thumb: MCP, PIP, IP, TIP
+#   other fingers: MCP, PIP, IP, DIP, TIP
+#
+# MediaPipe-style output keeps four joints per finger.  For the four
+# non-thumb fingers, MANUS IP is the additional intermediate point and is
+# intentionally omitted.
+LANDMARK_KEYS = [
+    ("thumb", "mcp"),
+    ("thumb", "pip"),
+    ("thumb", "ip"),
+    ("thumb", "tip"),
+
+    ("index", "mcp"),
+    ("index", "pip"),
+    ("index", "dip"),
+    ("index", "tip"),
+
+    ("middle", "mcp"),
+    ("middle", "pip"),
+    ("middle", "dip"),
+    ("middle", "tip"),
+
+    ("ring", "mcp"),
+    ("ring", "pip"),
+    ("ring", "dip"),
+    ("ring", "tip"),
+
+    ("pinky", "mcp"),
+    ("pinky", "pip"),
+    ("pinky", "dip"),
+    ("pinky", "tip"),
+]
+
+
+def _normalize(value: str) -> str:
+    return str(value).strip().lower()
+
+
+def manus_message_to_hand_frame(msg: Any) -> HandFrame:
+    """Convert a ManusGlove ROS 2 message to a somehand HandFrame."""
+
+    if len(msg.raw_nodes) != msg.raw_node_count:
+        raise ValueError(
+            f"raw_node_count={msg.raw_node_count}, "
+            f"actual={len(msg.raw_nodes)}"
+        )
+
+    nodes_by_id = {
+        int(node.node_id): node
+        for node in msg.raw_nodes
+    }
+
+    if 0 not in nodes_by_id:
+        raise ValueError("MANUS message does not contain wrist node_id=0")
+
+    nodes_by_type = {
+        (
+            _normalize(node.chain_type),
+            _normalize(node.joint_type),
+        ): node
+        for node in msg.raw_nodes
+    }
+
+    ordered_nodes = [nodes_by_id[0]]
+    missing: list[str] = []
+
+    for chain_name, joint_name in LANDMARK_KEYS:
+        node = nodes_by_type.get((chain_name, joint_name))
+
+        if node is None:
+            missing.append(f"{chain_name}/{joint_name}")
+        else:
+            ordered_nodes.append(node)
+
+    if missing:
+        raise ValueError(
+            "Missing MANUS nodes: " + ", ".join(missing)
+        )
+
+    landmarks = np.asarray(
+        [
+            [
+                node.pose.position.x,
+                node.pose.position.y,
+                node.pose.position.z,
+            ]
+            for node in ordered_nodes
+        ],
+        dtype=np.float64,
+    )
+
+    if landmarks.shape != (21, 3):
+        raise ValueError(
+            f"Expected landmarks shape (21, 3), got {landmarks.shape}"
+        )
+
+    if not np.all(np.isfinite(landmarks)):
+        raise ValueError("MANUS landmarks contain NaN or infinity")
+
+    return HandFrame(
+        landmarks_3d=landmarks,
+        landmarks_2d=None,
+        hand_side=normalize_hand_side(msg.side),
+    )
+
+
+class ManusRos2InputSource:
+    """Read MANUS glove frames from a ROS 2 topic."""
+
+    def __init__(
+        self,
+        *,
+        topic: str,
+        hand_side: str,
+        timeout: float = 2.0,
+        nominal_fps: int = 120,
+    ) -> None:
+        # Lazy import: users running webcam mode should not need ROS 2.
+        try:
+            import rclpy
+            from manus_ros2_msgs.msg import ManusGlove
+            from rclpy.qos import qos_profile_sensor_data
+        except ImportError as exc:
+            raise RuntimeError(
+                "MANUS input requires ROS 2 Humble, rclpy and "
+                "manus_ros2_msgs. Source the ROS environments first."
+            ) from exc
+
+        self.source_desc = f"ros2://{topic}"
+        self.hand_side = normalize_hand_side(hand_side)
+        self.timeout = float(timeout)
+        self._fps = int(nominal_fps)
+
+        self._rclpy = rclpy
+        self._latest_msg = None
+        self._available = True
+        self._received_count = 0
+        self._converted_count = 0
+
+        self._owns_context = not rclpy.ok()
+
+        if self._owns_context:
+            rclpy.init(args=None)
+
+        self._node = rclpy.create_node(
+            f"somehand_manus_input_{self.hand_side}"
+        )
+
+        self._subscription = self._node.create_subscription(
+            ManusGlove,
+            topic,
+            self._message_callback,
+            qos_profile_sensor_data,
+        )
+
+    @property
+    def fps(self) -> int:
+        return self._fps
+
+    def _message_callback(self, msg: Any) -> None:
+        self._received_count += 1
+
+        try:
+            message_side = normalize_hand_side(msg.side)
+        except ValueError:
+            return
+
+        if message_side != self.hand_side:
+            return
+
+        # Only preserve the latest frame to avoid queue buildup.
+        self._latest_msg = msg
+
+    def is_available(self) -> bool:
+        return (
+            self._available
+            and self._rclpy.ok()
+        )
+
+    def get_frame(self) -> SourceFrame:
+        if not self.is_available():
+            raise StopIteration
+
+        deadline = time.monotonic() + self.timeout
+
+        while self._latest_msg is None:
+            if not self._rclpy.ok():
+                self._available = False
+                raise StopIteration
+
+            remaining = deadline - time.monotonic()
+
+            if remaining <= 0.0:
+                # No glove frame during this interval. Keep session alive.
+                return SourceFrame(detection=None)
+
+            self._rclpy.spin_once(
+                self._node,
+                timeout_sec=min(0.05, remaining),
+            )
+
+        msg = self._latest_msg
+        self._latest_msg = None
+
+        frame = manus_message_to_hand_frame(msg)
+        self._converted_count += 1
+
+        return SourceFrame(detection=frame)
+
+    def reset(self) -> bool:
+        return False
+
+    def close(self) -> None:
+        if not self._available:
+            return
+
+        self._available = False
+
+        try:
+            self._node.destroy_node()
+        finally:
+            if self._owns_context and self._rclpy.ok():
+                self._rclpy.shutdown()
+
+    def stats_snapshot(self) -> dict[str, object]:
+        return {
+            "messages_received": self._received_count,
+            "frames_converted": self._converted_count,
+        }
+
+
+def create_manus_ros2_source(
+    *,
+    topic: str,
+    hand_side: str,
+    timeout: float,
+) -> ManusRos2InputSource:
+    return ManusRos2InputSource(
+        topic=topic,
+        hand_side=hand_side,
+        timeout=timeout,
+    )
diff --git a/tests/ros2_manus_source_probe.py b/tests/ros2_manus_source_probe.py
new file mode 100644
index 0000000..6463bf1
--- /dev/null
+++ b/tests/ros2_manus_source_probe.py
@@ -0,0 +1,141 @@
+"""Offline ROS 2 integration probe for ManusRos2InputSource."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from typing import Any
+
+import numpy as np
+
+from somehand.runtime.manus_source import ManusRos2InputSource
+
+
+def _side_text(value: Any) -> str:
+    """Normalize a string or enum-like hand side for diagnostics."""
+    raw = getattr(value, "value", value)
+    return str(raw).strip().lower()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--topic", required=True)
+    parser.add_argument("--side", choices=("left", "right"), required=True)
+    parser.add_argument("--frames", type=int, default=12)
+    parser.add_argument("--frame-timeout", type=float, default=1.0)
+    parser.add_argument("--overall-timeout", type=float, default=15.0)
+    args = parser.parse_args()
+
+    source = ManusRos2InputSource(
+        topic=args.topic,
+        hand_side=args.side,
+        timeout=args.frame_timeout,
+        nominal_fps=30,
+    )
+
+    landmarks_history: list[np.ndarray] = []
+    observed_sides: list[str] = []
+    deadline = time.monotonic() + args.overall_timeout
+
+    try:
+        while (
+            len(landmarks_history) < args.frames
+            and time.monotonic() < deadline
+        ):
+            source_frame = source.get_frame()
+            detection = source_frame.detection
+
+            if detection is None:
+                continue
+
+            landmarks = np.asarray(
+                detection.landmarks_3d,
+                dtype=np.float64,
+            )
+
+            if landmarks.shape != (21, 3):
+                raise RuntimeError(
+                    f"Expected landmarks shape (21, 3), "
+                    f"got {landmarks.shape}"
+                )
+
+            if not np.all(np.isfinite(landmarks)):
+                raise RuntimeError(
+                    "Received landmarks containing NaN or infinity"
+                )
+
+            actual_side = _side_text(detection.hand_side)
+
+            if actual_side != args.side:
+                raise RuntimeError(
+                    f"Expected side={args.side}, got {actual_side}"
+                )
+
+            landmarks_history.append(landmarks.copy())
+            observed_sides.append(actual_side)
+
+        stats = source.stats_snapshot()
+
+    finally:
+        source.close()
+
+    if len(landmarks_history) != args.frames:
+        raise RuntimeError(
+            f"Expected {args.frames} converted frames, "
+            f"got {len(landmarks_history)}"
+        )
+
+    stacked = np.stack(landmarks_history, axis=0)
+
+    motion_span = float(
+        np.max(
+            np.max(stacked, axis=0)
+            - np.min(stacked, axis=0)
+        )
+    )
+
+    if motion_span <= 1e-8:
+        raise RuntimeError(
+            f"Synthetic sequence did not move: "
+            f"motion_span={motion_span}"
+        )
+
+    converted = int(stats["frames_converted"])
+    received = int(stats["messages_received"])
+
+    if converted < args.frames:
+        raise RuntimeError(
+            f"frames_converted={converted}, expected at least {args.frames}"
+        )
+
+    if received < converted:
+        raise RuntimeError(
+            f"messages_received={received} is less than "
+            f"frames_converted={converted}"
+        )
+
+    result = {
+        "status": "PASS",
+        "topic": args.topic,
+        "expected_side": args.side,
+        "observed_sides": sorted(set(observed_sides)),
+        "requested_frames": args.frames,
+        "received_frames": len(landmarks_history),
+        "landmark_shape": list(stacked.shape[1:]),
+        "motion_span": motion_span,
+        "first_wrist": stacked[0, 0].tolist(),
+        "first_thumb_tip": stacked[0, 4].tolist(),
+        "first_index_mcp": stacked[0, 5].tolist(),
+        "stats": stats,
+    }
+
+    print(
+        "PHASE1E_MANUS_SOURCE_PROBE="
+        + json.dumps(result, sort_keys=True)
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_manus_source.py b/tests/test_manus_source.py
new file mode 100644
index 0000000..986aa53
--- /dev/null
+++ b/tests/test_manus_source.py
@@ -0,0 +1,114 @@
+"""Unit tests for the MANUS ROS 2 input conversion."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+from somehand.runtime.manus_source import manus_message_to_hand_frame
+
+
+MANUS_SPECS = [
+    ("Hand", "Invalid"),
+    ("Thumb", "MCP"),
+    ("Thumb", "PIP"),
+    ("Thumb", "IP"),
+    ("Thumb", "TIP"),
+    ("Index", "MCP"),
+    ("Index", "PIP"),
+    ("Index", "IP"),
+    ("Index", "DIP"),
+    ("Index", "TIP"),
+    ("Middle", "MCP"),
+    ("Middle", "PIP"),
+    ("Middle", "IP"),
+    ("Middle", "DIP"),
+    ("Middle", "TIP"),
+    ("Ring", "MCP"),
+    ("Ring", "PIP"),
+    ("Ring", "IP"),
+    ("Ring", "DIP"),
+    ("Ring", "TIP"),
+    ("Pinky", "MCP"),
+    ("Pinky", "PIP"),
+    ("Pinky", "IP"),
+    ("Pinky", "DIP"),
+    ("Pinky", "TIP"),
+]
+
+EXPECTED_NODE_IDS = [
+    0,
+    1, 2, 3, 4,
+    5, 6, 8, 9,
+    10, 11, 13, 14,
+    15, 16, 18, 19,
+    20, 21, 23, 24,
+]
+
+
+def _node(node_id: int, chain: str, joint: str):
+    return SimpleNamespace(
+        node_id=node_id,
+        chain_type=chain,
+        joint_type=joint,
+        pose=SimpleNamespace(
+            position=SimpleNamespace(
+                x=float(node_id),
+                y=float(node_id) + 0.1,
+                z=float(node_id) + 0.2,
+            )
+        ),
+    )
+
+
+def _message(side: str = "Right"):
+    nodes = [
+        _node(node_id, chain, joint)
+        for node_id, (chain, joint) in enumerate(MANUS_SPECS)
+    ]
+
+    return SimpleNamespace(
+        side=side,
+        raw_nodes=nodes,
+        raw_node_count=len(nodes),
+    )
+
+
+def test_complete_manus_message_maps_to_media_pipe_order():
+    frame = manus_message_to_hand_frame(_message())
+
+    assert frame.landmarks_3d.shape == (21, 3)
+    assert np.all(np.isfinite(frame.landmarks_3d))
+
+    np.testing.assert_allclose(
+        frame.landmarks_3d[:, 0],
+        np.asarray(EXPECTED_NODE_IDS, dtype=np.float64),
+    )
+
+
+def test_thumb_uses_ip_not_nonexistent_dip():
+    msg = _message()
+
+    # Remove the real Thumb/IP semantic and replace it with Thumb/DIP.
+    msg.raw_nodes[3].joint_type = "DIP"
+
+    with pytest.raises(ValueError, match=r"thumb/ip"):
+        manus_message_to_hand_frame(msg)
+
+
+def test_raw_node_count_mismatch_is_rejected():
+    msg = _message()
+    msg.raw_node_count += 1
+
+    with pytest.raises(ValueError, match="raw_node_count"):
+        manus_message_to_hand_frame(msg)
+
+
+def test_nan_landmark_is_rejected():
+    msg = _message()
+    msg.raw_nodes[8].pose.position.x = float("nan")
+
+    with pytest.raises(ValueError, match="NaN or infinity"):
+        manus_message_to_hand_frame(msg)

From a2e6fda022da2e322f22a1604381360501d953af Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BD=A0=E7=9A=84=E5=90=8D=E5=AD=97?= <你的邮箱>
Date: Sat, 18 Jul 2026 05:43:20 +0800
Subject: [PATCH 2/5] feat(cli): add MANUS ROS 2 input command

---
 README.md                        |  17 +++++
 src/somehand/cli/commands.py     |  51 ++++++++++++++
 src/somehand/cli/main.py         |   3 +
 src/somehand/cli/parser.py       |  30 +++++++-
 src/somehand/runtime/__init__.py |  10 +++
 tests/test_cli.py                | 117 +++++++++++++++++++++++++++++++
 tests/test_lazy_imports.py       |  25 +++++++
 tests/test_runtime_imports.py    |   1 +
 8 files changed, 251 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index d8c0d50..e58743c 100644
--- a/README.md
+++ b/README.md
@@ -66,6 +66,23 @@ python scripts/setup/download_assets.py --only mjcf mediapipe examples
 somehand replay --recording recordings/pico_right.pkl
 ```
 
+Use a MANUS ROS 2 glove topic with the safe viewer backend:
+
+```bash
+source /opt/ros/humble/setup.bash
+source ~/manus_ros2_ws/install/local_setup.bash
+
+somehand manus-ros2 \
+    --topic /manus_glove_0 \
+    --hand left \
+    --backend viewer \
+    --config configs/retargeting/left/revo2_left.yaml
+```
+
+The topic number does not define left or right. The adapter also validates the
+`side` field inside every `ManusGlove` message. Real hardware output remains an
+explicit `--backend real` choice.
+
 ## API Quick Start
 
 See [API Usage](docs/en/api.md) for stable imports, one-step retargeting, and session orchestration.
diff --git a/src/somehand/cli/commands.py b/src/somehand/cli/commands.py
index 4bee32d..a2ca5f3 100644
--- a/src/somehand/cli/commands.py
+++ b/src/somehand/cli/commands.py
@@ -15,6 +15,7 @@
     create_bihand_pico_source,
     create_bihand_recording_source,
     create_hc_mocap_udp_source,
+    create_manus_ros2_source,
     create_pico_source,
     create_recording_source,
     save_bihand_recording_artifact,
@@ -256,6 +257,56 @@ def _run_pico(args: argparse.Namespace) -> None:
     _finalize_run(args, summary=summary, source=source)
 
 
+def _run_manus_ros2(args: argparse.Namespace) -> None:
+    source, recording_controller = _wrap_source_for_interactive_recording(
+        _wrap_live_hand_source(
+            create_manus_ros2_source(
+                topic=args.topic,
+                hand_side=args.hand,
+                timeout=args.manus_timeout,
+            ),
+            args=args,
+        ),
+        record_output_path=args.record_output,
+    )
+    engine = _build_engine(args, input_type="manus_ros2")
+    session = _build_runtime_session(
+        engine,
+        args,
+        visualize=True,
+        show_preview=False,
+        key_callback=None if recording_controller is None else recording_controller.handle_keypress,
+    )
+    extra_lines = [
+        f"Backend: {args.backend}",
+        f"Signal sampling: {source.fps} fps",
+        f"MANUS ROS 2 topic: {args.topic}",
+        f"Expected message side: {display_hand_side(args.hand)}",
+        "Topic numbering does not define handedness; message.side is validated.",
+    ]
+    if recording_controller is not None:
+        extra_lines.append("Press 'r' in the terminal or robot-hand viewer to start recording.")
+        extra_lines.append("Press 's' in the terminal or robot-hand viewer to stop recording, save, and exit.")
+    _print_startup(
+        engine,
+        source_desc=source.source_desc,
+        tracking_desc=f"Tracking MANUS hand: {display_hand_side(args.hand)} | Source fps: {source.fps}",
+        extra_lines=extra_lines,
+    )
+    if recording_controller is not None:
+        recording_controller.start()
+    try:
+        summary = session.run(
+            source,
+            input_type="manus_ros2",
+            stop_condition=None if recording_controller is None else (lambda: recording_controller.stop_requested),
+        )
+    finally:
+        if recording_controller is not None:
+            recording_controller.close()
+    _finalize_run(args, summary=summary, source=source)
+
+
 def _run_hc_mocap_udp(args: argparse.Namespace) -> None:
     source = _wrap_source_for_recording(
         _wrap_live_hand_source(
diff --git a/src/somehand/cli/main.py b/src/somehand/cli/main.py
index b8cc112..62f922a 100644
--- a/src/somehand/cli/main.py
+++ b/src/somehand/cli/main.py
@@ -59,6 +59,9 @@ def main(argv: list[str] | None = None) -> None:
             return
         commands._run_pico(args)
         return
+    if args.command == "manus-ros2":
+        commands._run_manus_ros2(args)
+        return
     if args.command == "hc-mocap":
         if args.hand == "both":
             if args.backend != "viewer":
diff --git a/src/somehand/cli/parser.py b/src/somehand/cli/parser.py
index 8196ec7..0807574 100644
--- a/src/somehand/cli/parser.py
+++ b/src/somehand/cli/parser.py
@@ -22,20 +22,26 @@ def parse_hand_selector(value: str) -> str:
     return normalize_hand_side(value)
 
 
-def add_common_args(parser: argparse.ArgumentParser) -> None:
+def add_common_args(parser: argparse.ArgumentParser, *, allow_both: bool = True) -> None:
     parser.add_argument(
         "-c",
         "--config",
         default=str(DEFAULT_CONFIG_PATH),
         help="Path to retargeting config YAML",
     )
+    hand_choices = ["left", "right", "both"] if allow_both else ["left", "right"]
+    hand_help = (
+        "Hand side for the current channel, or 'both' for two-hand mode"
+        if allow_both
+        else "Expected hand side for the selected input topic"
+    )
     parser.add_argument(
         "-H",
         "--hand",
         type=parse_hand_selector,
-        choices=["left", "right", "both"],
+        choices=hand_choices,
         default="right",
-        help="Hand side for the current channel, or 'both' for two-hand mode",
+        help=hand_help,
     )
     parser.add_argument(
         "--record-output",
@@ -144,6 +150,24 @@ def build_parser() -> argparse.ArgumentParser:
         help="Timeout in seconds while waiting for PICO Bridge hand-tracking frames",
     )
 
+    manus_ros2 = subparsers.add_parser(
+        "manus-ros2",
+        help="Retarget from a MANUS glove ROS 2 topic",
+    )
+    add_common_args(manus_ros2, allow_both=False)
+    add_live_sampling_args(manus_ros2)
+    manus_ros2.add_argument(
+        "--topic",
+        required=True,
+        help="MANUS ManusGlove topic, for example /manus_glove_0",
+    )
+    manus_ros2.add_argument(
+        "--manus-timeout",
+        type=float,
+        default=2.0,
+        help="Seconds to wait for the next matching MANUS frame",
+    )
+
     hc_mocap = subparsers.add_parser("hc-mocap", help="Retarget from a live hc_mocap UDP stream")
     add_common_args(hc_mocap)
     add_live_sampling_args(hc_mocap)
diff --git a/src/somehand/runtime/__init__.py b/src/somehand/runtime/__init__.py
index bbc3a3a..afbcb59 100644
--- a/src/somehand/runtime/__init__.py
+++ b/src/somehand/runtime/__init__.py
@@ -6,6 +6,12 @@
 
 from .config_validation import validate_runtime_bihand_config, validate_runtime_retargeting_config
 
+_MANUS_EXPORTS = {
+    "ManusRos2InputSource",
+    "create_manus_ros2_source",
+    "manus_message_to_hand_frame",
+}
+
 _INFRA_EXPORTS = {
     "AsyncBiHandLandmarkOutputSink",
     "AsyncLandmarkOutputSink",
@@ -50,6 +56,7 @@
 
 __all__ = sorted(
     _INFRA_EXPORTS
+    | _MANUS_EXPORTS
     | {
         "validate_runtime_bihand_config",
         "validate_runtime_retargeting_config",
@@ -62,6 +69,9 @@ def __getattr__(name: str):
         return validate_runtime_bihand_config
     if name == "validate_runtime_retargeting_config":
         return validate_runtime_retargeting_config
+    if name in _MANUS_EXPORTS:
+        manus_source = import_module("somehand.runtime.manus_source")
+        return getattr(manus_source, name)
     if name in _INFRA_EXPORTS:
         infrastructure = import_module("somehand.infrastructure")
         return getattr(infrastructure, name)
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 434ea3d..9ae7094 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -475,3 +475,120 @@ def test_webcam_command_uses_default_camera():
 
     assert args.command == "webcam"
     assert args.camera == 0
+
+
+def test_manus_ros2_command_uses_safe_defaults():
+    parser = build_parser()
+    args = parser.parse_args(
+        ["manus-ros2", "--topic", "/manus_glove_1", "--hand", "right"]
+    )
+
+    assert args.command == "manus-ros2"
+    assert args.topic == "/manus_glove_1"
+    assert args.hand == "right"
+    assert args.backend == "viewer"
+    assert args.manus_timeout == 2.0
+    assert args.signal_fps is None
+
+
+def test_manus_ros2_command_requires_topic():
+    parser = build_parser()
+
+    with pytest.raises(SystemExit):
+        parser.parse_args(["manus-ros2", "--hand", "left"])
+
+
+def test_manus_ros2_command_rejects_both_hand_selector():
+    parser = build_parser()
+
+    with pytest.raises(SystemExit):
+        parser.parse_args(
+            ["manus-ros2", "--topic", "/manus_glove_0", "--hand", "both"]
+        )
+
+
+def test_manus_ros2_dispatches_to_command_handler(monkeypatch):
+    called = []
+
+    monkeypatch.setattr(
+        cli_module,
+        "_run_manus_ros2",
+        lambda args: called.append((args.topic, args.hand, args.backend)),
+    )
+
+    cli_main_module.main(
+        ["manus-ros2", "--topic", "/manus_glove_0", "--hand", "left"]
+    )
+
+    assert called == [("/manus_glove_0", "left", "viewer")]
+
+
+def test_run_manus_ros2_builds_live_source_and_session(monkeypatch):
+    calls = {}
+
+    class _FakeSource:
+        source_desc = "ros2:///manus_glove_0"
+        fps = 30
+
+    class _FakeSession:
+        def run(self, source, **kwargs):
+            calls["run_source"] = source
+            calls["run_kwargs"] = kwargs
+            return SimpleNamespace(
+                num_frames=12,
+                num_detected=12,
+                source_desc=source.source_desc,
+                input_type="manus_ros2",
+            )
+
+    def _fake_create_manus_ros2_source(**kwargs):
+        calls["source_kwargs"] = kwargs
+        return _FakeSource()
+
+    def _fake_build_engine(args, **kwargs):
+        calls["engine_kwargs"] = kwargs
+        return SimpleNamespace(
+            describe=lambda: {"model_name": "revo2", "dof": 6, "vector_pairs": 5}
+        )
+
+    def _fake_build_runtime_session(engine, args, **kwargs):
+        calls["session_kwargs"] = kwargs
+        return _FakeSession()
+
+    monkeypatch.setattr(cli_module, "create_manus_ros2_source", _fake_create_manus_ros2_source)
+    monkeypatch.setattr(cli_module, "_wrap_live_hand_source", lambda source, **kwargs: source)
+    monkeypatch.setattr(
+        cli_module,
+        "_wrap_source_for_interactive_recording",
+        lambda source, **kwargs: (source, None),
+    )
+    monkeypatch.setattr(cli_module, "_build_engine", _fake_build_engine)
+    monkeypatch.setattr(cli_module, "_build_runtime_session", _fake_build_runtime_session)
+    monkeypatch.setattr(cli_module, "_print_startup", lambda *args, **kwargs: None)
+    monkeypatch.setattr(cli_module, "_finalize_run", lambda *args, **kwargs: None)
+
+    args = SimpleNamespace(
+        topic="/manus_glove_0",
+        hand="left",
+        manus_timeout=1.5,
+        signal_fps=30,
+        record_output=None,
+        backend="viewer",
+        config="unused.yaml",
+    )
+
+    cli_module._run_manus_ros2(args)
+
+    assert calls["source_kwargs"] == {
+        "topic": "/manus_glove_0",
+        "hand_side": "left",
+        "timeout": 1.5,
+    }
+    assert calls["engine_kwargs"] == {"input_type": "manus_ros2"}
+    assert calls["session_kwargs"] == {
+        "visualize": True,
+        "show_preview": False,
+        "key_callback": None,
+    }
+    assert calls["run_kwargs"]["input_type"] == "manus_ros2"
+    assert calls["run_kwargs"]["stop_condition"] is None
diff --git a/tests/test_lazy_imports.py b/tests/test_lazy_imports.py
index bfd5cef..791301b 100644
--- a/tests/test_lazy_imports.py
+++ b/tests/test_lazy_imports.py
@@ -39,3 +39,28 @@ def test_pico_input_import_does_not_import_cv2():
 
 def test_pico_source_adapter_import_does_not_import_cv2():
     _assert_import_does_not_import_cv2("somehand.runtime.source_adapters")
+
+
+def test_manus_source_import_does_not_import_rclpy():
+    env = dict(os.environ)
+    env["PYTHONPATH"] = str(_SRC)
+    script = """
+import builtins
+import importlib
+
+original_import = builtins.__import__
+
+def guarded_import(name, globals=None, locals=None, fromlist=(), level=0):
+    if name == "rclpy" or name.startswith("rclpy."):
+        raise AssertionError("manus_source should import rclpy lazily")
+    return original_import(name, globals, locals, fromlist, level)
+
+builtins.__import__ = guarded_import
+importlib.import_module("somehand.runtime.manus_source")
+"""
+    subprocess.run(
+        [sys.executable, "-c", script],
+        cwd=_REPO_ROOT,
+        env=env,
+        check=True,
+    )
diff --git a/tests/test_runtime_imports.py b/tests/test_runtime_imports.py
index 2177f3e..d4119c2 100644
--- a/tests/test_runtime_imports.py
+++ b/tests/test_runtime_imports.py
@@ -14,6 +14,7 @@
     [
         "somehand.runtime.sink_outputs",
         "somehand.runtime.sink_rendering",
+        "somehand.runtime.manus_source",
         "somehand.runtime.source_adapters",
         "somehand.runtime.source_recording",
         "somehand.runtime.source_sampling",

From 0e65e857040693bfafe481cfa939cff113a1e51e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BD=A0=E7=9A=84=E5=90=8D=E5=AD=97?= <你的邮箱>
Date: Mon, 20 Jul 2026 12:44:41 +0800
Subject: [PATCH 3/5] fix(manus): support MetaGlove input and stop on stream
 timeout

---
 src/somehand/runtime/manus_source.py | 91 +++++++++++++++++++++-----
 tests/test_manus_source.py           | 98 ++++++++++++++++++++++++++--
 tests/test_source_sampling.py        | 37 +++++++++++
 3 files changed, 203 insertions(+), 23 deletions(-)

diff --git a/src/somehand/runtime/manus_source.py b/src/somehand/runtime/manus_source.py
index 2305b59..1016b6c 100644
--- a/src/somehand/runtime/manus_source.py
+++ b/src/somehand/runtime/manus_source.py
@@ -10,24 +10,23 @@
 from somehand.core import HandFrame, SourceFrame, normalize_hand_side
 
 
-# MANUS 25 节点 → MediaPipe 风格 21 点。
-#
-# 0: wrist
-# 1-4: thumb
-# 5-8: index
-# 9-12: middle
-# 13-16: ring
-# 17-20: pinky
 # MANUS 25-node skeleton -> MediaPipe-style 21 landmarks.
 #
-# MANUS semantics:
+# Two skeleton layouts are supported.
+#
+# Legacy/synthetic layout:
 #   thumb: MCP, PIP, IP, TIP
 #   other fingers: MCP, PIP, IP, DIP, TIP
 #
-# MediaPipe-style output keeps four joints per finger.  For the four
-# non-thumb fingers, MANUS IP is the additional intermediate point and is
-# intentionally omitted.
-LANDMARK_KEYS = [
+# Real MetaGlove layout observed from hardware:
+#   thumb: MCP, PIP, DIP, TIP
+#   other fingers: MCP, PIP, IP, DIP, TIP
+#
+# For the real MetaGlove, MANUS includes an additional metacarpal
+# point between the wrist and the anatomical finger MCP landmark.
+# Therefore the MediaPipe-style four points for index/middle/ring/
+# pinky are selected as PIP, IP, DIP, TIP.
+LEGACY_LANDMARK_KEYS = [
     ("thumb", "mcp"),
     ("thumb", "pip"),
     ("thumb", "ip"),
@@ -55,6 +54,34 @@
 ]
 
 
+REAL_METAGLOVE_LANDMARK_KEYS = [
+    ("thumb", "mcp"),
+    ("thumb", "pip"),
+    ("thumb", "dip"),
+    ("thumb", "tip"),
+
+    ("index", "pip"),
+    ("index", "ip"),
+    ("index", "dip"),
+    ("index", "tip"),
+
+    ("middle", "pip"),
+    ("middle", "ip"),
+    ("middle", "dip"),
+    ("middle", "tip"),
+
+    ("ring", "pip"),
+    ("ring", "ip"),
+    ("ring", "dip"),
+    ("ring", "tip"),
+
+    ("pinky", "pip"),
+    ("pinky", "ip"),
+    ("pinky", "dip"),
+    ("pinky", "tip"),
+]
+
+
 def _normalize(value: str) -> str:
     return str(value).strip().lower()
 
@@ -84,14 +111,29 @@ def manus_message_to_hand_frame(msg: Any) -> HandFrame:
         for node in msg.raw_nodes
     }
 
+    is_real_metaglove = (
+        ("thumb", "dip") in nodes_by_type
+        and ("thumb", "ip") not in nodes_by_type
+    )
+
+    landmark_keys = (
+        REAL_METAGLOVE_LANDMARK_KEYS
+        if is_real_metaglove
+        else LEGACY_LANDMARK_KEYS
+    )
+
     ordered_nodes = [nodes_by_id[0]]
     missing: list[str] = []
 
-    for chain_name, joint_name in LANDMARK_KEYS:
-        node = nodes_by_type.get((chain_name, joint_name))
+    for chain_name, joint_name in landmark_keys:
+        node = nodes_by_type.get(
+            (chain_name, joint_name)
+        )
 
         if node is None:
-            missing.append(f"{chain_name}/{joint_name}")
+            missing.append(
+                f"{chain_name}/{joint_name}"
+            )
         else:
             ordered_nodes.append(node)
 
@@ -159,6 +201,7 @@ def __init__(
         self._available = True
         self._received_count = 0
         self._converted_count = 0
+        self._timeout_count = 0
 
         self._owns_context = not rclpy.ok()
 
@@ -214,8 +257,19 @@ def get_frame(self) -> SourceFrame:
             remaining = deadline - time.monotonic()
 
             if remaining <= 0.0:
-                # No glove frame during this interval. Keep session alive.
-                return SourceFrame(detection=None)
+                self._timeout_count += 1
+
+                print(
+                    "MANUS input timeout: "
+                    f"no matching {self.hand_side} frame received "
+                    f"from {self.source_desc} for "
+                    f"{self.timeout:.3f}s; stopping session.",
+                    flush=True,
+                )
+
+                # A live robot must never continue using the last
+                # command after the MANUS stream disappears.
+                raise StopIteration
 
             self._rclpy.spin_once(
                 self._node,
@@ -249,6 +303,7 @@ def stats_snapshot(self) -> dict[str, object]:
         return {
             "messages_received": self._received_count,
             "frames_converted": self._converted_count,
+            "timeouts": self._timeout_count,
         }
 
 
diff --git a/tests/test_manus_source.py b/tests/test_manus_source.py
index 986aa53..15327e9 100644
--- a/tests/test_manus_source.py
+++ b/tests/test_manus_source.py
@@ -7,7 +7,7 @@
 import numpy as np
 import pytest
 
-from somehand.runtime.manus_source import manus_message_to_hand_frame
+from somehand.runtime.manus_source import ManusRos2InputSource, manus_message_to_hand_frame
 
 
 MANUS_SPECS = [
@@ -88,14 +88,19 @@ def test_complete_manus_message_maps_to_media_pipe_order():
     )
 
 
-def test_thumb_uses_ip_not_nonexistent_dip():
+def test_real_thumb_dip_is_accepted_as_thumb_ip_landmark():
     msg = _message()
 
-    # Remove the real Thumb/IP semantic and replace it with Thumb/DIP.
+    # Real MetaGlove messages use Thumb/DIP where the synthetic fixture
+    # historically used Thumb/IP.
     msg.raw_nodes[3].joint_type = "DIP"
 
-    with pytest.raises(ValueError, match=r"thumb/ip"):
-        manus_message_to_hand_frame(msg)
+    frame = manus_message_to_hand_frame(msg)
+
+    np.testing.assert_allclose(
+        frame.landmarks_3d[3],
+        np.asarray([3.0, 3.1, 3.2]),
+    )
 
 
 def test_raw_node_count_mismatch_is_rejected():
@@ -112,3 +117,86 @@ def test_nan_landmark_is_rejected():
 
     with pytest.raises(ValueError, match="NaN or infinity"):
         manus_message_to_hand_frame(msg)
+
+
+def test_real_metaglove_shifts_non_thumb_landmarks():
+    msg = _message()
+
+    # Real MetaGlove uses Thumb/DIP instead of the legacy Thumb/IP.
+    msg.raw_nodes[3].joint_type = "DIP"
+
+    frame = manus_message_to_hand_frame(msg)
+
+    def node_position(node_index):
+        position = msg.raw_nodes[node_index].pose.position
+        return np.asarray(
+            [
+                position.x,
+                position.y,
+                position.z,
+            ],
+            dtype=np.float64,
+        )
+
+    # Real MetaGlove non-thumb layout:
+    # MCP, PIP, IP, DIP, TIP
+    #
+    # MediaPipe output:
+    # MCP, PIP, DIP, TIP
+    #
+    # Therefore the first MANUS MCP point is omitted.
+    expected_raw_indices = [
+        6, 7, 8, 9,        # index
+        11, 12, 13, 14,    # middle
+        16, 17, 18, 19,    # ring
+        21, 22, 23, 24,    # pinky
+    ]
+
+    for output_index, raw_index in enumerate(
+        expected_raw_indices,
+        start=5,
+    ):
+        np.testing.assert_allclose(
+            frame.landmarks_3d[output_index],
+            node_position(raw_index),
+        )
+
+
+
+class _FakeRclpy:
+    def ok(self) -> bool:
+        return True
+
+    def spin_once(
+        self,
+        node,
+        *,
+        timeout_sec: float,
+    ) -> None:
+        return None
+
+
+def test_manus_source_timeout_raises_stop_iteration(
+    capsys,
+):
+    source = ManusRos2InputSource.__new__(
+        ManusRos2InputSource
+    )
+
+    source.source_desc = "ros2:///manus_glove_0"
+    source.hand_side = "right"
+    source.timeout = 0.0
+    source._rclpy = _FakeRclpy()
+    source._node = object()
+    source._latest_msg = None
+    source._available = True
+    source._timeout_count = 0
+
+    with pytest.raises(StopIteration):
+        source.get_frame()
+
+    captured = capsys.readouterr()
+
+    assert "MANUS input timeout" in captured.out
+    assert source._timeout_count == 1
+    assert source._available is True
diff --git a/tests/test_source_sampling.py b/tests/test_source_sampling.py
index 6e00fcc..5f12711 100644
--- a/tests/test_source_sampling.py
+++ b/tests/test_source_sampling.py
@@ -2,6 +2,7 @@
 from pathlib import Path
 
 import numpy as np
+import pytest
 
 sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
 
@@ -122,3 +123,39 @@ def test_fixed_rate_bihand_source_uses_requested_sample_fps():
     assert frame.right is not None
     np.testing.assert_allclose(frame.left.landmarks_3d, 2.0)
     np.testing.assert_allclose(frame.right.landmarks_3d, 4.0)
+
+
+
+class _StoppingHandSource:
+    source_desc = "fake://stopping-hand"
+
+    @property
+    def fps(self) -> int:
+        return 120
+
+    def is_available(self) -> bool:
+        return True
+
+    def get_frame(self):
+        raise StopIteration
+
+    def reset(self) -> bool:
+        return False
+
+    def close(self) -> None:
+        return None
+
+    def stats_snapshot(self):
+        return {}
+
+
+def test_fixed_rate_hand_source_propagates_stop_iteration():
+    source = _StoppingHandSource()
+
+    wrapped = FixedRateHandTrackingSource(
+        source,
+        sample_fps=30,
+    )
+
+    with pytest.raises(StopIteration):
+        wrapped.get_frame()

From 0d86b5c48e0ed48dadcd5a130e899cb81ae0aa0f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BD=A0=E7=9A=84=E5=90=8D=E5=AD=97?= <你的邮箱>
Date: Mon, 20 Jul 2026 13:47:28 +0800
Subject: [PATCH 4/5] feat(manus): add calibrated Revo2 viewer mapping

---
 src/somehand/application/manus_calibration.py | 587 ++++++++++++++++++
 src/somehand/cli/commands.py                  |  15 +
 src/somehand/cli/parser.py                    |   9 +
 src/somehand/cli/runtime.py                   |  26 +-
 tests/test_manus_calibration.py               | 166 +++++
 tests/test_manus_pinky_metric.py              |  79 +++
 6 files changed, 881 insertions(+), 1 deletion(-)
 create mode 100644 src/somehand/application/manus_calibration.py
 create mode 100644 tests/test_manus_calibration.py
 create mode 100644 tests/test_manus_pinky_metric.py

diff --git a/src/somehand/application/manus_calibration.py b/src/somehand/application/manus_calibration.py
new file mode 100644
index 0000000..ee24f3d
--- /dev/null
+++ b/src/somehand/application/manus_calibration.py
@@ -0,0 +1,587 @@
+"""Calibrated MANUS finger-curl mapping for Revo2.
+
+The default retargeting engine remains unchanged. This module is enabled only
+when the MANUS CLI receives an explicit calibration profile.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import math
+from pathlib import Path
+from typing import Mapping
+
+import numpy as np
+
+from somehand.domain import HandFrame, RetargetingStepResult
+from somehand.domain.hand_side import normalize_hand_side
+from somehand.infrastructure.config_loader import load_retargeting_config
+
+from .engine import RetargetingEngine
+
+
+FINGER_INDICES = {
+    "thumb": (1, 2, 3, 4),
+    "index": (5, 6, 7, 8),
+    "middle": (9, 10, 11, 12),
+    "ring": (13, 14, 15, 16),
+    "pinky": (17, 18, 19, 20),
+}
+
+FINGER_NAMES = tuple(FINGER_INDICES)
+FOUR_FINGER_NAMES = ("index", "middle", "ring", "pinky")
+
+DEFAULT_DEADBANDS = {
+    "thumb": 0.22,
+    "index": 0.08,
+    "middle": 0.08,
+    "ring": 0.08,
+    "pinky": 0.14,
+}
+
+
+@dataclass(frozen=True, slots=True)
+class FingerCalibration:
+    open_deg: float
+    comfortable_closed_deg: float
+    metric: str = "current"
+
+    @property
+    def span_deg(self) -> float:
+        return self.comfortable_closed_deg - self.open_deg
+
+
+@dataclass(frozen=True, slots=True)
+class ManusCalibrationProfile:
+    version: int
+    side: str
+    fingers: Mapping[str, FingerCalibration]
+
+
+def _finite_float(value: object, *, label: str) -> float:
+    result = float(value)
+    if not math.isfinite(result):
+        raise ValueError(f"{label} must be finite")
+    return result
+
+
+def load_manus_calibration_profile(
+    path: str | Path,
+    *,
+    expected_side: str | None = None,
+) -> ManusCalibrationProfile:
+    profile_path = Path(path).expanduser().resolve()
+    if not profile_path.is_file():
+        raise FileNotFoundError(
+            f"MANUS calibration file not found: {profile_path}"
+        )
+
+    data = json.loads(profile_path.read_text(encoding="utf-8"))
+    version = int(data.get("version", 1))
+    if version != 1:
+        raise ValueError(
+            f"Unsupported MANUS calibration version: {version}"
+        )
+
+    side = normalize_hand_side(data.get("side", ""))
+    if expected_side is not None:
+        normalized_expected = normalize_hand_side(expected_side)
+        if side != normalized_expected:
+            raise ValueError(
+                "MANUS calibration side "
+                f"{side!r} does not match requested side "
+                f"{normalized_expected!r}"
+            )
+
+    raw_fingers = data.get("fingers")
+    if not isinstance(raw_fingers, dict):
+        raise ValueError(
+            "MANUS calibration must contain a 'fingers' object"
+        )
+
+    missing = [
+        name for name in FINGER_NAMES
+        if name not in raw_fingers
+    ]
+    extra = sorted(set(raw_fingers) - set(FINGER_NAMES))
+    if missing or extra:
+        details = []
+        if missing:
+            details.append("missing=" + ",".join(missing))
+        if extra:
+            details.append("extra=" + ",".join(extra))
+        raise ValueError(
+            "Invalid MANUS calibration fingers: "
+            + "; ".join(details)
+        )
+
+    fingers: dict[str, FingerCalibration] = {}
+    for name in FINGER_NAMES:
+        item = raw_fingers[name]
+        if not isinstance(item, dict):
+            raise ValueError(
+                f"Calibration for finger {name!r} must be an object"
+            )
+        metric = str(
+            item.get("metric", "current")
+        ).strip().lower()
+
+        if metric not in {
+            "current",
+            "joint_sum",
+        }:
+            raise ValueError(
+                f"Unsupported calibration metric for "
+                f"{name!r}: {metric!r}"
+            )
+
+        calibration = FingerCalibration(
+            open_deg=_finite_float(
+                item["open_deg"],
+                label=f"{name}.open_deg",
+            ),
+            comfortable_closed_deg=_finite_float(
+                item["comfortable_closed_deg"],
+                label=f"{name}.comfortable_closed_deg",
+            ),
+            metric=metric,
+        )
+        if calibration.span_deg <= 1e-6:
+            raise ValueError(
+                f"Calibration span for {name!r} must be positive"
+            )
+        fingers[name] = calibration
+
+    return ManusCalibrationProfile(
+        version=version,
+        side=side,
+        fingers=fingers,
+    )
+
+
+def unit(vector: np.ndarray) -> np.ndarray:
+    norm = float(np.linalg.norm(vector))
+    if norm < 1e-12:
+        return np.zeros(3, dtype=np.float64)
+    return vector / norm
+
+
+def _vector_angle_degrees(
+    first: np.ndarray,
+    second: np.ndarray,
+) -> float:
+    first_unit = unit(first)
+    second_unit = unit(second)
+
+    cosine = float(
+        np.clip(
+            np.dot(first_unit, second_unit),
+            -1.0,
+            1.0,
+        )
+    )
+
+    return float(
+        np.degrees(
+            np.arccos(cosine)
+        )
+    )
+
+
+def finger_curl_degrees(
+    landmarks: np.ndarray,
+    indices: tuple[int, int, int, int],
+) -> float:
+    mcp, pip, dip, tip = indices
+
+    proximal = landmarks[pip] - landmarks[mcp]
+    distal = landmarks[tip] - landmarks[dip]
+
+    return _vector_angle_degrees(
+        proximal,
+        distal,
+    )
+
+
+def finger_joint_sum_degrees(
+    landmarks: np.ndarray,
+    indices: tuple[int, int, int, int],
+) -> float:
+    """Sum MCP, PIP and DIP flexion angles."""
+
+    mcp, pip, dip, tip = indices
+
+    palm_segment = landmarks[mcp] - landmarks[0]
+    proximal_segment = landmarks[pip] - landmarks[mcp]
+    middle_segment = landmarks[dip] - landmarks[pip]
+    distal_segment = landmarks[tip] - landmarks[dip]
+
+    return float(
+        _vector_angle_degrees(
+            palm_segment,
+            proximal_segment,
+        )
+        + _vector_angle_degrees(
+            proximal_segment,
+            middle_segment,
+        )
+        + _vector_angle_degrees(
+            middle_segment,
+            distal_segment,
+        )
+    )
+
+
+def apply_deadband(value: float, deadband: float) -> float:
+    normalized = float(np.clip(value, 0.0, 1.0))
+    if not 0.0 <= deadband < 1.0:
+        raise ValueError("deadband must be in [0, 1)")
+    if normalized <= deadband:
+        return 0.0
+    return float((normalized - deadband) / (1.0 - deadband))
+
+
+def map_to_safe_range(
+    normalized: float,
+    bound: tuple[float | None, float | None],
+    safe_fraction: float,
+) -> float:
+    low, high = bound
+    if low is None or high is None:
+        raise RuntimeError(f"Missing valid joint bound: {bound}")
+    if not 0.0 <= safe_fraction <= 1.0:
+        raise ValueError("safe_fraction must be in [0, 1]")
+
+    low_value = float(low)
+    high_value = float(high)
+    safe_high = low_value + safe_fraction * (high_value - low_value)
+    clipped = float(np.clip(normalized, 0.0, 1.0))
+    return low_value + clipped * (safe_high - low_value)
+
+
+class Revo2ManusQposMapper:
+    """Convert calibrated finger curls into conservative Revo2 qpos."""
+
+    def __init__(
+        self,
+        *,
+        hand_model,
+        bounds,
+        profile: ManusCalibrationProfile,
+        side: str,
+        four_finger_safe_range: float = 0.65,
+        thumb_metacarpal_safe_range: float = 0.35,
+        thumb_proximal_safe_range: float = 0.55,
+        deadbands: Mapping[str, float] = DEFAULT_DEADBANDS,
+        pinky_ring_dominance_compensation: float = 0.60,
+        smooth_alpha: float = 0.22,
+        max_step_rad: float = 0.035,
+    ) -> None:
+        self.side = normalize_hand_side(side)
+        if profile.side != self.side:
+            raise ValueError(
+                "MANUS calibration side does not match hand model side"
+            )
+
+        self.hand_model = hand_model
+        self.bounds = bounds
+        self.profile = profile
+        self.four_finger_safe_range = float(
+            four_finger_safe_range
+        )
+        self.thumb_metacarpal_safe_range = float(
+            thumb_metacarpal_safe_range
+        )
+        self.thumb_proximal_safe_range = float(
+            thumb_proximal_safe_range
+        )
+        self.deadbands = {
+            name: float(deadbands[name])
+            for name in FINGER_NAMES
+        }
+        self.pinky_ring_dominance_compensation = float(
+            pinky_ring_dominance_compensation
+        )
+        self.smooth_alpha = float(smooth_alpha)
+        self.max_step_rad = float(max_step_rad)
+
+        if not 0.0 < self.smooth_alpha <= 1.0:
+            raise ValueError("smooth_alpha must be in (0, 1]")
+        if self.max_step_rad <= 0.0:
+            raise ValueError("max_step_rad must be > 0")
+
+        joint_to_qpos = (
+            hand_model.get_joint_name_to_qpos_index()
+        )
+        required_names = {
+            "thumb_metacarpal":
+                f"{self.side}_thumb_metacarpal_joint",
+            "thumb_proximal":
+                f"{self.side}_thumb_proximal_joint",
+            "index":
+                f"{self.side}_index_proximal_joint",
+            "middle":
+                f"{self.side}_middle_proximal_joint",
+            "ring":
+                f"{self.side}_ring_proximal_joint",
+            "pinky":
+                f"{self.side}_pinky_proximal_joint",
+        }
+
+        missing = [
+            joint_name
+            for joint_name in required_names.values()
+            if joint_name not in joint_to_qpos
+        ]
+        if missing:
+            raise ValueError(
+                "Revo2 MANUS mapping is missing joints: "
+                + ", ".join(missing)
+            )
+
+        self.qpos_indices = {
+            key: int(joint_to_qpos[joint_name])
+            for key, joint_name in required_names.items()
+        }
+        self.controlled_indices = [
+            self.qpos_indices["thumb_metacarpal"],
+            self.qpos_indices["thumb_proximal"],
+            self.qpos_indices["index"],
+            self.qpos_indices["middle"],
+            self.qpos_indices["ring"],
+            self.qpos_indices["pinky"],
+        ]
+        self._previous: np.ndarray | None = None
+
+    def reset(self) -> None:
+        self._previous = None
+
+    def map(
+        self,
+        landmarks: np.ndarray,
+        base_qpos: np.ndarray,
+    ) -> tuple[np.ndarray, dict[str, float], dict[str, float]]:
+        points = np.asarray(landmarks, dtype=np.float64)
+        if points.shape != (21, 3):
+            raise ValueError(
+                f"Expected MANUS landmarks shape (21, 3), "
+                f"got {points.shape}"
+            )
+        if not np.all(np.isfinite(points)):
+            raise ValueError(
+                "MANUS landmarks contain NaN or infinity"
+            )
+
+        raw_normalized: dict[str, float] = {}
+        for finger_name in FINGER_NAMES:
+            calibration = self.profile.fingers[
+                finger_name
+            ]
+
+            if calibration.metric == "current":
+                curl = finger_curl_degrees(
+                    points,
+                    FINGER_INDICES[finger_name],
+                )
+            elif calibration.metric == "joint_sum":
+                curl = finger_joint_sum_degrees(
+                    points,
+                    FINGER_INDICES[finger_name],
+                )
+            else:
+                raise RuntimeError(
+                    "Unsupported MANUS finger metric: "
+                    f"{calibration.metric!r}"
+                )
+
+            normalized = (
+                (curl - calibration.open_deg)
+                / calibration.span_deg
+            )
+            raw_normalized[finger_name] = float(
+                np.clip(normalized, 0.0, 1.0)
+            )
+
+        corrected = {
+            finger_name: apply_deadband(
+                raw_normalized[finger_name],
+                self.deadbands[finger_name],
+            )
+            for finger_name in FINGER_NAMES
+        }
+
+        pinky_excess = max(
+            0.0,
+            corrected["pinky"] - corrected["ring"],
+        )
+        corrected["ring"] = float(
+            np.clip(
+                corrected["ring"]
+                - self.pinky_ring_dominance_compensation
+                * pinky_excess,
+                0.0,
+                1.0,
+            )
+        )
+
+        desired = np.asarray(
+            base_qpos,
+            dtype=np.float64,
+        ).copy()
+
+        thumb_metacarpal = self.qpos_indices[
+            "thumb_metacarpal"
+        ]
+        thumb_proximal = self.qpos_indices[
+            "thumb_proximal"
+        ]
+
+        desired[thumb_metacarpal] = map_to_safe_range(
+            corrected["thumb"],
+            self.bounds[thumb_metacarpal],
+            self.thumb_metacarpal_safe_range,
+        )
+        desired[thumb_proximal] = map_to_safe_range(
+            corrected["thumb"],
+            self.bounds[thumb_proximal],
+            self.thumb_proximal_safe_range,
+        )
+
+        for finger_name in FOUR_FINGER_NAMES:
+            qpos_index = self.qpos_indices[finger_name]
+            desired[qpos_index] = map_to_safe_range(
+                corrected[finger_name],
+                self.bounds[qpos_index],
+                self.four_finger_safe_range,
+            )
+
+        previous = self._previous
+        if previous is None:
+            previous = desired.copy()
+
+        filtered = previous.copy()
+        for qpos_index in self.controlled_indices:
+            smoothed_target = (
+                previous[qpos_index]
+                + self.smooth_alpha
+                * (
+                    desired[qpos_index]
+                    - previous[qpos_index]
+                )
+            )
+            delta = float(
+                np.clip(
+                    smoothed_target - previous[qpos_index],
+                    -self.max_step_rad,
+                    self.max_step_rad,
+                )
+            )
+            filtered[qpos_index] = previous[qpos_index] + delta
+
+        filtered = self.hand_model.apply_mimic_constraints(
+            filtered
+        )
+        self._previous = filtered.copy()
+        return filtered, raw_normalized, corrected
+
+
+class CalibratedManusRetargetingEngine(RetargetingEngine):
+    """Retargeting engine with explicit calibrated MANUS override."""
+
+    def __init__(
+        self,
+        config,
+        *,
+        calibration_path: str | Path,
+        input_type: str = "manus_ros2",
+    ) -> None:
+        super().__init__(config, input_type=input_type)
+        if not config.hand.name.startswith("revo2_"):
+            raise ValueError(
+                "Calibrated MANUS mapping currently supports "
+                "Revo2 configs only"
+            )
+
+        self.calibration_path = str(
+            Path(calibration_path).expanduser().resolve()
+        )
+        self.calibration_profile = (
+            load_manus_calibration_profile(
+                self.calibration_path,
+                expected_side=config.hand.side,
+            )
+        )
+        self._calibrated_mapper = Revo2ManusQposMapper(
+            hand_model=self.hand_model,
+            bounds=self.retargeter._bounds,
+            profile=self.calibration_profile,
+            side=config.hand.side,
+        )
+        self._calibrated_frame_count = 0
+
+    @classmethod
+    def from_config_path(
+        cls,
+        config_path: str,
+        *,
+        calibration_path: str | Path,
+        input_type: str = "manus_ros2",
+    ) -> "CalibratedManusRetargetingEngine":
+        return cls(
+            load_retargeting_config(config_path),
+            calibration_path=calibration_path,
+            input_type=input_type,
+        )
+
+    def process(
+        self,
+        frame: HandFrame,
+    ) -> RetargetingStepResult:
+        base_result = super().process(frame)
+        qpos, raw, corrected = self._calibrated_mapper.map(
+            frame.landmarks_3d,
+            base_result.qpos,
+        )
+
+        self._calibrated_frame_count += 1
+        frame_count = self._calibrated_frame_count
+        if frame_count <= 3 or frame_count % 30 == 0:
+            raw_display = {
+                name: round(value, 3)
+                for name, value in raw.items()
+            }
+            corrected_display = {
+                name: round(value, 3)
+                for name, value in corrected.items()
+            }
+            print(
+                "CALIBRATED_MANUS_TRACE "
+                f"frame={frame_count} "
+                f"raw={raw_display} "
+                f"corrected={corrected_display} "
+                f"qpos={np.round(qpos, 4).tolist()}",
+                flush=True,
+            )
+
+        return RetargetingStepResult(
+            qpos=qpos.copy(),
+            target_directions=base_result.target_directions,
+            processed_landmarks=base_result.processed_landmarks,
+            hand_side=base_result.hand_side,
+            target_qpos=base_result.target_qpos,
+            backend=base_result.backend,
+        )
+
+
+__all__ = [
+    "CalibratedManusRetargetingEngine",
+    "FingerCalibration",
+    "ManusCalibrationProfile",
+    "Revo2ManusQposMapper",
+    "apply_deadband",
+    "finger_curl_degrees",
+    "finger_joint_sum_degrees",
+    "load_manus_calibration_profile",
+    "map_to_safe_range",
+]
diff --git a/src/somehand/cli/commands.py b/src/somehand/cli/commands.py
index a2ca5f3..20fb07a 100644
--- a/src/somehand/cli/commands.py
+++ b/src/somehand/cli/commands.py
@@ -284,6 +284,21 @@ def _run_manus_ros2(args: argparse.Namespace) -> None:
         f"Expected message side: {display_hand_side(args.hand)}",
         "Topic numbering does not define handedness; message.side is validated.",
     ]
+    manus_calibration = getattr(
+        args,
+        "manus_calibration",
+        None,
+    )
+    if manus_calibration is not None:
+        extra_lines.extend(
+            [
+                f"MANUS calibration: {manus_calibration}",
+                "Calibrated mapping: Revo2 viewer validation only",
+                "Real hand output: DISABLED",
+                "CAN: DISABLED",
+                "Modbus: DISABLED",
+            ]
+        )
     if recording_controller is not None:
         extra_lines.append("Press 'r' in the terminal or robot-hand viewer to start recording.")
         extra_lines.append("Press 's' in the terminal or robot-hand viewer to stop recording, save, and exit.")
diff --git a/src/somehand/cli/parser.py b/src/somehand/cli/parser.py
index 0807574..deb6501 100644
--- a/src/somehand/cli/parser.py
+++ b/src/somehand/cli/parser.py
@@ -167,6 +167,15 @@ def build_parser() -> argparse.ArgumentParser:
         default=2.0,
         help="Seconds to wait for the next matching MANUS frame",
     )
+    manus_ros2.add_argument(
+        "--manus-calibration",
+        default=None,
+        help=(
+            "Optional calibrated MANUS finger profile JSON. "
+            "During validation this mode supports Revo2 with "
+            "--backend viewer only."
+        ),
+    )
 
     hc_mocap = subparsers.add_parser("hc-mocap", help="Retarget from a live hc_mocap UDP stream")
     add_common_args(hc_mocap)
diff --git a/src/somehand/cli/runtime.py b/src/somehand/cli/runtime.py
index 274862f..de9259d 100644
--- a/src/somehand/cli/runtime.py
+++ b/src/somehand/cli/runtime.py
@@ -11,6 +11,9 @@
     RetargetingEngine,
     RetargetingSession,
 )
+from somehand.application.manus_calibration import (
+    CalibratedManusRetargetingEngine,
+)
 from somehand.runtime import (
     AsyncBiHandLandmarkOutputSink,
     AsyncLandmarkOutputSink,
@@ -178,7 +181,28 @@ def _build_bihand_visual_sinks(
 
 
 def build_engine(args: argparse.Namespace, *, input_type: str) -> RetargetingEngine:
-    return RetargetingEngine.from_config_path(args.config, input_type=input_type)
+    calibration_path = getattr(args, "manus_calibration", None)
+    if calibration_path is None:
+        return RetargetingEngine.from_config_path(
+            args.config,
+            input_type=input_type,
+        )
+
+    if input_type != "manus_ros2":
+        raise ValueError(
+            "--manus-calibration is supported only by manus-ros2"
+        )
+    if getattr(args, "backend", "viewer") != "viewer":
+        raise ValueError(
+            "Calibrated MANUS mode is viewer-only during validation; "
+            "do not use --backend real or --backend sim"
+        )
+
+    return CalibratedManusRetargetingEngine.from_config_path(
+        args.config,
+        calibration_path=calibration_path,
+        input_type=input_type,
+    )
 
 
 def build_bihand_engine(args: argparse.Namespace, *, input_type: str) -> BiHandRetargetingEngine:
diff --git a/tests/test_manus_calibration.py b/tests/test_manus_calibration.py
new file mode 100644
index 0000000..1de6d9a
--- /dev/null
+++ b/tests/test_manus_calibration.py
@@ -0,0 +1,166 @@
+"""Tests for calibrated MANUS -> Revo2 mapping."""
+
+from __future__ import annotations
+
+import json
+
+import numpy as np
+import pytest
+
+from somehand.application.manus_calibration import (
+    FINGER_NAMES,
+    ManusCalibrationProfile,
+    Revo2ManusQposMapper,
+    apply_deadband,
+    load_manus_calibration_profile,
+)
+
+
+def _profile(side: str = "right") -> ManusCalibrationProfile:
+    from somehand.application.manus_calibration import (
+        FingerCalibration,
+    )
+
+    return ManusCalibrationProfile(
+        version=1,
+        side=side,
+        fingers={
+            name: FingerCalibration(
+                open_deg=0.0,
+                comfortable_closed_deg=90.0,
+            )
+            for name in FINGER_NAMES
+        },
+    )
+
+
+def _closed_landmarks() -> np.ndarray:
+    landmarks = np.zeros((21, 3), dtype=np.float64)
+    bases = (1, 5, 9, 13, 17)
+    for base in bases:
+        landmarks[base] = [0.0, 0.0, 0.0]
+        landmarks[base + 1] = [1.0, 0.0, 0.0]
+        landmarks[base + 2] = [1.0, 0.0, 0.0]
+        landmarks[base + 3] = [1.0, 1.0, 0.0]
+    return landmarks
+
+
+class _FakeHandModel:
+    def get_joint_name_to_qpos_index(self):
+        return {
+            "right_thumb_metacarpal_joint": 0,
+            "right_thumb_proximal_joint": 1,
+            "right_index_proximal_joint": 3,
+            "right_middle_proximal_joint": 5,
+            "right_ring_proximal_joint": 7,
+            "right_pinky_proximal_joint": 9,
+        }
+
+    def apply_mimic_constraints(self, qpos):
+        result = np.asarray(qpos, dtype=np.float64).copy()
+        result[2] = result[1]
+        result[4] = result[3]
+        result[6] = result[5]
+        result[8] = result[7]
+        result[10] = result[9]
+        return result
+
+
+def test_load_profile_validates_side(tmp_path):
+    path = tmp_path / "calibration.json"
+    path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "side": "Right",
+                "fingers": {
+                    name: {
+                        "open_deg": 0.0,
+                        "comfortable_closed_deg": 90.0,
+                    }
+                    for name in FINGER_NAMES
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    profile = load_manus_calibration_profile(
+        path,
+        expected_side="right",
+    )
+    assert profile.side == "right"
+
+    with pytest.raises(ValueError, match="does not match"):
+        load_manus_calibration_profile(
+            path,
+            expected_side="left",
+        )
+
+
+def test_load_profile_rejects_non_positive_span(tmp_path):
+    path = tmp_path / "calibration.json"
+    path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "side": "right",
+                "fingers": {
+                    name: {
+                        "open_deg": 10.0,
+                        "comfortable_closed_deg": 10.0,
+                    }
+                    for name in FINGER_NAMES
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    with pytest.raises(ValueError, match="span"):
+        load_manus_calibration_profile(path)
+
+
+def test_apply_deadband():
+    assert apply_deadband(0.05, 0.08) == 0.0
+    assert apply_deadband(1.0, 0.08) == pytest.approx(1.0)
+    assert apply_deadband(0.54, 0.08) == pytest.approx(0.5)
+
+
+def test_mapper_applies_safe_ranges_and_mimic_constraints():
+    mapper = Revo2ManusQposMapper(
+        hand_model=_FakeHandModel(),
+        bounds=[(0.0, 1.0)] * 11,
+        profile=_profile(),
+        side="right",
+    )
+
+    qpos, raw, corrected = mapper.map(
+        _closed_landmarks(),
+        np.zeros(11, dtype=np.float64),
+    )
+
+    assert raw == pytest.approx(
+        {name: 1.0 for name in FINGER_NAMES}
+    )
+    assert corrected == pytest.approx(
+        {name: 1.0 for name in FINGER_NAMES}
+    )
+
+    assert qpos[0] == pytest.approx(0.35)
+    assert qpos[1] == pytest.approx(0.55)
+    assert qpos[2] == pytest.approx(0.55)
+
+    for source, mimic in ((3, 4), (5, 6), (7, 8), (9, 10)):
+        assert qpos[source] == pytest.approx(0.65)
+        assert qpos[mimic] == pytest.approx(0.65)
+
+
+def test_mapper_rejects_wrong_side():
+    with pytest.raises(ValueError, match="side"):
+        Revo2ManusQposMapper(
+            hand_model=_FakeHandModel(),
+            bounds=[(0.0, 1.0)] * 11,
+            profile=_profile(side="left"),
+            side="right",
+        )
diff --git a/tests/test_manus_pinky_metric.py b/tests/test_manus_pinky_metric.py
new file mode 100644
index 0000000..0863abb
--- /dev/null
+++ b/tests/test_manus_pinky_metric.py
@@ -0,0 +1,79 @@
+"""Tests for per-finger MANUS curl metrics."""
+
+from __future__ import annotations
+
+import json
+
+import numpy as np
+
+from somehand.application.manus_calibration import (
+    FINGER_NAMES,
+    finger_curl_degrees,
+    finger_joint_sum_degrees,
+    load_manus_calibration_profile,
+)
+
+
+def test_joint_sum_detects_mcp_only_flexion():
+    landmarks = np.zeros(
+        (21, 3),
+        dtype=np.float64,
+    )
+
+    # Wrist -> MCP points along +X.
+    landmarks[0] = [0.0, 0.0, 0.0]
+    landmarks[17] = [1.0, 0.0, 0.0]
+
+    # The entire pinky points along +Y. This represents an
+    # MCP-only bend: internal segments remain parallel.
+    landmarks[18] = [1.0, 1.0, 0.0]
+    landmarks[19] = [1.0, 2.0, 0.0]
+    landmarks[20] = [1.0, 3.0, 0.0]
+
+    indices = (17, 18, 19, 20)
+
+    assert finger_curl_degrees(
+        landmarks,
+        indices,
+    ) == 0.0
+
+    assert finger_joint_sum_degrees(
+        landmarks,
+        indices,
+    ) == 90.0
+
+
+def test_profile_supports_per_finger_metric(tmp_path):
+    path = tmp_path / "profile.json"
+
+    fingers = {
+        name: {
+            "open_deg": 0.0,
+            "comfortable_closed_deg": 90.0,
+            "metric": (
+                "joint_sum"
+                if name == "pinky"
+                else "current"
+            ),
+        }
+        for name in FINGER_NAMES
+    }
+
+    path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "side": "right",
+                "fingers": fingers,
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    profile = load_manus_calibration_profile(
+        path,
+        expected_side="right",
+    )
+
+    assert profile.fingers["ring"].metric == "current"
+    assert profile.fingers["pinky"].metric == "joint_sum"

From ff3108b8bc0ad3892080ce8505f68b3b955bcffc Mon Sep 17 00:00:00 2001
From: Wang Pengrui <3519854206@qq.com>
Date: Mon, 20 Jul 2026 23:03:20 +0800
Subject: [PATCH 5/5] feat: add viewer-only bimanual MANUS ROS 2 input

---
 src/somehand/application/bihand_engine.py |  65 ++++-
 src/somehand/cli/commands.py              |  73 ++++++
 src/somehand/cli/main.py                  |   3 +
 src/somehand/cli/parser.py                |  30 +++
 src/somehand/cli/runtime.py               |  30 ++-
 src/somehand/runtime/__init__.py          |   2 +
 src/somehand/runtime/manus_source.py      | 199 ++++++++++++++-
 tests/test_manus_bihand.py                | 283 ++++++++++++++++++++++
 8 files changed, 676 insertions(+), 9 deletions(-)
 create mode 100644 tests/test_manus_bihand.py

diff --git a/src/somehand/application/bihand_engine.py b/src/somehand/application/bihand_engine.py
index f2b8091..685a533 100644
--- a/src/somehand/application/bihand_engine.py
+++ b/src/somehand/application/bihand_engine.py
@@ -13,6 +13,7 @@
 from somehand.infrastructure.config_loader import load_bihand_config
 
 from .engine import RetargetingEngine
+from .manus_calibration import CalibratedManusRetargetingEngine
 
 
 def _copy_step_result(result: RetargetingStepResult) -> RetargetingStepResult:
@@ -27,18 +28,70 @@ def _copy_step_result(result: RetargetingStepResult) -> RetargetingStepResult:
 class BiHandRetargetingEngine:
     """Stable application-layer entry for one-step bi-hand retargeting."""
 
-    def __init__(self, config: BiHandRetargetingConfig, *, input_type: str = "landmarks"):
+    def __init__(
+        self,
+        config: BiHandRetargetingConfig,
+        *,
+        input_type: str = "landmarks",
+        left_engine: RetargetingEngine | None = None,
+        right_engine: RetargetingEngine | None = None,
+    ):
         self.config = config
         self.input_type = input_type
-        self.left_engine = RetargetingEngine.from_config_path(config.left_config_path, input_type=input_type)
-        self.right_engine = RetargetingEngine.from_config_path(config.right_config_path, input_type=input_type)
-        self._left_result = self._neutral_result(self.left_engine, hand_side="left")
-        self._right_result = self._neutral_result(self.right_engine, hand_side="right")
+        self.left_engine = (
+            left_engine
+            if left_engine is not None
+            else RetargetingEngine.from_config_path(
+                config.left_config_path, input_type=input_type
+            )
+        )
+        self.right_engine = (
+            right_engine
+            if right_engine is not None
+            else RetargetingEngine.from_config_path(
+                config.right_config_path, input_type=input_type
+            )
+        )
+        self._left_result = self._neutral_result(
+            self.left_engine, hand_side="left"
+        )
+        self._right_result = self._neutral_result(
+            self.right_engine, hand_side="right"
+        )
 
     @classmethod
-    def from_config_path(cls, config_path: str, *, input_type: str = "landmarks") -> "BiHandRetargetingEngine":
+    def from_config_path(
+        cls, config_path: str, *, input_type: str = "landmarks"
+    ) -> "BiHandRetargetingEngine":
         return cls(load_bihand_config(config_path), input_type=input_type)
 
+    @classmethod
+    def from_calibrated_manus_paths(
+        cls,
+        *,
+        config_path: str,
+        left_calibration_path: str,
+        right_calibration_path: str,
+        input_type: str = "manus_bihand_ros2",
+    ) -> "BiHandRetargetingEngine":
+        config = load_bihand_config(config_path)
+        left_engine = CalibratedManusRetargetingEngine.from_config_path(
+            config.left_config_path,
+            calibration_path=left_calibration_path,
+            input_type=input_type,
+        )
+        right_engine = CalibratedManusRetargetingEngine.from_config_path(
+            config.right_config_path,
+            calibration_path=right_calibration_path,
+            input_type=input_type,
+        )
+        return cls(
+            config,
+            input_type=input_type,
+            left_engine=left_engine,
+            right_engine=right_engine,
+        )
+
     def describe(self) -> dict[str, object]:
         return {
             "left_model_name": self.left_engine.config.hand.name,
diff --git a/src/somehand/cli/commands.py b/src/somehand/cli/commands.py
index 20fb07a..6ee6362 100644
--- a/src/somehand/cli/commands.py
+++ b/src/somehand/cli/commands.py
@@ -12,6 +12,7 @@
     RecordingHandTrackingSource,
     TerminalRecordingController,
     create_bihand_hc_mocap_udp_source,
+    create_bihand_manus_ros2_source,
     create_bihand_pico_source,
     create_bihand_recording_source,
     create_hc_mocap_udp_source,
@@ -322,6 +323,78 @@ def _run_manus_ros2(args: argparse.Namespace) -> None:
     _finalize_run(args, summary=summary, source=source)
 
 
+def _run_bihand_manus_ros2(args: argparse.Namespace) -> None:
+    source, recording_controller = (
+        _wrap_bihand_source_for_interactive_recording(
+            _wrap_live_bihand_source(
+                create_bihand_manus_ros2_source(
+                    left_topic=args.left_topic,
+                    right_topic=args.right_topic,
+                    timeout=args.manus_timeout,
+                ),
+                args=args,
+            ),
+            record_output_path=args.record_output,
+        )
+    )
+    engine = _build_bihand_engine(
+        args, input_type="manus_bihand_ros2"
+    )
+    session = _build_bihand_session(
+        engine,
+        visualize=True,
+        show_preview=False,
+        key_callback=(
+            None
+            if recording_controller is None
+            else recording_controller.handle_keypress
+        ),
+    )
+    extra_lines = [
+        "Backend: viewer",
+        f"Signal sampling: {source.fps} fps",
+        f"Left MANUS ROS 2 topic: {args.left_topic}",
+        f"Right MANUS ROS 2 topic: {args.right_topic}",
+        "Each message.side is validated.",
+        f"Left MANUS calibration: {args.left_manus_calibration}",
+        f"Right MANUS calibration: {args.right_manus_calibration}",
+        "Calibrated mapping: Revo2 bi-hand viewer only",
+        "Real hand output: DISABLED",
+        "CAN: DISABLED",
+        "Modbus: DISABLED",
+        "INSPIRE output: DISABLED",
+    ]
+    if recording_controller is not None:
+        extra_lines.extend([
+            "Press 'r' to start recording.",
+            "Press 's' to stop, save, and exit.",
+        ])
+    _print_bihand_startup(
+        engine,
+        source_desc=source.source_desc,
+        tracking_desc=(
+            f"Tracking MANUS hands: Left+Right | Source fps: {source.fps}"
+        ),
+        extra_lines=extra_lines,
+    )
+    if recording_controller is not None:
+        recording_controller.start()
+    try:
+        summary = session.run(
+            source,
+            input_type="manus_bihand_ros2",
+            stop_condition=(
+                None
+                if recording_controller is None
+                else lambda: recording_controller.stop_requested
+            ),
+        )
+    finally:
+        if recording_controller is not None:
+            recording_controller.close()
+    _finalize_bihand_run(args, summary=summary, source=source)
+
+
 def _run_hc_mocap_udp(args: argparse.Namespace) -> None:
     source = _wrap_source_for_recording(
         _wrap_live_hand_source(
diff --git a/src/somehand/cli/main.py b/src/somehand/cli/main.py
index 62f922a..ff042e2 100644
--- a/src/somehand/cli/main.py
+++ b/src/somehand/cli/main.py
@@ -62,6 +62,9 @@ def main(argv: list[str] | None = None) -> None:
     if args.command == "manus-ros2":
         commands._run_manus_ros2(args)
         return
+    if args.command == "manus-bihand-ros2":
+        commands._run_bihand_manus_ros2(args)
+        return
     if args.command == "hc-mocap":
         if args.hand == "both":
             if args.backend != "viewer":
diff --git a/src/somehand/cli/parser.py b/src/somehand/cli/parser.py
index deb6501..feedb90 100644
--- a/src/somehand/cli/parser.py
+++ b/src/somehand/cli/parser.py
@@ -177,6 +177,36 @@ def build_parser() -> argparse.ArgumentParser:
         ),
     )
 
+    manus_bihand_ros2 = subparsers.add_parser(
+        "manus-bihand-ros2",
+        help="Retarget two MANUS topics in one viewer-only session",
+    )
+    manus_bihand_ros2.add_argument(
+        "-c", "--config",
+        default=str(DEFAULT_BIHAND_CONFIG_PATH),
+        help="Path to bi-hand retargeting config YAML",
+    )
+    manus_bihand_ros2.add_argument(
+        "--record-output", default=None,
+        help="Output pickle file for recorded bi-hand frames",
+    )
+    manus_bihand_ros2.add_argument(
+        "--backend", choices=["viewer"], default="viewer",
+        help="Bi-hand MANUS is viewer-only",
+    )
+    add_live_sampling_args(manus_bihand_ros2)
+    manus_bihand_ros2.add_argument("--left-topic", required=True)
+    manus_bihand_ros2.add_argument("--right-topic", required=True)
+    manus_bihand_ros2.add_argument(
+        "--manus-timeout", type=float, default=2.0
+    )
+    manus_bihand_ros2.add_argument(
+        "--left-manus-calibration", required=True
+    )
+    manus_bihand_ros2.add_argument(
+        "--right-manus-calibration", required=True
+    )
+
     hc_mocap = subparsers.add_parser("hc-mocap", help="Retarget from a live hc_mocap UDP stream")
     add_common_args(hc_mocap)
     add_live_sampling_args(hc_mocap)
diff --git a/src/somehand/cli/runtime.py b/src/somehand/cli/runtime.py
index de9259d..e45eb90 100644
--- a/src/somehand/cli/runtime.py
+++ b/src/somehand/cli/runtime.py
@@ -205,8 +205,34 @@ def build_engine(args: argparse.Namespace, *, input_type: str) -> RetargetingEng
     )
 
 
-def build_bihand_engine(args: argparse.Namespace, *, input_type: str) -> BiHandRetargetingEngine:
-    return BiHandRetargetingEngine.from_config_path(args.config, input_type=input_type)
+def build_bihand_engine(
+    args: argparse.Namespace, *, input_type: str
+) -> BiHandRetargetingEngine:
+    left_calibration = getattr(args, "left_manus_calibration", None)
+    right_calibration = getattr(args, "right_manus_calibration", None)
+    if left_calibration is None and right_calibration is None:
+        return BiHandRetargetingEngine.from_config_path(
+            args.config, input_type=input_type
+        )
+    if input_type != "manus_bihand_ros2":
+        raise ValueError(
+            "Bi-hand MANUS calibration is supported only by "
+            "manus-bihand-ros2"
+        )
+    if getattr(args, "backend", "viewer") != "viewer":
+        raise ValueError(
+            "Calibrated bi-hand MANUS mode is viewer-only"
+        )
+    if left_calibration is None or right_calibration is None:
+        raise ValueError(
+            "Both left and right MANUS calibrations are required"
+        )
+    return BiHandRetargetingEngine.from_calibrated_manus_paths(
+        config_path=args.config,
+        left_calibration_path=left_calibration,
+        right_calibration_path=right_calibration,
+        input_type=input_type,
+    )
 
 
 def build_session(
diff --git a/src/somehand/runtime/__init__.py b/src/somehand/runtime/__init__.py
index afbcb59..2f68e70 100644
--- a/src/somehand/runtime/__init__.py
+++ b/src/somehand/runtime/__init__.py
@@ -7,7 +7,9 @@
 from .config_validation import validate_runtime_bihand_config, validate_runtime_retargeting_config
 
 _MANUS_EXPORTS = {
+    "BiHandManusRos2InputSource",
     "ManusRos2InputSource",
+    "create_bihand_manus_ros2_source",
     "create_manus_ros2_source",
     "manus_message_to_hand_frame",
 }
diff --git a/src/somehand/runtime/manus_source.py b/src/somehand/runtime/manus_source.py
index 1016b6c..d9a930d 100644
--- a/src/somehand/runtime/manus_source.py
+++ b/src/somehand/runtime/manus_source.py
@@ -7,7 +7,13 @@
 
 import numpy as np
 
-from somehand.core import HandFrame, SourceFrame, normalize_hand_side
+from somehand.core import (
+    BiHandFrame,
+    BiHandSourceFrame,
+    HandFrame,
+    SourceFrame,
+    normalize_hand_side,
+)
 
 
 # MANUS 25-node skeleton -> MediaPipe-style 21 landmarks.
@@ -318,3 +324,194 @@ def create_manus_ros2_source(
         hand_side=hand_side,
         timeout=timeout,
     )
+
+
+class BiHandManusRos2InputSource:
+    """Read fresh left and right MANUS frames from two ROS 2 topics.
+
+    A bi-hand frame is emitted only after both sides provide a fresh message.
+    If either side stops for ``timeout`` seconds, ``StopIteration`` is raised
+    so the session exits rather than reusing stale glove data.
+    """
+
+    def __init__(
+        self,
+        *,
+        left_topic: str,
+        right_topic: str,
+        timeout: float = 2.0,
+        nominal_fps: int = 120,
+    ) -> None:
+        if left_topic == right_topic:
+            raise ValueError("left_topic and right_topic must differ")
+        if float(timeout) <= 0.0:
+            raise ValueError("timeout must be > 0")
+        if int(nominal_fps) <= 0:
+            raise ValueError("nominal_fps must be > 0")
+
+        try:
+            import rclpy
+            from manus_ros2_msgs.msg import ManusGlove
+            from rclpy.qos import qos_profile_sensor_data
+        except ImportError as exc:
+            raise RuntimeError(
+                "MANUS bi-hand input requires ROS 2 Humble, rclpy and "
+                "manus_ros2_msgs. Source the ROS environments first."
+            ) from exc
+
+        self.left_topic = str(left_topic)
+        self.right_topic = str(right_topic)
+        self.source_desc = (
+            f"ros2://left={self.left_topic};right={self.right_topic}"
+        )
+        self.timeout = float(timeout)
+        self._fps = int(nominal_fps)
+        self._rclpy = rclpy
+        self._latest_msgs: dict[str, Any | None] = {
+            "left": None,
+            "right": None,
+        }
+        self._available = True
+        self._received_count = {"left": 0, "right": 0}
+        self._converted_count = {"left": 0, "right": 0}
+        self._side_mismatch_count = {"left": 0, "right": 0}
+        self._timeout_count = 0
+
+        self._owns_context = not rclpy.ok()
+        if self._owns_context:
+            rclpy.init(args=None)
+
+        self._node = rclpy.create_node("somehand_manus_bihand_input")
+        self._left_subscription = self._node.create_subscription(
+            ManusGlove,
+            self.left_topic,
+            self._left_message_callback,
+            qos_profile_sensor_data,
+        )
+        self._right_subscription = self._node.create_subscription(
+            ManusGlove,
+            self.right_topic,
+            self._right_message_callback,
+            qos_profile_sensor_data,
+        )
+
+    @property
+    def fps(self) -> int:
+        return self._fps
+
+    def _left_message_callback(self, msg: Any) -> None:
+        self._store_message("left", msg)
+
+    def _right_message_callback(self, msg: Any) -> None:
+        self._store_message("right", msg)
+
+    def _store_message(self, expected_side: str, msg: Any) -> None:
+        self._received_count[expected_side] += 1
+        try:
+            message_side = normalize_hand_side(msg.side)
+        except ValueError:
+            self._side_mismatch_count[expected_side] += 1
+            return
+        if message_side != expected_side:
+            self._side_mismatch_count[expected_side] += 1
+            return
+        self._latest_msgs[expected_side] = msg
+
+    def is_available(self) -> bool:
+        return self._available and self._rclpy.ok()
+
+    def get_frame(self) -> BiHandSourceFrame:
+        if not self.is_available():
+            raise StopIteration
+
+        deadline = time.monotonic() + self.timeout
+        while (
+            self._latest_msgs["left"] is None
+            or self._latest_msgs["right"] is None
+        ):
+            if not self._rclpy.ok():
+                self._available = False
+                raise StopIteration
+
+            remaining = deadline - time.monotonic()
+            if remaining <= 0.0:
+                self._timeout_count += 1
+                missing = [
+                    side
+                    for side in ("left", "right")
+                    if self._latest_msgs[side] is None
+                ]
+                print(
+                    "MANUS bi-hand input timeout: "
+                    f"no fresh matching {'+'.join(missing)} frame received "
+                    f"for {self.timeout:.3f}s; stopping session.",
+                    flush=True,
+                )
+                raise StopIteration
+
+            self._rclpy.spin_once(
+                self._node,
+                timeout_sec=min(0.05, remaining),
+            )
+
+        left_msg = self._latest_msgs["left"]
+        right_msg = self._latest_msgs["right"]
+        self._latest_msgs["left"] = None
+        self._latest_msgs["right"] = None
+
+        left_frame = manus_message_to_hand_frame(left_msg)
+        right_frame = manus_message_to_hand_frame(right_msg)
+        if left_frame.hand_side != "left":
+            raise ValueError(
+                f"Expected left MANUS frame, got {left_frame.hand_side!r}"
+            )
+        if right_frame.hand_side != "right":
+            raise ValueError(
+                f"Expected right MANUS frame, got {right_frame.hand_side!r}"
+            )
+
+        self._converted_count["left"] += 1
+        self._converted_count["right"] += 1
+        return BiHandSourceFrame(
+            detection=BiHandFrame(
+                left=left_frame,
+                right=right_frame,
+            )
+        )
+
+    def reset(self) -> bool:
+        return False
+
+    def close(self) -> None:
+        if not self._available:
+            return
+        self._available = False
+        try:
+            self._node.destroy_node()
+        finally:
+            if self._owns_context and self._rclpy.ok():
+                self._rclpy.shutdown()
+
+    def stats_snapshot(self) -> dict[str, object]:
+        return {
+            "left_messages_received": self._received_count["left"],
+            "right_messages_received": self._received_count["right"],
+            "left_frames_converted": self._converted_count["left"],
+            "right_frames_converted": self._converted_count["right"],
+            "left_side_mismatches": self._side_mismatch_count["left"],
+            "right_side_mismatches": self._side_mismatch_count["right"],
+            "timeouts": self._timeout_count,
+        }
+
+
+def create_bihand_manus_ros2_source(
+    *,
+    left_topic: str,
+    right_topic: str,
+    timeout: float,
+) -> BiHandManusRos2InputSource:
+    return BiHandManusRos2InputSource(
+        left_topic=left_topic,
+        right_topic=right_topic,
+        timeout=timeout,
+    )
diff --git a/tests/test_manus_bihand.py b/tests/test_manus_bihand.py
new file mode 100644
index 0000000..2b6afcf
--- /dev/null
+++ b/tests/test_manus_bihand.py
@@ -0,0 +1,283 @@
+"""Tests for viewer-only MANUS bi-hand ROS 2 integration."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+import importlib
+
+import somehand.cli.commands as cli_commands
+import somehand.cli.runtime as cli_runtime
+
+cli_main = importlib.import_module("somehand.cli.main")
+from somehand.cli import build_parser
+from somehand.runtime.manus_source import BiHandManusRos2InputSource
+
+
+MANUS_SPECS = [
+    ("Hand", "Invalid"),
+    ("Thumb", "MCP"), ("Thumb", "PIP"),
+    ("Thumb", "IP"), ("Thumb", "TIP"),
+    ("Index", "MCP"), ("Index", "PIP"),
+    ("Index", "IP"), ("Index", "DIP"), ("Index", "TIP"),
+    ("Middle", "MCP"), ("Middle", "PIP"),
+    ("Middle", "IP"), ("Middle", "DIP"), ("Middle", "TIP"),
+    ("Ring", "MCP"), ("Ring", "PIP"),
+    ("Ring", "IP"), ("Ring", "DIP"), ("Ring", "TIP"),
+    ("Pinky", "MCP"), ("Pinky", "PIP"),
+    ("Pinky", "IP"), ("Pinky", "DIP"), ("Pinky", "TIP"),
+]
+
+
+def _node(node_id: int, chain: str, joint: str):
+    return SimpleNamespace(
+        node_id=node_id,
+        chain_type=chain,
+        joint_type=joint,
+        pose=SimpleNamespace(
+            position=SimpleNamespace(
+                x=float(node_id),
+                y=float(node_id) + 0.1,
+                z=float(node_id) + 0.2,
+            )
+        ),
+    )
+
+
+def _message(side: str):
+    nodes = [
+        _node(node_id, chain, joint)
+        for node_id, (chain, joint) in enumerate(MANUS_SPECS)
+    ]
+    return SimpleNamespace(
+        side=side,
+        raw_nodes=nodes,
+        raw_node_count=len(nodes),
+    )
+
+
+class _FakeRclpy:
+    def ok(self) -> bool:
+        return True
+
+    def spin_once(self, node, *, timeout_sec: float) -> None:
+        return None
+
+
+def _bare_source() -> BiHandManusRos2InputSource:
+    source = BiHandManusRos2InputSource.__new__(
+        BiHandManusRos2InputSource
+    )
+    source.left_topic = "/left"
+    source.right_topic = "/right"
+    source.source_desc = "ros2://left=/left;right=/right"
+    source.timeout = 1.0
+    source._fps = 120
+    source._rclpy = _FakeRclpy()
+    source._node = object()
+    source._latest_msgs = {"left": None, "right": None}
+    source._available = True
+    source._received_count = {"left": 0, "right": 0}
+    source._converted_count = {"left": 0, "right": 0}
+    source._side_mismatch_count = {"left": 0, "right": 0}
+    source._timeout_count = 0
+    return source
+
+
+def test_bihand_manus_source_emits_fresh_pair() -> None:
+    source = _bare_source()
+    source._latest_msgs = {
+        "left": _message("Left"),
+        "right": _message("Right"),
+    }
+    result = source.get_frame()
+    assert result.detection is not None
+    assert result.detection.left.hand_side == "left"
+    assert result.detection.right.hand_side == "right"
+    assert result.detection.left.landmarks_3d.shape == (21, 3)
+    assert result.detection.right.landmarks_3d.shape == (21, 3)
+    assert source._latest_msgs == {"left": None, "right": None}
+
+
+def test_bihand_manus_source_rejects_side_mismatch() -> None:
+    source = _bare_source()
+    source._store_message("left", _message("Right"))
+    assert source._latest_msgs["left"] is None
+    assert source._side_mismatch_count["left"] == 1
+
+
+def test_bihand_manus_source_stops_on_one_side_timeout(capsys) -> None:
+    source = _bare_source()
+    source.timeout = 0.0
+    source._latest_msgs["right"] = _message("Right")
+    with pytest.raises(StopIteration):
+        source.get_frame()
+    captured = capsys.readouterr()
+    assert "MANUS bi-hand input timeout" in captured.out
+    assert "left" in captured.out
+    assert source._timeout_count == 1
+
+
+def _args() -> list[str]:
+    return [
+        "manus-bihand-ros2",
+        "--left-topic", "/manus_glove_1",
+        "--right-topic", "/manus_glove_0",
+        "--left-manus-calibration", "/tmp/left.json",
+        "--right-manus-calibration", "/tmp/right.json",
+    ]
+
+
+def test_manus_bihand_parser_is_viewer_only() -> None:
+    args = build_parser().parse_args(_args())
+    assert args.command == "manus-bihand-ros2"
+    assert args.backend == "viewer"
+    assert args.manus_timeout == 2.0
+    assert args.signal_fps is None
+
+
+def test_manus_bihand_parser_rejects_real_backend() -> None:
+    with pytest.raises(SystemExit):
+        build_parser().parse_args([*_args(), "--backend", "real"])
+
+
+def test_manus_bihand_dispatches_to_handler(monkeypatch) -> None:
+    called = []
+    monkeypatch.setattr(
+        cli_commands,
+        "_run_bihand_manus_ros2",
+        lambda args: called.append(args.command),
+    )
+    cli_main.main(_args())
+    assert called == ["manus-bihand-ros2"]
+
+
+def test_build_bihand_engine_uses_both_profiles(monkeypatch) -> None:
+    calls = {}
+
+    class _FakeEngineType:
+        @staticmethod
+        def from_calibrated_manus_paths(**kwargs):
+            calls.update(kwargs)
+            return "engine"
+
+    monkeypatch.setattr(
+        cli_runtime,
+        "BiHandRetargetingEngine",
+        _FakeEngineType,
+    )
+    args = SimpleNamespace(
+        config="bihand.yaml",
+        backend="viewer",
+        left_manus_calibration="left.json",
+        right_manus_calibration="right.json",
+    )
+    result = cli_runtime.build_bihand_engine(
+        args,
+        input_type="manus_bihand_ros2",
+    )
+    assert result == "engine"
+    assert calls == {
+        "config_path": "bihand.yaml",
+        "left_calibration_path": "left.json",
+        "right_calibration_path": "right.json",
+        "input_type": "manus_bihand_ros2",
+    }
+
+
+def test_run_bihand_manus_builds_source_and_session(monkeypatch) -> None:
+    calls = {}
+
+    class _FakeSource:
+        source_desc = "ros2://left=/manus_glove_1;right=/manus_glove_0"
+        fps = 30
+
+    class _FakeSession:
+        def run(self, source, **kwargs):
+            calls["run_kwargs"] = kwargs
+            return SimpleNamespace(
+                num_frames=8,
+                num_detected=8,
+                num_detected_left=8,
+                num_detected_right=8,
+                num_detected_both=8,
+                source_desc=source.source_desc,
+                input_type="manus_bihand_ros2",
+            )
+
+    def fake_source(**kwargs):
+        calls["source_kwargs"] = kwargs
+        return _FakeSource()
+
+    def fake_engine(args, **kwargs):
+        calls["engine_kwargs"] = kwargs
+        return SimpleNamespace(
+            describe=lambda: {
+                "left_model_name": "revo2_left",
+                "right_model_name": "revo2_right",
+                "left_dof": 11,
+                "right_dof": 11,
+            }
+        )
+
+    def fake_session(engine, **kwargs):
+        calls["session_kwargs"] = kwargs
+        return _FakeSession()
+
+    monkeypatch.setattr(
+        cli_commands,
+        "create_bihand_manus_ros2_source",
+        fake_source,
+    )
+    monkeypatch.setattr(
+        cli_commands,
+        "_wrap_live_bihand_source",
+        lambda source, **kwargs: source,
+    )
+    monkeypatch.setattr(
+        cli_commands,
+        "_wrap_bihand_source_for_interactive_recording",
+        lambda source, **kwargs: (source, None),
+    )
+    monkeypatch.setattr(cli_commands, "_build_bihand_engine", fake_engine)
+    monkeypatch.setattr(cli_commands, "_build_bihand_session", fake_session)
+    monkeypatch.setattr(
+        cli_commands,
+        "_print_bihand_startup",
+        lambda *args, **kwargs: None,
+    )
+    monkeypatch.setattr(
+        cli_commands,
+        "_finalize_bihand_run",
+        lambda *args, **kwargs: None,
+    )
+
+    args = SimpleNamespace(
+        left_topic="/manus_glove_1",
+        right_topic="/manus_glove_0",
+        left_manus_calibration="left.json",
+        right_manus_calibration="right.json",
+        manus_timeout=1.0,
+        signal_fps=30,
+        record_output=None,
+        backend="viewer",
+        config="revo2_bihand.yaml",
+    )
+    cli_commands._run_bihand_manus_ros2(args)
+    assert calls["source_kwargs"] == {
+        "left_topic": "/manus_glove_1",
+        "right_topic": "/manus_glove_0",
+        "timeout": 1.0,
+    }
+    assert calls["engine_kwargs"] == {
+        "input_type": "manus_bihand_ros2"
+    }
+    assert calls["session_kwargs"] == {
+        "visualize": True,
+        "show_preview": False,
+        "key_callback": None,
+    }
+    assert calls["run_kwargs"]["input_type"] == "manus_bihand_ros2"
+    assert calls["run_kwargs"]["stop_condition"] is None