Blame for processor/app/processing/modules/depth_hand_3d.py

"""Depth-image hand 3D pipeline — internal helper for Hand Skeleton.

Combines an RGB video, a depth PNG sequence (16-bit millimetres), and
calibration data to lift MediaPipe 2D hand landmarks into camera-coordinate
3D points in metres. The imported demo retains its historical filename.

No standalone card is registered. Hand Skeleton calls ``run_depth_hand_3d``
when mode=auto finds compatible depth data; otherwise it falls back to the
standalone RGB camera-relative 3D path.

The parquet stores only smoothed measurements. Display-only anchors are not
fed back into tracking and never contaminate exported coordinates.
"""

from __future__ import annotations

import importlib.util
import json
import re
import sys
from pathlib import Path
from typing import Any

import numpy as np

from app.processing import JobContext, ArtifactRef
from app.lerobot_v21 import DepthVideoReader, is_depth_source, iter_video_streams

_BLACK_GLOVE_ROOT = Path(__file__).resolve().parents[1] / "black_glove"
_DEPTH_HAND_DEMO_FILE = _BLACK_GLOVE_ROOT / "d435_hands_demo.py"


def _resolve_hand_landmarker_model() -> Path:
    """Resolve the MediaPipe model from the application or demo package."""
    candidates = (
        Path(__file__).resolve().parents[3] / "models" / "hand_landmarker.task",
        _BLACK_GLOVE_ROOT / "hand_landmarker.task",
    )
    for candidate in candidates:
        if candidate.is_file() and candidate.stat().st_size > 0:
            return candidate
    searched = ", ".join(str(p) for p in candidates)
    raise FileNotFoundError(f"MediaPipe hand model not found; searched: {searched}")


_MODEL = _resolve_hand_landmarker_model()


def _load_demo():
    """importlib 加载_hands_demo.py(自包含、有 __main__ 保护,零副作用)。"""
    try:
        spec = importlib.util.spec_from_file_location(
            "depth_hand_demo_loaded", str(_DEPTH_HAND_DEMO_FILE))
        if spec is None or spec.loader is None:
            return None
        mod = importlib.util.module_from_spec(spec)
        # @dataclass 在 exec 期间要查 sys.modules[__module__].__dict__,
        # 必须先注册进 sys.modules
        sys.modules[spec.name] = mod
        spec.loader.exec_module(mod)
        return mod
    except Exception as exc:
        print(f"[depth_hand_3d] demo load failed: {exc}")
        return None


def _finger_gesture_from_pts(demo, pts3d: np.ndarray) -> tuple[str, int]:
    """(21,3) 相机系米制 → (gesture 标签, fingers bitmask)。

    与 demo HandResult.extended / stereo _finger_gesture_from_world 同一套
    角度判据(Thumb: MCP>145° 且 IP>150°;其余: PIP>150° 且 DIP>140°)。
    """
    pts = np.asarray(pts3d, np.float64).reshape(21, 3)
    angles = {(f, j): demo._angle_between(pts[p], pts[v], pts[n])
              for f, j, v, p, n, _c in demo.JOINT_SPECS}
    extended: list[str] = []
    mask = 0
    bits = {"Thumb": 1, "Index": 2, "Middle": 4, "Ring": 8, "Pinky": 16}
    for finger in demo.FINGERS:
        if finger == "Thumb":
            ok = (angles.get((finger, "MCP"), 0) > 145
                  and angles.get((finger, "IP"), 0) > 150)
        else:
            ok = (angles.get((finger, "PIP"), 0) > 150
                  and angles.get((finger, "DIP"), 0) > 140)
        if ok:
            extended.append(finger.lower())
            mask |= bits[finger]
    label = "fist" if not extended else "open:" + ",".join(extended)
    return label, mask


