Spaces:
Running
Running
File size: 8,999 Bytes
d646dd7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | """RGB_TO_2D_BlackGlove — black-glove RGB-only keypoint workflow.
This module deliberately never searches for or reads a depth stream. It uses
the black-glove detector for 2D landmarks and the shared RGBWorldTracker to
place those landmarks in a camera-relative estimated 3D coordinate frame.
The depth-based counterpart is ``RGB-D_3D_BlackGlove``.
"""
from __future__ import annotations
import json
from typing import Any
from app.processing import ArtifactRef, JobContext, ProcessingModule, field
from app.processing.registry import register
from app.processing.theme import HAND3D_COLOR
from app.processing.modules.black_glove_hand import (
_hand_3d_rows,
_safe_key,
run_local,
)
def _config_schema() -> tuple[dict[str, Any], ...]:
return (
field("mode", "select", "Input mode", "auto",
options=["auto", "mono", "stereo"]),
field("max_hands", "number", "Max hands", 2, min=1, max=2),
field("det_conf", "number", "Glove detection confidence", 0.05,
min=0.01, max=1, step=0.01),
field("device", "select", "Detector device", "auto",
options=["auto", "cpu", "cuda"]),
field("pose_device", "select", "Pose device", "auto",
options=["auto", "cpu", "cuda"]),
field("pose_backend", "select", "Pose backend", "rtmpose",
options=["rtmpose", "mediapipe"]),
field("pose_model", "string", "MediaPipe task path (optional)", ""),
field("imgsz", "number", "Detector image size", 640,
min=160, max=1280, step=32),
field("smooth", "boolean", "One-Euro smoothing", True),
field("preview_3d", "boolean", "Display-only 3D preview", True),
field("freq_min", "number", "Smoothing cutoff Hz", 5.0,
min=1, max=60, step=1),
field("beta", "number", "Smoothing speed coefficient", 0.05,
min=0, max=2, step=0.05),
field("use_tracker", "boolean", "Track hands between frames", True),
field("movement_thresh", "number", "Pose refresh movement threshold (px)",
1.5, min=0, max=20, step=0.5),
field("skip_timeout", "number", "Maximum skipped pose frames", 3,
min=1, max=30, step=1),
field("box_alpha", "number", "Tracking box smoothing", 0.7,
min=0, max=1, step=0.05),
field("pose_conf_thr", "number", "Pose confidence threshold", 0.15,
min=0, max=1, step=0.05),
field("pose_box_raw", "boolean", "Use raw box for pose crop", False),
field("hold_translate", "boolean", "Compensate held pose motion", True),
field("new_track_conf", "number", "New track confidence gate", 0.1,
min=0, max=1, step=0.01),
field("lost_timeout", "number", "Lost track timeout (frames)", 8,
min=1, max=60, step=1),
field("hold_max", "number", "Low-confidence hold limit (frames)", 12,
min=1, max=120, step=1),
field("spawn_confirm", "number", "New box confirm frames", 2,
min=1, max=5, step=1),
field("match_contain_thr", "number", "Cross-hand box reject ratio (0=off)", 0.7,
min=0, max=1, step=0.05),
)
@register
class BlackHandRGB3DModule(ProcessingModule):
slug = "rgb_to_2d_black_glove"
version = "1.3"
category = "process"
label = "RGB_TO_2D_BlackGlove"
icon = "ant-design:deployment-unit-outlined"
color = HAND3D_COLOR
inputs = ({"key": "video", "label": "RGB Video"},)
# The RGB-derived hand_3d parquet is retained only for the 3D preview.
# Downstream workflow nodes receive the public 2D contract.
outputs = ({"key": "hand_keypoints", "label": "Hand 2D"},)
default_config = {
"mode": "auto", "max_hands": 2, "det_conf": 0.05,
"device": "auto", "pose_device": "auto", "imgsz": 640,
"pose_backend": "rtmpose", "pose_model": "",
"smooth": True, "freq_min": 5.0, "beta": 0.05,
"preview_3d": True,
"use_tracker": True, "movement_thresh": 1.5,
"skip_timeout": 3, "box_alpha": 0.7, "pose_conf_thr": 0.15,
"pose_box_raw": False, "hold_translate": True,
"new_track_conf": 0.1,
"lost_timeout": 8, "hold_max": 12, "spawn_confirm": 2,
"match_contain_thr": 0.7,
}
config_schema = _config_schema()
execution_target = "worker"
capabilities = ("black_glove", "hand_keypoints", "rgb_preview_3d",
"camera_relative", "video_overlay",
"mono", "stereo", "no_depth")
def run(self, ctx: JobContext) -> dict[str, ArtifactRef] | None:
import pandas as pd
video_refs = [ref for ref in ctx.incoming.values() if ref.kind == "video"]
if not video_refs:
ctx.skip("No RGB video input — skipped")
video_refs.sort(key=lambda ref: (ref.source_key or "", ref.path or ""))
mode = str(ctx.config.get("mode") or "auto")
if mode == "mono":
video_refs = video_refs[:1]
elif mode == "stereo":
video_refs = video_refs[:2]
else:
video_refs = video_refs[:2]
output_kp = ctx.output_root / "hand_keypoints"
output_video = ctx.output_root / "skeleton"
output_3d = ctx.output_root / "hand_3d"
preview_3d = bool(ctx.config.get("preview_3d", True))
if preview_3d:
output_3d.mkdir(parents=True, exist_ok=True)
out: dict[str, ArtifactRef] = {}
for index, ref in enumerate(video_refs):
path = ctx.resolve(ref)
if not path or not path.exists():
ctx.skip(f"RGB video artifact missing: {ref.path}")
source_key = str(ref.source_key or path.stem)
key = _safe_key(ref.source_key, path.stem)
# depth_sampler is intentionally omitted. run_local creates the
# RGBWorldTracker path when this argument is None.
result = run_local(
path, output_kp, output_video, ctx.config,
lambda value, i=index, n=len(video_refs):
ctx.progress((i + value) / n),
depth_sampler=None,
keypoint_name=f"{key}.parquet",
)
preview_3d = bool(ctx.config.get("preview_3d", True))
result["manifest"].update({
"source_key": source_key,
"mode": "black_glove_rgb_estimated_3d" if preview_3d else "black_glove_rgb_2d",
"unit": "rgb_estimated_meters" if preview_3d else "image_normalized",
"coordinate_frame": "camera_relative" if preview_3d else "image_normalized",
"world_coordinates": False,
"metric_3d_available": False,
"preview_3d": preview_3d,
"method": ("YOLO-World + "
+ str(ctx.config.get("pose_backend", "rtmpose"))
+ " 2D keypoints + "
"RGB hand-model PnP and image-scale fallback with "
"stable local hand geometry and One-Euro smoothing"),
"orientation_mode": "stable_local_model_wrist_pnp_only",
})
(result["parquet"].parent / f"{key}.manifest.json").write_text(
json.dumps(result["manifest"], indent=2), encoding="utf-8")
keypoint_handle = "hand_keypoints" if index == 0 else f"hand_keypoints#{index + 1}"
out[keypoint_handle] = ctx.ref(
"hand_keypoints", result["parquet"], source_key=source_key,
metadata=result["manifest"])
skeleton_handle = "skeleton_video" if index == 0 else f"skeleton_video#{index + 1}"
out[skeleton_handle] = ctx.ref(
"video", result["video"], source_key=source_key,
metadata={**result["manifest"], "skeleton": True})
if preview_3d:
world_path = output_3d / f"{key}.parquet"
# run_local already applies the RGB One-Euro filter. The
# helper also supports depth lifting and would otherwise
# apply the depth-only temporal smoother a second time to
# RGB estimates.
rgb_3d_config = {**ctx.config, "depth_smooth": False}
world_rows = _hand_3d_rows(
result["rows"], result["manifest"]["fps"], rgb_3d_config)
pd.DataFrame(world_rows).to_parquet(world_path, index=False)
world_manifest = {
**result["manifest"],
"artifact": "black_hand_rgb_3d",
"views": len(video_refs),
"view": "left" if index == 0 else "right",
"preview_3d": True,
}
(output_3d / f"{key}.manifest.json").write_text(
json.dumps(world_manifest, indent=2), encoding="utf-8")
if ctx.progress:
ctx.progress(1.0)
return out
|