Spaces:
Running
Running
File size: 5,268 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 | """Camera/video grouping — 恢复双目(左右目)语义。
上传的 zip 里一台双目相机产生多个独立 mp4(videos/stereo_left/…、
videos/stereo_right/…),扁平 camera 列表会丢失"组"的语义,前端只能
把它们当成互不相关的视频。本模块按命名模式把扁平 camera 聚合为:
- 双目组(stereo):left + right 配对,``*_aux`` 为辅助流
- 单目组(single):无配对的独立 camera(如 head_left_rgb)
新旧数据兼容:新上传数据在 ``meta.camera_groups`` 里携带分组结果,
旧数据没有该字段,由 media-groups API 用本模块实时推导。
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
# 后缀从长到短,避免 stereo_left_aux 被误配为 ("stereo_left", "aux")
_ROLE_SUFFIXES = (
("_left_aux", "left_aux"),
("_right_aux", "right_aux"),
("_aux_left", "left_aux"),
("_aux_right", "right_aux"),
("_left", "left"),
("_right", "right"),
("_aux", "aux"),
)
def _role_of(name: str) -> tuple[str, str]:
"""提取 (前缀, 角色): stereo_left_aux → ("stereo", "left_aux")"""
normalized = str(name).strip().lower()
for suffix, role in _ROLE_SUFFIXES:
if normalized.endswith(suffix):
return normalized[: -len(suffix)].rstrip("_"), role
return normalized, "primary"
def _member(name: str, role: str, stream: dict) -> dict:
# 注意:role 必须最后赋值 —— stream 里可能带上游 _camera_role 的粗糙
# role(如 stereo_left_aux → "left"),解包会错误覆盖推导出的 left_aux。
return {"source_key": name, **(stream or {}), "role": role}
def group_camera_streams(camera_streams: dict[str, Any]) -> dict[str, Any]:
"""把扁平 ``{camera: stream}`` 聚合为 groups + singles。
输入例如::
{"stereo_left": {"path": "...", "frame_count": 258, "fps": 25},
"stereo_right": {...}, "stereo_left_aux": {...}, "stereo_right_aux": {...}}
返回::
{"groups": [{"id": "stereo", "type": "stereo", "label": "stereo 双目",
"members": [left, right], "aux": [left_aux, right_aux]}],
"singles": [{"source_key": "head_left_rgb", "role": "primary", ...}]}
"""
parsed: dict[str, list[tuple[str, str, dict]]] = {}
for name, stream in (camera_streams or {}).items():
prefix, role = _role_of(name)
parsed.setdefault(prefix, []).append((name, role, stream or {}))
groups: list[dict] = []
singles: list[dict] = []
for prefix, members in parsed.items():
roles = {role for _, role, _ in members}
has_left = "left" in roles or "left_aux" in roles
has_right = "right" in roles or "right_aux" in roles
if has_left and has_right:
group: dict[str, Any] = {
"id": prefix,
"type": "stereo",
"label": prefix,
"members": [],
"aux": [],
}
for name, role, stream in members:
entry = _member(name, role, stream)
if role in ("left", "right"):
group["members"].append(entry)
else:
group["aux"].append(entry)
groups.append(group)
else:
for name, role, stream in members:
singles.append(_member(name, role, stream))
return {"groups": groups, "singles": singles}
def stereo_device_label(session_dir: Path | None, fallback: str,
episode_id: str | None = None) -> str:
"""从 ``calibration/*.json`` 找双目设备名作为组标签。
``head_stereo.json`` (type=stereo_rgbd_camera) → "head_stereo 双目",
找不到则回退为前缀命名。
"""
if session_dir is not None:
calib_dir = Path(session_dir) / "calibration"
if episode_id:
namespaced = Path(session_dir) / "meta" / "calibration" / str(episode_id)
if namespaced.is_dir():
calib_dir = namespaced
if not calib_dir.is_dir():
calib_dir = Path(session_dir) / "meta" / "calibration"
if calib_dir.is_dir():
for f in sorted(calib_dir.glob("*.json")):
try:
data = json.loads(f.read_text(encoding="utf-8"))
except Exception:
continue
if str(data.get("type", "")).startswith("stereo"):
name = data.get("name") or f.stem
if name:
return name
return fallback
def has_skeleton_video(session_dir: Path | None, camera: str,
episode_id: str | None = None) -> bool:
"""探测批次内是否已有骨骼叠加视频(worker 输出)。
手部叠加由前端将 parquet 关键点绘制到原始 RGB 视频上，因此这里不再
探测或保留第二份骨骼视频。
"""
if session_dir is None:
return False
return False
def video_dir_has_skeleton(video_root: Path) -> bool:
"""``videos/<camera>/`` 目录下是否有骨骼叠加视频。"""
if not video_root.is_dir():
return False
return any("_skeleton" in p.name for p in video_root.glob("*_skeleton*.mp4"))
|