class DepthHand3DDriver:
    """demo main() 逐帧状态机的可驱动封装(demo L1928-2140 数据路径)。

    每帧输入 (rgb_bgr, depth_mm, frame_idx) → 输出双槽位平滑 3D
    (彩色相机系米制,NaN=无效)。只做数据路径:渲染/伪彩/视频输出
    全部剔除;centroid_anchor/view_anchor 展示路径不回流(不进 parquet)。
    """

    def __init__(self, demo, model_path: str, det_conf: float = 0.4,
                 track_conf: float = 0.4, fill: int = 1,
                 propagate_max: int = 15,
                 color_intr: dict | None = None,
                 depth_to_color: dict | None = None,
                 depth_intr: dict | None = None):
        calib = None
        if color_intr is None or depth_to_color is None:
            calib = json.loads(json.dumps(demo._EMBEDDED_CALIB))
            color_intr = calib["color_intrinsics"]
            depth_to_color = calib["depth_to_color"]
        if depth_intr is None and calib:
            depth_intr = calib["depth_intrinsics"]
        self._aligner = demo.LiveAligner(color_intr, depth_to_color,
                                         depth_intr, fill_passes=fill)
        self._color_intr = (self._aligner.fx_c, self._aligner.fy_c,
                            self._aligner.cx_c, self._aligner.cy_c)
        self._det = demo.MediaPipeDetector(
            model_path=model_path, num_hands=2,
            det_conf=det_conf, track_conf=track_conf)
        self._voter = demo.HandednessVoter()
        self._tracker = demo.HandSlotTracker(max_lost=propagate_max)
        self._smoother = demo.Hand3DSmoother()
        self._soft = demo._SoftSmoother(self._smoother)
        self._lost_counts = [0, 0]
        self._zc_slot: list[float | None] = [None, None]
        self._ws_prev: list[np.ndarray | None] = [None, None]
        self._ws_streak = [0, 0]
        self._gate_streak = [np.zeros(21, np.int64), np.zeros(21, np.int64)]
        # 2D display identity is independent from the metric-3D tracker.
        # This prevents a bad depth sample from hiding a valid RGB detection.
        self._last_kp2d: list[np.ndarray | None] = [None, None]
        self.stats = {"real": [0, 0], "propagated": [0, 0], "absent": [0, 0]}

    def process_frame(self, demo, rgb_bgr, depth_mm, frame_idx: int) -> dict:
        """一帧状态机(镜像 demo L1949-2093)。返回:
        {"slots_3d": (2,21,3) float32 米(平滑后), "labels": [str,str],
         "states": [real|propagated|absent], "presents": [bool,bool],
         "dets": [DetectedHand|None, ...]}"""
        aligner = self._aligner
        if depth_mm is None or depth_mm.shape[:2] != (aligner.dh, aligner.dw):
            aligned = np.zeros((aligner.ch, aligner.cw), np.float32)
        else:
            aligned = aligner.align_depth_to_color(depth_mm)

        hands = self._det.detect(rgb_bgr)
        # 空帧不喂 voter:空帧会清空轨迹(scene reset 语义),短暂漏检
        # 会清票仓 → 重建期原始 label 闪烁 → 两手同 label(demo 注释)。
        if hands:
            self._voter.update(hands, frame_w=rgb_bgr.shape[1],
                               frame_h=rgb_bgr.shape[0], frame=frame_idx,
                               cam="depth_camera")

        pairs = [demo.lift_hand(hd, aligner, aligned) for hd in hands]
        same_lab = (len(pairs) == 2 and pairs[0].left_label
                    and pairs[0].left_label == pairs[1].left_label)
        if same_lab:
            for p in pairs:
                p.left_label = ""
        out = demo.assign_mono_slots(pairs, self._tracker, frame_idx,
                                     self._color_intr,
                                     lost_counts=tuple(self._lost_counts))
        if same_lab:
            for s in range(2):
                if out[s] is not None:
                    sl = self._tracker.slot_label(s)
                    if sl:
                        out[s].left_label = sl

        slot_pairs, slot_dets, states = [], [], []
        for s in range(2):
            if out[s] is not None:
                p = out[s]
                # M5:补点深度锚定到槽级稳定 zc(换手/首帧取实测)
                meas = getattr(p, "measured", None)
                if meas is not None and meas.any():
                    pts3d = np.asarray(p.result.points_3d, np.float64) \
                        .reshape(21, 3)
                    zf = float(np.median(pts3d[meas, 2]))
                    if self._tracker.slot_label(s) != p.left_label \
                            or self._zc_slot[s] is None:
                        self._zc_slot[s] = zf
                    else:
                        self._zc_slot[s] = 0.5 * self._zc_slot[s] + 0.5 * zf
                    demo.apply_slot_zc(p, self._zc_slot[s], aligner)
                # 时序一致性门:与槽预测差 >150mm 的点判可疑置 NaN
                gated, wholesale = demo.gate_observations(
                    p.result.points_3d, self._tracker.predict(s, frame_idx))
                # M6:门控锁死豁免(连续被门控 ≥_GATE_FORGIVE 且观测恢复
                # 有限时采信观测);换手帧不豁免。
                if not wholesale:
                    if self._tracker.slot_label(s) != p.left_label:
                        self._gate_streak[s][:] = 0
                    meas3d = np.asarray(p.result.points_3d, np.float64) \
                        .reshape(21, 3)
                    g_fin = np.isfinite(
                        np.asarray(gated, np.float64).reshape(-1, 3)).all(axis=1)
                    m_fin = np.isfinite(meas3d).all(axis=1)
                    gs = self._gate_streak[s]
                    latched = ~g_fin & m_fin
                    gs[latched] += 1
                    gs[~latched] = 0
                    forgive = (gs >= demo._GATE_FORGIVE) & m_fin
                    if forgive.any():
                        gated[forgive] = meas3d[forgive]
                if wholesale:
                    # M3②:整手级不匹配先两帧确认,连续两帧观测互相一致
                    # 才判槽状态过时 → 借 label 翻转触发槽位重置。
                    if demo._ws_agree(self._ws_prev[s], gated) \
                            or self._ws_streak[s] >= 3:
                        self._tracker.observe_slot(
                            s, "\x00reset", np.full((21, 3), np.nan), frame_idx)
                        self._tracker.observe_slot(s, p.left_label, gated,
                                                   frame_idx)
                        p.result.points_3d = gated
                        self._lost_counts[s] = 0
                        self._ws_prev[s] = None
                        self._ws_streak[s] = 0
                        self._gate_streak[s][:] = 0
                        slot_pairs.append(p)
                        slot_dets.append(p.det)
                        states.append("real")
                    else:
                        self._ws_prev[s] = gated
                        self._ws_streak[s] += 1
                        pred_now = self._tracker.predict(s, frame_idx)
                        if pred_now is not None:
                            slot_pairs.append(demo._pred_pair(
                                pred_now, self._tracker.slot_label(s)))
                            slot_dets.append(None)
                            states.append("propagated")
                        else:
                            slot_pairs.append(demo._nan_pair(
                                self._tracker.slot_label(s)))
                            slot_dets.append(None)
                            states.append("absent")
                else:
                    self._tracker.observe_slot(s, p.left_label, gated,
                                               frame_idx)
                    p.result.points_3d = gated
                    self._lost_counts[s] = 0
                    self._ws_prev[s] = None
                    self._ws_streak[s] = 0
                    slot_pairs.append(p)
                    slot_dets.append(p.det)
                    states.append("real")
            else:
                pred = self._tracker.predict(s, frame_idx)
                self._tracker.mark_lost(s, frame_idx)
                self._lost_counts[s] += 1
                self._ws_prev[s] = None
                self._ws_streak[s] = 0
                if pred is not None:
                    slot_pairs.append(demo._pred_pair(
                        pred, self._tracker.slot_label(s)))
                    slot_dets.append(None)
                    states.append("propagated")
                else:
                    slot_pairs.append(demo._nan_pair(
                        self._tracker.slot_label(s)))
                    slot_dets.append(None)
                    states.append("absent")
        for s, st in enumerate(states):
            self.stats[st][s] += 1

        presents = [st != "absent" for st in states]
        labels = [slot_pairs[s].left_label if out[s] is not None
                  else self._tracker.slot_label(s) for s in range(2)]

        # The metric tracker may reject a 3D observation while the RGB
        # detector still has a perfectly usable 2D hand.  Assign those
        # detections to display slots using handedness first, then the last
        # 2D position, and finally detector order.  This is intentionally a
        # separate association path: 3D validity must not control 2D review.
        display_dets = [slot_dets[s] for s in range(2)]
        used = {id(det) for det in display_dets if det is not None}
        remaining = [p for p in pairs if p.det is not None and id(p.det) not in used]
        for s in range(2):
            if display_dets[s] is not None:
                continue
            slot_label = self._tracker.slot_label(s)
            labeled = [p for p in remaining
                       if slot_label and p.left_label == slot_label]
            if labeled:
                chosen = labeled[0]
            elif remaining and self._last_kp2d[s] is not None:
                prev = self._last_kp2d[s]
                chosen = min(
                    remaining,
                    key=lambda p: float(np.linalg.norm(
                        np.asarray(p.hand2d, np.float64).reshape(-1, 2).mean(0)
                        - prev
                    )),
                )
            elif remaining:
                chosen = remaining[0]
            else:
                chosen = None
            if chosen is not None:
                display_dets[s] = chosen.det
                remaining = [p for p in remaining if p is not chosen]
        for s, det in enumerate(display_dets):
            if det is not None:
                pts2d = np.asarray(det.landmarks, np.float32).reshape(-1, 2)
                self._last_kp2d[s] = pts2d.mean(axis=0)

        # (2,21,3) 槽位 3D(tracker αβ 已平滑)→ OneEuro 再平滑压静止抖动
        h3 = np.stack([np.asarray(p.result.points_3d, np.float64)
                       .reshape(21, 3) for p in slot_pairs])
        valids = [int(np.isfinite(h3[s]).all(axis=1).sum()) for s in range(2)]
        smoothed = self._soft.update(h3, labels, valids)
        return {"slots_3d": smoothed, "labels": labels, "states": states,
                "presents": presents, "dets": slot_dets,
                "display_dets": display_dets}

    def close(self) -> None:
        try:
            self._det.close()
        except Exception:
            pass


