Spaces:
Running
Running
File size: 3,903 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 | """Stereo RGB-D Camera — two RGB views plus one paired depth stream.
This is a distinct input contract from ``stereo_camera``. The latter only
publishes left/right RGB video, while this module also publishes a typed
``Depth`` output that can be connected to the RGB-D 3D hand modules.
The concrete camera name remains a source key in the run snapshot; the card
label is the stable device category used by the workflow editor.
"""
from app.processing import ProcessingModule, JobContext, ArtifactRef, field
from app.processing.registry import register
from app.processing.theme import INPUT_COLOR
@register
class StereoRGBDCameraModule(ProcessingModule):
slug = "stereo_rgbd_camera"
version = "1.0"
category = "input"
label = "Stereo RGB-D Camera"
icon = "ant-design:video-camera-outlined"
color = INPUT_COLOR
outputs = (
{"key": "video_left", "label": "Left RGB Video"},
{"key": "video_right", "label": "Right RGB Video"},
{"key": "depth", "label": "Depth"},
)
default_config = {
"source_keys": "", "source_key": "", "position": "", "fps": 30,
}
config_schema = (
field("source_keys", "string",
"Device names (comma; blank = auto-detect)", ""),
field("source_key", "string", "Source key", ""),
)
execution_target = "server"
capabilities = ("video_input", "depth", "rgbd", "stereo")
def run(self, ctx: JobContext) -> dict[str, ArtifactRef] | None:
videos = ctx.find_videos()
if len(videos) < 2:
ctx.skip("Stereo RGB-D Camera requires left and right RGB video — skipped")
from app.lerobot_v21 import canonical_source_key, source_key_from_video
actual_keys = [source_key_from_video(path, ctx.input_root / "videos")
for path in videos[:2]]
raw_keys = ctx.config.get("source_keys")
if isinstance(raw_keys, str):
keys = [key.strip() for key in raw_keys.split(",") if key.strip()]
elif isinstance(raw_keys, (list, tuple)):
keys = [str(key).strip() for key in raw_keys if str(key).strip()]
else:
keys = []
keys = [canonical_source_key(key) for key in keys]
refs = [
ctx.ref("video", path,
source_key=(keys[index] if index < len(keys)
else actual_keys[index]))
for index, path in enumerate(videos[:2])
]
# Use the same physical-device pairing logic as RGB-D 3D processing.
# A real v2.1 depth video (or a legacy PNG directory) can produce the
# typed Depth output; this card never fabricates depth from RGB.
from app.processing.modules.depth_hand_3d import _find_device_pairs
try:
pairs = _find_device_pairs(ctx, refs)
except (OSError, RuntimeError, ValueError) as exc:
ctx.skip(f"Stereo RGB-D Camera depth stream is unavailable ({exc}) — skipped")
pair = next((item for item in pairs
if (item.get("depth_video") is not None
or item.get("depth_dir") is not None)
and item.get("depth_source")), None)
if pair is None:
return {"video_left": refs[0], "video_right": refs[1]}
depth_path = pair.get("depth_video") or pair.get("depth_dir")
depth_ref = ctx.ref(
"depth", depth_path, source_key=pair.get("depth_source"),
metadata={
"rgb_source": pair.get("rgb_source"),
"depth_source": pair.get("depth_source"),
"device_key": (pair.get("device") or {}).get("key"),
"device_name": (pair.get("device") or {}).get("name"),
},
)
return {
"video_left": refs[0],
"video_right": refs[1],
"depth": depth_ref,
}
|