Spaces:
Running
Running
File size: 2,897 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 | """RGB-D Camera — generic input for a color stream with depth metadata.
The physical device category is intentionally separate from the uploaded
stream/source key. A D435 or a future RGB-D camera can therefore use the
same workflow card while the selected source key continues to identify the
real files on disk.
"""
from app.processing import ProcessingModule, JobContext, ArtifactRef, field
from app.processing.registry import register
from app.processing.theme import INPUT_COLOR
@register
class RGBDCameraModule(ProcessingModule):
slug = "rgbd_camera"
version = "1.0"
category = "input"
label = "RGB-D Camera"
icon = "ant-design:video-camera-outlined"
color = INPUT_COLOR
outputs = ({"key": "video", "label": "RGB Video"},
{"key": "depth", "label": "Depth"})
default_config = {"source_key": "", "fps": 30}
config_schema = (
field("source_key", "string",
"Device name / source key (blank = auto-detect)", ""),
)
execution_target = "server"
capabilities = ("video_input", "depth", "rgbd")
def run(self, ctx: JobContext) -> dict[str, ArtifactRef] | None:
videos = ctx.find_videos()
if not videos:
ctx.skip("Configured RGB-D video source is missing — skipped")
from app.lerobot_v21 import canonical_source_key, source_key_from_video
source_key = ctx.config.get("source_key") or None
source_key = source_key or source_key_from_video(
videos[0], ctx.input_root / "videos")
source_key = canonical_source_key(source_key)
video_ref = ctx.ref("video", videos[0], source_key=source_key or None)
# Publish the actual depth stream as a typed artifact. LeRobot v2.1
# stores metric depth as a video; the legacy PNG directory remains a
# read-only compatibility path for old batches.
depth_ref = None
try:
from app.processing.modules.depth_hand_3d import _find_device_pairs
pair = next(iter(_find_device_pairs(ctx, [video_ref])), None)
depth_path = None
if pair:
depth_path = pair.get("depth_video") or pair.get("depth_dir")
if depth_path is not None and depth_path.exists():
depth_ref = ctx.ref(
"depth", depth_path, source_key=pair.get("depth_source"),
metadata={"rgb_source": source_key,
"depth_source": pair.get("depth_source")})
except Exception as exc:
# Pairing is validated again by the RGB-D processing module; an
# unavailable sidecar must not make the input card crash.
print(f"[rgbd_camera] depth artifact discovery skipped: {exc}")
outputs = {"video": video_ref}
if depth_ref is not None:
outputs["depth"] = depth_ref
return outputs
|