def _row(demo, frame_index: int, res: dict, w: int, h: int) -> dict:
    """一帧 → stereo 同构行(hand_0/1_* 列,63 位扁平,NaN 填充)。"""
    row: dict[str, Any] = {"frame_index": int(frame_index)}
    for si in (0, 1):
        pts = np.asarray(res["slots_3d"][si], np.float64)
        present = bool(res["presents"][si])
        det = (res.get("display_dets") or res["dets"])[si]
        label = str(res["labels"][si] or "")
        if det is not None:
            kp2 = np.asarray(det.landmarks, np.float32).reshape(-1, 2)
            kp = np.zeros((21, 3), np.float32)
            kp[:, 0] = kp2[:, 0] / max(1, w)
            kp[:, 1] = kp2[:, 1] / max(1, h)
            kp[:, :2] = np.clip(kp[:, :2], 0.0, 1.0)
            kp = kp.reshape(63).tolist()
        else:
            kp = np.full(63, np.nan, np.float32).tolist()
        if present:
            gesture, fingers = _finger_gesture_from_pts(demo, pts)
        else:
            gesture, fingers = "", -1
        row.update({
            f"hand_{si}_present": present,
            f"hand_{si}_2d_present": bool(det is not None),
            f"hand_{si}_landmarks_3d": (pts.reshape(63).tolist() if present
                                        else np.full(63, np.nan,
                                                     np.float32).tolist()),
            f"hand_{si}_keypoints": kp,
            f"hand_{si}_label": label if present else "",
            f"hand_{si}_reprojection_error": float("nan"),
            f"hand_{si}_gesture": gesture if present else "",
            f"hand_{si}_fingers": int(fingers) if present else -1,
            f"hand_{si}_confidence": float(det.score)
            if det is not None else 0.0,
        })
    return row




def _batch_root_of(ctx: JobContext, depth_cam: str = "") -> tuple[Path, str] | None:
    """定位批次根目录。worker 输入包以 `<批次名>/...` 归档,解压后
    input_root 下多一层批次目录;旧布局(直接平铺)时 input_root 即批次根。"""
    roots = [ctx.input_root]
    roots.extend(sub for sub in sorted(ctx.input_root.iterdir()) if sub.is_dir())
    for root in roots:
        depth_root = root / "depth"
        if not depth_root.is_dir():
            depth_root = root / "meta" / "depth"
        if depth_root.is_dir():
            candidates = ([depth_root / depth_cam] if depth_cam else
                          [p for p in sorted(depth_root.iterdir()) if p.is_dir()])
            for candidate in candidates:
                if candidate.is_dir() and any(candidate.glob("*.png")):
                    return root, candidate.name
        try:
            for source, _path in iter_video_streams(root / "videos"):
                if is_depth_source(source) and (not depth_cam or source == depth_cam):
                    return root, source
        except Exception:
            pass
    return None


def _safe_source_key(value: str | None) -> str:
    """Return a stable filename/feature suffix for a device stream."""
    value = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(value or "").strip())
    return value.strip("._") or "device"


def _session_roots(input_root: Path) -> list[Path]:
    """Candidate unpacked episode roots, including video-only depth batches."""
    roots = [Path(input_root)]
    roots.extend(p for p in sorted(Path(input_root).iterdir()) if p.is_dir())
    result = []
    for root in roots:
        if (root / "depth").is_dir() or (root / "meta" / "depth").is_dir():
            result.append(root)
            continue
        try:
            if any(is_depth_source(source)
                   for source, _path in iter_video_streams(root / "videos")):
                result.append(root)
        except (OSError, TypeError, ValueError):
            continue
    return result


def _device_records(batch_root: Path,
                    embedded_metadata: dict | None = None) -> list[dict]:
    """Read physical-device/slot declarations from the uploaded metadata.

    Older uploads may not contain ``devices``; callers then use the source-name
    fallback below.  The declaration is deliberately kept local to the worker
    so a multi-device episode cannot accidentally use another episode's slots.
    """
    documents = []
    if isinstance(embedded_metadata, dict):
        documents.append(embedded_metadata)
    for name in ("metadata.json", "meta/info.json"):
        path = batch_root / name
        if not path.is_file():
            continue
        try:
            documents.append(json.loads(path.read_text(encoding="utf-8")))
        except (OSError, ValueError, TypeError):
            continue
    for doc in documents:
        if not isinstance(doc, dict):
            continue
        raw = doc.get("devices") or []
        if isinstance(raw, dict):
            raw = [dict(value or {}, key=key) for key, value in raw.items()]
        result = []
        for index, item in enumerate(raw):
            if not isinstance(item, dict):
                continue
            slots = [str(v) for v in (item.get("slots") or []) if str(v).strip()]
            result.append({
                "key": str(item.get("key") or item.get("id") or f"device_{index}"),
                "name": str(item.get("name") or item.get("device_name") or ""),
                "slots": slots,
                "calibration": str(item.get("calibration") or ""),
                "serial": str(item.get("serial") or item.get("serial_number") or ""),
            })
        if result:
            return result
    return []


def _device_token(value: str | None) -> str:
    """Normalize D405_depth_rgb and D405_depth to the same device token."""
    text = re.sub(r"[^a-z0-9]+", "_", str(value or "").lower()).strip("_")
    for suffix in ("_depth_rgb", "_depth_color", "_color", "_rgb", "_depth"):
        if text.endswith(suffix):
            text = text[:-len(suffix)].rstrip("_")
            break
    return text


def _find_device_pairs(ctx: JobContext, video_refs: list) -> list[dict]:
    """Pair every RGB artifact with its physical device's depth stream.

    Matching order is metadata slots → normalized device token → single-depth
    legacy fallback.  A pair is returned even when its depth is unavailable so
    the caller can report that device explicitly instead of silently assigning
    another camera's depth image.
    """
    roots = _session_roots(ctx.input_root)
    if not roots:
        return []
    root = roots[0]
    depth_root = root / "depth"
    if not depth_root.is_dir():
        depth_root = root / "meta" / "depth"
    depth_dirs = {
        p.name: p for p in sorted(depth_root.iterdir())
        if depth_root.is_dir() and p.is_dir() and any(p.glob("*.png"))
    } if depth_root.is_dir() else {}
    episode_index = None
    episode_row = None
    try:
        from app.project_dataset import project_episode_rows
        rows = project_episode_rows(root)
        if len(rows) == 1:
            episode_row = rows[0]
            episode_index = int(episode_row.get("episode_index"))
    except (OSError, ValueError, TypeError):
        episode_index = None
    depth_videos = {
        source: path for source, path in iter_video_streams(root / "videos")
        if is_depth_source(source)
        and (episode_index is None
             or path.stem == f"episode_{episode_index:06d}")
    }
    depth_sources = set(depth_dirs) | set(depth_videos)
    embedded_collector = {}
    if isinstance(episode_row, dict):
        collector = episode_row.get("collector")
        if isinstance(collector, dict):
            candidate = collector.get("metadata.json")
            if isinstance(candidate, dict):
                embedded_collector = candidate
    records = _device_records(root, embedded_collector)
    pairs = []
    for ref in video_refs:
        source = str(ref.source_key or Path(str(ref.path or "video")).stem)
        source_lower = source.lower()
        if ("depth" in source_lower and
                not any(token in source_lower for token in ("rgb", "color", "video"))):
            continue
        record = next((d for d in records if source in d["slots"]), None)
        if record is None:
            # Workflow input cards often use the physical device token
            # (``D405``) while uploaded metadata uses a slot name
            # (``D405_depth_rgb``).  Resolve that alias before falling back to
            # a synthetic record, so serial/calibration declarations are not
            # lost and another same-resolution device cannot win inference.
            token = _device_token(source)
            record = next(
                (d for d in records
                 if token and any(_device_token(slot) == token
                                  for slot in d["slots"])),
                None,
            )
        depth_name = ""
        if record:
            depth_name = next((slot for slot in record["slots"]
                               if slot in depth_sources and slot != source), "")
        if not depth_name:
            token = _device_token(source)
            depth_name = next((name for name in depth_sources
                               if _device_token(name) == token), "")
        if not depth_name and len(depth_sources) == 1:
            depth_name = next(iter(depth_sources))
        pairs.append({
            "rgb_ref": ref,
            "rgb_source": source,
            "depth_source": depth_name or None,
            "depth_dir": depth_dirs.get(depth_name),
            "depth_video": depth_videos.get(depth_name),
            "device": record or {
                "key": _device_token(source) or "device_0",
                "name": _device_token(source) or source,
                "slots": [source, depth_name] if depth_name else [source],
                "calibration": "",
                "serial": "",
            },
            "batch_root": root,
            "episode_calibration": (
                episode_row.get("calibration") if isinstance(episode_row, dict)
                else {}
            ),
        })
    return pairs


def _intrinsics(values, width: int, height: int) -> dict | None:
    try:
        fx, fy, cx, cy = [float(v) for v in values[:4]]
    except (TypeError, ValueError, IndexError):
        return None
    if min(fx, fy) <= 0:
        return None
    return {"width": int(width), "height": int(height),
            "fx": fx, "fy": fy, "cx": cx, "cy": cy,
            "model": "distortion.none", "coeffs": [0.0] * 5}


def _approx_intrinsics(width: int, height: int) -> dict:
    """Create a deterministic projection for a resolution-mapped depth pair.

    This fallback is only used when the uploaded batch explicitly identifies
    one physical D435 and has no calibration document.  The focal lengths are
    scaled with the image dimensions so an identity transform maps the two
    image grids by normalized coordinates instead of incorrectly treating
    848x480 pixels as 1280x720 pixels.  It never changes the stored depth
    codes; the manifest records that XY is approximate.
    """
    width = int(width)
    height = int(height)
    # A stable 16:9 pinhole approximation.  Keeping fx/fy proportional to
    # the corresponding dimensions makes the same normalized ray land at the
    # same normalized pixel in the other resolution.
    fx = float(width) * 0.9
    fy = float(height) * 0.9
    return {"width": int(width), "height": int(height),
            "fx": fx, "fy": fy,
            "cx": (float(width) - 1.0) / 2.0,
            "cy": (float(height) - 1.0) / 2.0,
            "model": "approximate_pinhole", "coeffs": [0.0] * 5}


def _is_d435_pair(pair: dict) -> bool:
    """Return whether metadata identifies the pair as one physical D435."""
    device = pair.get("device") or {}
    values = (
        device.get("kind"), device.get("name"), device.get("key"),
        pair.get("rgb_source"), pair.get("depth_source"),
    )
    text = " ".join(str(value or "").lower() for value in values)
    return "d435" in text


def _calibration_resolution(raw: dict) -> tuple[int, int] | None:
    """Return a calibration file's declared (width, height), if available."""
    if not isinstance(raw, dict):
        return None
    values = []
    depth_intrinsics = raw.get("depth_intrinsics")
    if isinstance(depth_intrinsics, dict):
        values.append((depth_intrinsics.get("width"),
                       depth_intrinsics.get("height")))
    values.append(raw.get("resolution"))
    for value in values:
        if isinstance(value, (list, tuple)) and len(value) >= 2:
            try:
                width, height = int(value[0]), int(value[1])
            except (TypeError, ValueError):
                continue
            if width > 0 and height > 0:
                return width, height
    return None


def _calibration_token_score(path: Path, raw: dict, pair: dict) -> int:
    """Prefer calibration names that identify the same physical device."""
    text = " ".join([
        path.stem,
        str(raw.get("name") or ""),
        str(pair.get("rgb_source") or ""),
        str(pair.get("depth_source") or ""),
        str((pair.get("device") or {}).get("name") or ""),
    ]).lower()
    tokens = set()
    for value in (pair.get("rgb_source"), pair.get("depth_source"),
                  (pair.get("device") or {}).get("name")):
        token = _device_token(value)
        if token and len(token) > 1:
            tokens.add(token)
    return sum(1 for token in tokens if token in text)


def _infer_calibration_path(pair: dict, depth_shape: tuple[int, int]) -> Path | None:
    """Find a calibration for this device without hard-coding D405/D435.

    Upload metadata from older sessions does not carry calibration paths.  A
    calibration is safe to infer only when its declared resolution equals the
    actual depth PNG resolution.  This prevents a D435 calibration from being
    silently applied to a different camera just because it is the first JSON
    file in the directory.
    """
    root = Path(pair["batch_root"])
    calibration_root = root / "calibration"
    if not calibration_root.is_dir():
        calibration_root = root / "meta" / "calibration"
    if not calibration_root.is_dir():
        return None
    depth_width, depth_height = int(depth_shape[1]), int(depth_shape[0])
    matches: list[tuple[int, Path]] = []
    for path in sorted(calibration_root.rglob("*.json")):
        try:
            raw = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError, TypeError):
            continue
        if not isinstance(raw, dict):
            continue
        if _calibration_resolution(raw) != (depth_width, depth_height):
            continue
        matches.append((_calibration_token_score(path, raw, pair), path))
    if not matches:
        return None
    matches.sort(key=lambda item: (-item[0], str(item[1])))
    return matches[0][1]


def _pair_calibration(pair: dict, demo, rgb_shape: tuple[int, int],
                      depth_shape: tuple[int, int], config: dict | None = None
                      ) -> tuple[dict, str]:
    """Build the calibration arguments expected by ``DepthHand3DDriver``.

    New calibration files may contain the full color/depth/transform schema.
    Legacy files without an explicit RGB/depth registration are not enough to
    produce camera-coordinate 3D.  The canonical S80C ``stereo_depth`` stream
    is an exception: the collector contract says it is already in the RGB
    pixel grid, so this direct-depth path uses identity registration and never
    uses the stereo baseline. A bundled calibration profile may only be used
    when its declared serial matches the physical device.
    """
    import copy
    root = pair["batch_root"]
    device = pair["device"]
    raw = None
    raw_path = ""
    calibration = str(device.get("calibration") or "")
    candidate = root / calibration if calibration else None
    if candidate is None or not candidate.is_file():
        candidate = _infer_calibration_path(pair, depth_shape)
    if candidate is not None and candidate.is_file():
        raw_path = str(candidate.relative_to(root))
        try:
            raw = json.loads(candidate.read_text(encoding="utf-8"))
        except (OSError, ValueError, TypeError):
            raw = None
    raw = raw if isinstance(raw, dict) else {}
    if not raw:
        embedded = pair.get("episode_calibration")
        if isinstance(embedded, dict):
            # The migration stores calibration JSON inside the episode row as
            # {"head_stereo.json": {...}}.  Select a profile matching the
            # decoded depth resolution, then use the same validation below.
            candidates = [value for value in embedded.values()
                          if isinstance(value, dict)]
            dw, dh = int(depth_shape[1]), int(depth_shape[0])
            for profile in candidates:
                if _calibration_resolution(profile) in {(dw, dh), None}:
                    raw = copy.deepcopy(profile)
                    raw_path = "episode.calibration"
                    break
    rw, rh = int(rgb_shape[1]), int(rgb_shape[0])
    dw, dh = int(depth_shape[1]), int(depth_shape[0])
    if all(isinstance(raw.get(key), dict) for key in
           ("color_intrinsics", "depth_intrinsics", "depth_to_color")):
        color = copy.deepcopy(raw["color_intrinsics"])
        depth = copy.deepcopy(raw["depth_intrinsics"])
        transform = copy.deepcopy(raw["depth_to_color"])
        color["width"], color["height"] = rw, rh
        depth["width"], depth["height"] = dw, dh
        return {"color_intrinsics": color, "depth_intrinsics": depth,
                "depth_to_color": transform}, raw_path or "full_calibration"

    rgb_source = str(pair.get("rgb_source") or "").lower()
    color_camera = (
        raw.get("right_camera") if "right" in rgb_source
        else raw.get("left_camera")
    ) or raw.get("color_camera") or raw.get("left_camera") or {}
    depth_camera = raw.get("depth_camera") or {}
    depth_values = depth_camera.get("intrinsic") or color_camera.get("intrinsic")
    color_values = color_camera.get("intrinsic") or depth_values
    depth = _intrinsics(depth_values, dw, dh)
    color = _intrinsics(color_values, rw, rh)
    aligned_flag = (
        raw.get("depth_aligned_to_color") is True
        or raw.get("aligned_depth_to_color") is True
        or raw.get("depth_registered") is True
        or str(raw.get("alignment") or "").lower()
        in {"aligned", "registered", "color", "color_aligned"}
    )
    configured_alignment = str((config or {}).get("depth_alignment") or "").lower()
    direct_depth_flag = configured_alignment in {
        "direct", "aligned", "same_pixel", "same-pixel", "identity"
    }
    # S80C writes one canonical depth slot beside the two RGB slots. The
    # collector's depth image is the authoritative aligned measurement for
    # both views; do not attempt left/right stereo triangulation here.
    canonical_direct_depth = (
        str(pair.get("depth_source") or "").lower() == "stereo_depth"
        and (rw, rh) == (dw, dh)
    )
    if canonical_direct_depth:
        # This collector contract is stronger than the legacy calibration
        # requirement: stereo_depth is a registered depth image, not a second
        # stereo colour camera.  Use calibration intrinsics when present and
        # a stable size-derived projection when the upload omitted them.
        color = color or _approx_intrinsics(rw, rh)
        depth = depth or copy.deepcopy(color)
        return {"color_intrinsics": color, "depth_intrinsics": depth,
                "depth_to_color": {"rotation": np.eye(3).tolist(),
                                    "translation": [0.0, 0.0, 0.0]}}, \
               (raw_path or "direct_depth_same_pixel") + \
               (":direct_depth_same_pixel" if not aligned_flag
                else ":declared_identity_alignment")
    if depth is not None and color is not None and (rw, rh) == (dw, dh) \
            and (aligned_flag or direct_depth_flag or canonical_direct_depth):
        if canonical_direct_depth and not aligned_flag:
            depth = copy.deepcopy(color)
        return {"color_intrinsics": color, "depth_intrinsics": depth,
                "depth_to_color": {"rotation": np.eye(3).tolist(),
                                    "translation": [0.0, 0.0, 0.0]}}, \
               (raw_path or "direct_depth_same_pixel") + \
               (":direct_depth_same_pixel" if canonical_direct_depth and not aligned_flag
                else ":declared_identity_alignment")

    # Uploaded D435 batches commonly contain the canonical color stream at
    # 1280x720 and the metric depth stream at 848x480, but omit the optional
    # calibration directory.  The batch metadata still gives us a strong
    # device pairing (D435_rgb + D435_depth).  Use a normalized-resolution
    # mapping so the workflow produces a usable depth-lifted preview instead
    # of silently dropping the entire 3D node.  This is deliberately limited
    # to D435 metadata and is marked approximate in the output manifest;
    # precise RGB/depth registration still takes precedence when supplied.
    if _is_d435_pair(pair) and depth is None and color is None:
        return {
            "color_intrinsics": _approx_intrinsics(rw, rh),
            "depth_intrinsics": _approx_intrinsics(dw, dh),
            "depth_to_color": {
                "rotation": np.eye(3).tolist(),
                "translation": [0.0, 0.0, 0.0],
            },
        }, "approximate_d435_resolution_mapping"

    embedded = copy.deepcopy(demo._EMBEDDED_CALIB)
    declared_serial = str((pair.get("device") or {}).get("serial") or "")
    embedded_serial = str(embedded.get("serial") or "")
    if not declared_serial or not embedded_serial \
            or declared_serial != embedded_serial:
        raise ValueError(
            "calibration has no explicit color/depth alignment for "
            f"{pair.get('rgb_source') or 'device'}; "
            "same resolution is not sufficient"
        )
    if depth is not None:
        embedded["depth_intrinsics"] = depth
    embedded_depth = embedded.get("depth_intrinsics") or {}
    embedded_shape = (int(embedded_depth.get("height") or 0),
                      int(embedded_depth.get("width") or 0))
    if embedded_shape != (dh, dw):
        raise ValueError(
            f"no calibration matches depth resolution {dw}x{dh}"
            f" for {pair.get('rgb_source') or 'device'}")
    embedded["color_intrinsics"]["width"] = rw
    embedded["color_intrinsics"]["height"] = rh
    return embedded, raw_path or "embedded_d435_fallback"


def _is_depth_video_source(source_key: str | None, depth_camera: str) -> bool:
    """Distinguish a pure depth slot from an RGB slot sharing its device name."""
    key = str(source_key or "").lower().strip()
    depth_name = str(depth_camera or "").lower().strip()
    if not key:
        return False
    if key == depth_name:
        return True
    if "depth" not in key:
        return False
    # D435_depth_rgb / D435_depth_color are RGB slots, not depth videos.
    return not any(token in key for token in ("rgb", "color", "video"))


def _depth_frame_path(depth_dir: Path, depth_pngs: list[Path],
                      frame_index: int) -> Path | None:
    """Resolve a depth frame for either 000000- or 000001-based recordings."""
    offset = 0 if (depth_dir / "000000.png").exists() else 1
    named = depth_dir / f"{frame_index + offset:06d}.png"
    if named.exists():
        return named
    if 0 <= frame_index < len(depth_pngs):
        candidate = depth_pngs[frame_index]
        expected = frame_index + offset
        if not candidate.stem.isdigit() or int(candidate.stem) == expected:
            return candidate
    return None


def _run_2d_only_pair(demo, pair: dict, video_path: Path, config: dict,
                      ctx: JobContext, out_dir: Path,
                      reason: str) -> ArtifactRef | None:
    """Write a safe per-device 2D artifact when metric 3D is unavailable.

    A missing/ambiguous RGB-depth registration must never be replaced with a
    guessed camera transform.  The RGB detector is still useful for review,
    so keep its normalized keypoints in the same hand_3d namespace while
    explicitly marking all metric 3D points invalid in the manifest.
    """
    import cv2
    import pandas as pd

    cap = cv2.VideoCapture(str(video_path))
    if not cap.isOpened():
        return None
    detector = demo.MediaPipeDetector(
        model_path=str(_MODEL), num_hands=2,
        det_conf=float(config.get("det_conf", 0.4)),
        track_conf=float(config.get("track_conf", 0.4)),
    )
    rows: list[dict] = []
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) or 1
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) or 1
    total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) or 0
    try:
        frame_index = 0
        while True:
            ok, frame = cap.read()
            if not ok:
                break
            hands = detector.detect(frame)
            row: dict[str, Any] = {"frame_index": int(frame_index)}
            for si in (0, 1):
                det = hands[si] if si < len(hands) else None
                if det is None:
                    kp = np.full(63, np.nan, np.float32).tolist()
                    label = ""
                    confidence = 0.0
                    present_2d = False
                else:
                    pts = np.asarray(det.landmarks, np.float32).reshape(-1, 2)
                    kp = np.zeros((21, 3), np.float32)
                    kp[:, :2] = pts / [max(1, width), max(1, height)]
                    kp[:, :2] = np.clip(kp[:, :2], 0.0, 1.0)
                    kp = kp.reshape(63).tolist()
                    label = str(det.label or "")
                    confidence = float(det.score)
                    present_2d = True
                row.update({
                    f"hand_{si}_present": False,
                    f"hand_{si}_2d_present": present_2d,
                    f"hand_{si}_landmarks_3d": np.full(
                        63, np.nan, np.float32).tolist(),
                    f"hand_{si}_keypoints": kp,
                    f"hand_{si}_label": label,
                    f"hand_{si}_reprojection_error": float("nan"),
                    f"hand_{si}_gesture": "",
                    f"hand_{si}_fingers": -1,
                    f"hand_{si}_confidence": confidence,
                })
            rows.append(row)
            frame_index += 1
            if total:
                ctx.progress(min(0.99, frame_index / max(1, total)))
    finally:
        cap.release()
        detector.close()
    if not rows:
        return None

    source_key = str(pair["rgb_source"])
    safe = _safe_source_key(source_key)
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / f"{safe}.parquet"
    pd.DataFrame(rows).to_parquet(out_path, index=False)
    device = pair.get("device") or {}
    manifest = {
        "frames": len(rows),
        "source_key": source_key,
        "rgb_source": source_key,
        "depth_camera": pair.get("depth_source"),
        "device_key": device.get("key"),
        "device_name": device.get("name"),
        "serial": device.get("serial"),
        "detector": "hand_landmarker",
        "mode": "2d_only",
        "unit": "image_normalized",
        "coordinate_frame": "image_normalized",
        "metric_3d_available": False,
        "processing_warnings": [reason],
        "render_video": "",
    }
    (out_dir / f"{safe}.manifest.json").write_text(
        json.dumps(manifest, indent=2), encoding="utf-8")
    return ctx.ref("hand_3d", out_path, source_key=source_key,
                   metadata=manifest)


def _run_rgb_estimated_pair(pair: dict, video_path: Path, config: dict,
                            ctx: JobContext, out_dir: Path,
                            reason: str) -> ArtifactRef | None:
    """Process one depth-failed RGB source with the internal RGB 3D path."""
    import pandas as pd

    # Imported lazily to keep the depth helper independent during normal
    # module discovery.  Both modules are inside processor; this is
    # not a dependency on the legacy Python project.
    from app.processing.modules.stereo_triangulate import (
        _detect_hands, _rgb_frame_rows,
    )

    video_path = Path(video_path)
    rgb_cfg = {
        "max_hands": int(config.get("max_hands", 2)),
        "min_detection_conf": float(config.get("min_detection_conf", 0.1)),
        "min_presence_conf": float(config.get("min_presence_conf", 0.1)),
        "min_tracking_conf": float(config.get("min_tracking_conf", 0.5)),
        "smooth": bool(config.get("smooth", True)),
        "freq_min": float(config.get("freq_min", 5.0)),
        "beta": float(config.get("beta", 0.05)),
        "device": str(config.get("device", "auto")),
    }
    det_df = _detect_hands(video_path, rgb_cfg, ctx.progress)
    rows = _rgb_frame_rows(det_df)
    if not rows:
        return None

    source_key = str(pair["rgb_source"])
    safe = _safe_source_key(source_key)
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / f"{safe}.parquet"
    pd.DataFrame(rows).to_parquet(out_path, index=False)
    device = pair.get("device") or {}
    manifest = {
        "frames": len(rows),
        "source_key": source_key,
        "rgb_source": source_key,
        "depth_camera": pair.get("depth_source"),
        "device_key": device.get("key"),
        "device_name": device.get("name"),
        "serial": device.get("serial"),
        "detector": "hand_landmarker",
        "mode": "rgb_estimated_3d",
        "unit": "rgb_estimated_meters",
        "coordinate_frame": "camera_relative",
        "metric_3d_available": False,
        "method": "MediaPipe RGB landmarks + hand-model PnP + image-scale fallback",
        "camera_model": "approximate_pinhole",
        "processing_warnings": [reason],
        "render_video": "",
    }
    (out_dir / f"{safe}.manifest.json").write_text(
        json.dumps(manifest, indent=2), encoding="utf-8")
    return ctx.ref("hand_3d", out_path, source_key=source_key,
                   metadata=manifest)


def run_depth_hand_3d(ctx: JobContext, video_refs: list, config: dict,
                      *, allow_rgb_fallback: bool = True
                      ) -> tuple[dict[str, ArtifactRef] | None, str | None]:
    """Run depth-image hand 3D once per physical RGB/depth device pair.

    The first returned handle is ``hand_3d`` for old workflows; additional
    devices use ``hand_3d#2``, ``hand_3d#3`` and so on.  The worker runner
    already forwards ``#`` siblings when an edge targets ``hand_3d``, so old
    export graphs continue to work while new exporters can keep each source
    in its own namespace.
    """
    import cv2
    import pandas as pd

    demo = _load_demo()
    if demo is None:
        return None, "depth hand demo not found"
    if not video_refs:
        return None, "no video input"
    configured_depth = str(config.get("depth_camera") or "").strip()
    pairs = _find_device_pairs(ctx, video_refs)
    if configured_depth:
        pairs = [pair for pair in pairs
                 if pair.get("depth_source") == configured_depth]
    if not pairs:
        detail = f"depth/{configured_depth}/" if configured_depth else "depth/*/"
        return None, f"no RGB/depth device pairs at {detail}"

    out_dir = ctx.output_root / "hand_3d"
    out_dir.mkdir(parents=True, exist_ok=True)
    outputs: dict[str, ArtifactRef] = {}
    failures: list[str] = []
    failed_pairs: list[tuple[dict, str]] = []
    successful = 0
    for pair in pairs:
        rgb_ref = pair["rgb_ref"]
        depth_cam = pair.get("depth_source")
        depth_dir = pair.get("depth_dir")
        depth_video = pair.get("depth_video")
        video_path = ctx.resolve(rgb_ref)
        if not video_path or not video_path.exists():
            failures.append(f"{pair['rgb_source']}: video artifact missing")
            continue
        if (depth_dir is None and depth_video is None) or not depth_cam:
            reason = f"{pair['rgb_source']}: metric 3D unavailable (depth pair missing)"
            # Do not terminate the Hand Skeleton workflow with a 2D-only
            # artifact here.  The caller must continue into its standalone
            # RGB camera-relative 3D fallback.
            failures.append(reason)
            failed_pairs.append((pair, reason))
            continue
        depth_pngs = sorted(depth_dir.glob("*.png")) if depth_dir else []
        depth_reader = None
        if depth_video is not None:
            try:
                depth_reader = DepthVideoReader(depth_video)
                first_depth = depth_reader.read()
            except (OSError, RuntimeError, ValueError) as exc:
                if depth_reader is not None:
                    depth_reader.close()
                reason = f"{pair['rgb_source']}: metric 3D unavailable (depth video unreadable: {exc})"
                failures.append(reason)
                failed_pairs.append((pair, reason))
                continue
        else:
            first_depth = cv2.imread(str(depth_pngs[0]), cv2.IMREAD_UNCHANGED) \
                if depth_pngs else None
        if first_depth is None:
            if depth_reader is not None:
                depth_reader.close()
            reason = f"{pair['rgb_source']}: metric 3D unavailable (depth frame unreadable)"
            # Let stereo_triangulate run the RGB-only estimator instead of
            # writing a misleading artifact with no 3D points.
            failures.append(reason)
            failed_pairs.append((pair, reason))
            continue

        cap = cv2.VideoCapture(str(video_path))
        if not cap.isOpened():
            if depth_reader is not None:
                depth_reader.close()
            failures.append(f"{pair['rgb_source']}: cannot open video")
            continue
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) or 0
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) or 0
        total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) or len(depth_pngs)
        try:
            calib, calib_source = _pair_calibration(
                pair, demo, (height, width), first_depth.shape[:2], config)
            driver = DepthHand3DDriver(
                demo, str(_MODEL),
                det_conf=float(config.get("det_conf", 0.4)),
                track_conf=float(config.get("track_conf", 0.4)),
                fill=int(config.get("fill", 1)),
                propagate_max=int(config.get("propagate_max", 15)),
                color_intr=calib["color_intrinsics"],
                depth_to_color=calib["depth_to_color"],
                depth_intr=calib["depth_intrinsics"])
        except (OSError, ValueError, TypeError, KeyError, RuntimeError) as exc:
            cap.release()
            if depth_reader is not None:
                depth_reader.close()
            reason = f"{pair['rgb_source']}: metric 3D unavailable ({exc})"
            # Calibration failure is also an RGB fallback case.
            failures.append(reason)
            failed_pairs.append((pair, reason))
            continue
        rows: list[dict] = []
        try:
            fi = 0
            while True:
                ok, frame = cap.read()
                if not ok:
                    break
                if depth_video is not None:
                    depth = first_depth if fi == 0 else depth_reader.read()
                else:
                    dpath = _depth_frame_path(depth_dir, depth_pngs, fi)
                    depth = (cv2.imread(str(dpath), cv2.IMREAD_UNCHANGED)
                             if dpath is not None else None)
                res = driver.process_frame(demo, frame, depth, fi)
                rows.append(_row(demo, fi, res, frame.shape[1], frame.shape[0]))
                fi += 1
                if total > 0 and fi % 30 == 0:
                    ctx.progress(min(1.0, (successful + fi / total) /
                                     max(1, len(pairs))))
        finally:
            cap.release()
            if depth_reader is not None:
                depth_reader.close()
            stats = dict(driver.stats)
            driver.close()
        if not rows:
            reason = f"{pair['rgb_source']}: no video frames processed"
            failures.append(reason)
            failed_pairs.append((pair, reason))
            continue

        source_key = str(pair["rgb_source"])
        safe = _safe_source_key(source_key)
        out_path = out_dir / f"{safe}.parquet"
        df = pd.DataFrame(rows)
        df.to_parquet(out_path, index=False)
        manifest = {
            "frames": len(rows),
            "source_key": source_key,
            "rgb_source": source_key,
            "depth_camera": depth_cam,
            "device_key": pair["device"].get("key"),
            "device_name": pair["device"].get("name"),
            "serial": pair["device"].get("serial"),
            "detector": "hand_landmarker",
            "mode": "depth_hand_3d",
            "unit": "camera_meters",
            "coordinate_frame": f"{source_key}_color_camera",
            "frame_offset": 1,
            "method": ("mediapipe 2D + direct aligned depth sampling + "
                       "slot tracking + One-Euro (per-camera intrinsics, meters)"),
            "calib_source": calib_source,
            "depth_alignment": (
                "resolution_scaled_same_device"
                if calib_source == "approximate_d435_resolution_mapping"
                else "direct_same_pixel"
            ),
            "stereo_triangulation": False,
            "calibration_quality": (
                "approximate"
                if calib_source == "approximate_d435_resolution_mapping"
                else "measured_or_declared"
            ),
            "processing_warnings": (
                ["RGB/Depth calibration file absent; XY uses normalized "
                 "D435 resolution mapping. Z comes from metric depth."]
                if calib_source == "approximate_d435_resolution_mapping"
                else []
            ),
            "smoothing": {"enabled": True, "freq_min": 3.0, "beta": 0.3},
            "slot_stats": stats,
            "render_video": "",
        }
        if failures:
            manifest["processing_warnings"] = list(failures)
        (out_dir / f"{safe}.manifest.json").write_text(
            json.dumps(manifest, indent=2), encoding="utf-8")
        handle = "hand_3d" if successful == 0 else f"hand_3d#{successful + 1}"
        outputs[handle] = ctx.ref("hand_3d", out_path,
                                  source_key=source_key,
                                  metadata=manifest)
        successful += 1
        print(f"[depth_hand_3d] wrote {len(rows)} frames for {source_key} "
              f"-> {out_path} (stats: {stats})")

    # RGB-only estimated 3D is retained only for legacy callers that explicitly
    # allow it.  The canonical RGB-D_3D workflow passes False: a missing or
    # unreadable depth stream must skip that module instead of creating a
    # misleading metric-looking 3D artifact.
    if not allow_rgb_fallback:
        if not outputs:
            return None, "; ".join(failures) or "no metric depth output"
        reason = "; ".join(failures) if failures else None
        return outputs, reason

    # Preserve successful metric outputs while also covering every RGB source
    # whose depth stream/calibration failed for legacy callers. Previously any
    # successful pair made this function return early at the caller, silently
    # dropping those failed devices from the batch.
    for pair, reason in failed_pairs:
        video_path = ctx.resolve(pair["rgb_ref"])
        if not video_path or not video_path.exists():
            continue
        try:
            fallback = _run_rgb_estimated_pair(
                pair, video_path, config, ctx, out_dir, reason)
        except Exception as exc:
            failures.append(f"{pair['rgb_source']}: RGB fallback failed ({exc})")
            continue
        if fallback is not None:
            handle = ("hand_3d" if successful == 0
                      else f"hand_3d#{successful + 1}")
            outputs[handle] = fallback
            successful += 1

    if not outputs:
        return None, "; ".join(failures) or "no device pair processed"
    reason = "; ".join(failures) if failures else None
    return outputs, reason