# BVH, Feb 2026 # RH20T dataset to AnyData unified format converter. # No dependency on rh20t_api at runtime. Several functions are adapted from # https://github.com/rh20t/rh20t_api -- see per-function docstrings for details. # # Usage: # python anydata/converters/rh20t.py \ # --src /datasets/basile/cv_downloaded_debug/RH20T_d1 \ # --dst /datasets/basile/cv_unified_debug/RH20T_d1 \ # --workers 4 # # Input layout (src/cfg{N}/task_{...}/): # cam_{serial}/color.mp4 # cam_{serial}/timestamps.npy # {color: [timestamps]} # transformed/tcp_base.npy # {serial: [{timestamp, tcp, robot_ft}]} # transformed/gripper.npy # {serial: {timestamp: {gripper_command, gripper_info}}} # metadata.json # {calib: int_ts, ...} # depth/cam_{serial}/depth.mp4 # optional # omniworld_text/cam_{serial}/*.txt # optional VLM captions # {src}/cfg{N}/calib/{ts}/ # cfg-level calibration (all cfgs; cfg1/2 via patch) # intrinsics.npy # {serial: (3,4)} # extrinsics.npy # {serial: [(4,4)]} some serials may be None # tcp.npy # (7,) flexiv/ur5; (6,) franka; (6,) kuka # # Output layout ({dst}/cfg{N}/task_{...}/): # metadata.json # rgb/{cam_name}/*.jpg # depth/{cam_name}/*.npz # float32 meters, key 'depth', 0=invalid; if depth available # lowdim/{cam_name}/*.npz # intrinsics/extrinsics/action per frame import os import json import argparse import numpy as np from argparse import Namespace from glob import glob from functools import partial from scipy.spatial.transform import Rotation from rich.console import Console from anydata.utils.read import read_image, read_npz from anydata.utils.write import write_json, write_lowdim, write_labels from anydata.converters.utils import run, add_key_to_dict, fill_metadata, parse_dst_seq, frame_name, crawl, prepare_lowdim from anydata.converters.utils import extract_frames_from_mp4 console = Console() # Config tables -- subset of fields from rh20t_api/configs/configs.json (verbatim values). RH20T_CONFIGS = { 1: dict( robot='flexiv', in_hand=['043322070878'], tc_mat=np.array([[0,-1,0,0],[1,0,0,0.077],[0,0,1,0.2665],[0,0,0,1]], dtype=np.float64), align_mat_base=np.eye(4, dtype=np.float64), ), 2: dict( robot='flexiv', in_hand=['104422070042'], tc_mat=np.array([[0,-1,0,0],[1,0,0,0.077],[0,0,1,0.2915],[0,0,0,1]], dtype=np.float64), align_mat_base=np.eye(4, dtype=np.float64), ), 3: dict( robot='ur5', in_hand=['045322071843'], tc_mat=np.array([[1,0,0,0.027],[0,1,0,0.0775],[0,0,1,0.240],[0,0,0,1]], dtype=np.float64), align_mat_base=np.array([[-1,0,0,0],[0,-1,0,0],[0,0,1,0],[0,0,0,1]], dtype=np.float64), ), 4: dict( robot='ur5', in_hand=['045322071843'], tc_mat=np.array([[1,0,0,0.027],[0,1,0,0.0775],[0,0,1,0.210],[0,0,0,1]], dtype=np.float64), align_mat_base=np.array([[-1,0,0,0],[0,-1,0,0],[0,0,1,0],[0,0,0,1]], dtype=np.float64), ), 5: dict( robot='franka', in_hand=['104422070042', '135122079702'], tc_mat=np.array([[0,1,0,0.033],[-1,0,0,-0.05],[0,0,1,0.09],[0,0,0,1]], dtype=np.float64), align_mat_base=np.eye(4, dtype=np.float64), ), 6: dict( robot='kuka', in_hand=['135122075425', '135122070361'], tc_mat=np.array([[0,-1,0,0.035],[1,0,0,0.055],[0,0,1,0.20],[0,0,0,1]], dtype=np.float64), align_mat_base=np.eye(4, dtype=np.float64), ), 7: dict( robot='kuka', in_hand=['135122075425', '135122070361'], tc_mat=np.array([[0,-1,0,0.035],[1,0,0,0.055],[0,0,1,0.20],[0,0,0,1]], dtype=np.float64), align_mat_base=np.eye(4, dtype=np.float64), ), } ####################################################### def pose7d_wxyz_to_matrix(tcp7d): """[x,y,z,qw,qx,qy,qz] -> 4x4 homogeneous transform matrix. Reimplemented from rh20t_api/transforms.py::pose_array_quat_2_matrix (uses scipy instead of transforms3d). """ x, y, z, qw, qx, qy, qz = tcp7d R = Rotation.from_quat([qx, qy, qz, qw]).as_matrix() T = np.eye(4, dtype=np.float64) T[:3, :3] = R T[:3, 3] = [x, y, z] return T def tcp_to_7d_xyzqxyzw(tcp, robot): """Convert raw robot tcp to unified 7D [x,y,z,qx,qy,qz,qw]. Inspired by rh20t_api/configurations.py tcp_preprocessor dispatch + tcp_as_q(), merged into a single function for per-frame tcp_base.npy data. 7D [x,y,z,qw,qx,qy,qz]: just reorder quaternion to xyzw convention. 6D: kuka uses bounded euler xyz; all others use rotvec (axis-angle). """ tcp = np.asarray(tcp, dtype=np.float64) if len(tcp) == 7: x, y, z, qw, qx, qy, qz = tcp return np.array([x, y, z, qx, qy, qz, qw], dtype=np.float32) elif len(tcp) == 6: x, y, z = tcp[:3] rvec = tcp[3:6] if robot == 'kuka': # kuka: euler angles bounded to [0, 2pi], normalize to [-pi, pi] euler = np.array([(a + np.pi) % (2 * np.pi) - np.pi for a in rvec]) q_xyzw = Rotation.from_euler('xyz', euler).as_quat() else: # franka / ur5: rotvec (axis-angle) q_xyzw = Rotation.from_rotvec(rvec).as_quat() return np.array([x, y, z, q_xyzw[0], q_xyzw[1], q_xyzw[2], q_xyzw[3]], dtype=np.float32) else: raise ValueError(f'Unexpected tcp length {len(tcp)} for robot {robot}') def preprocess_calib_tcp(raw_tcp, robot): """Preprocess raw calib/tcp.npy to 7D [x,y,z,qw,qx,qy,qz]. Adapted from rh20t_api/configurations.py::{franka,kuka}_tcp_preprocessor + tcp_as_q(), and scene.py::_load_calib_tcp(). flexiv/ur5: already 7D, returned as-is. franka: 6D rotvec, apply axis-flip correction matrix, extract quat. kuka: 6D euler in mm, convert mm to m, bound to [-pi,pi], extract quat. """ tcp = np.asarray(raw_tcp, dtype=np.float64) if robot in ('flexiv', 'ur5'): if len(tcp) == 7: return tcp # already [x,y,z,qw,qx,qy,qz] # TODO(bvh): handle 6D flexiv calib tcp (affects 1,322 cfg2 episodes, 5 late-2021 # calib timestamps >= 1639*). Format is [x,y,z,rx,ry,rz] rotvec, same as franka # but WITHOUT the axis-flip correction. Fix: plain rotvec-to-quat: # q_xyzw = Rotation.from_rotvec(tcp[3:6]).as_quat() # return np.array([*tcp[:3], q_xyzw[3], q_xyzw[0], q_xyzw[1], q_xyzw[2]]) raise ValueError(f'{robot} calib_tcp: expected 7D, got {len(tcp)}') elif robot == 'franka': # franka_tcp_preprocessor applies: mat @ [[1,0,0],[0,-1,0],[0,0,-1]] correction correction = np.array([[1,0,0,0],[0,-1,0,0],[0,0,-1,0],[0,0,0,1]], dtype=np.float64) T = np.eye(4, dtype=np.float64) T[:3, :3] = Rotation.from_rotvec(tcp[3:6]).as_matrix() T[:3, 3] = tcp[:3] T = T @ correction q_xyzw = Rotation.from_matrix(T[:3, :3]).as_quat() return np.array([T[0, 3], T[1, 3], T[2, 3], q_xyzw[3], q_xyzw[0], q_xyzw[1], q_xyzw[2]]) elif robot == 'kuka': xyz = tcp[:3] / 1000.0 # mm to m def bound_0_2pi(a): return (a + 2 * np.pi) if a < 0 else a euler_b = np.array([bound_0_2pi(tcp[3]), bound_0_2pi(tcp[4]), bound_0_2pi(tcp[5])]) euler = np.array([(a + np.pi) % (2 * np.pi) - np.pi for a in euler_b]) q_xyzw = Rotation.from_euler('xyz', euler).as_quat() return np.array([xyz[0], xyz[1], xyz[2], q_xyzw[3], q_xyzw[0], q_xyzw[1], q_xyzw[2]]) else: raise ValueError(f'Unknown robot type: {robot}') def raw_tcp_to_matrix(raw_tcp, robot): """Convert raw per-frame TCP (from tcp_base.npy) to 4x4 homogeneous matrix. Handles the same robot-specific formats as tcp_to_7d_xyzqxyzw: 7D [x,y,z,qw,qx,qy,qz]: flexiv, ur5 6D [x,y,z,rx,ry,rz]: franka (rotvec), kuka (euler xyz, bounded) """ tcp = np.asarray(raw_tcp, dtype=np.float64) if len(tcp) == 7: return pose7d_wxyz_to_matrix(tcp) # [x,y,z,qw,qx,qy,qz] elif len(tcp) == 6: T = np.eye(4, dtype=np.float64) T[:3, 3] = tcp[:3] if robot == 'kuka': euler = np.array([(a + np.pi) % (2 * np.pi) - np.pi for a in tcp[3:6]]) T[:3, :3] = Rotation.from_euler('xyz', euler).as_matrix() else: T[:3, :3] = Rotation.from_rotvec(tcp[3:6]).as_matrix() return T else: raise ValueError(f'Unexpected tcp length {len(tcp)} for robot {robot}') def compute_extrinsics_base_aligned(ext_marker, in_hand_serials, tc_mat, calib_tcp_7d, align_mat_base): """Compute base-aligned cam2world (Twc) extrinsics for EXTERNAL cameras only. Adapted from rh20t_api/scene.py::extrinsics_base_aligned property. The API formula produces world2cam (Tcw); we invert to get cam2world (Twc) to match the unified format convention. In-hand cameras are excluded: their extrinsics change per-frame with TCP motion and must be computed separately using compute_in_hand_extrinsic(). Returns {serial: (4,4) float32 cam2world} or None if in_hand camera not found in calib. """ ih_set = set(in_hand_serials) in_hand = next((s for s in in_hand_serials if s in ext_marker), None) if in_hand is None: return None calib_tcp_mat = pose7d_wxyz_to_matrix(calib_tcp_7d) base_world = ( np.linalg.inv(ext_marker[in_hand]) @ tc_mat @ np.linalg.inv(calib_tcp_mat) ) # API gives world2cam (Tcw); invert to cam2world (Twc) for unified format return { serial: np.linalg.inv(mat @ base_world @ align_mat_base).astype(np.float32) for serial, mat in ext_marker.items() if serial not in ih_set # in-hand cameras get per-frame extrinsics } def compute_in_hand_extrinsic(raw_tcp, robot, tc_mat, align_mat_base): """Compute per-frame cam2world (Twc) extrinsic for an in-hand camera. The API formula gives world2cam (Tcw): Tcw(t) = tc_mat @ inv(tcp_mat(t)) @ align_mat_base We invert to get cam2world (Twc) for the unified format. Returns (4,4) float32 cam2world, or None if raw_tcp is None. """ if raw_tcp is None: return None tcp_mat = raw_tcp_to_matrix(raw_tcp, robot) Tcw = tc_mat @ np.linalg.inv(tcp_mat) @ align_mat_base return np.linalg.inv(Tcw).astype(np.float32) ####################################################### # Task descriptions (native, from rh20t.github.io/static/task_description.json) _TASK_DESC = None def get_task_description(task_id): """Return English task description for a task_id like 'task_0001', or None.""" global _TASK_DESC if _TASK_DESC is None: json_path = os.path.join(os.path.dirname(__file__), 'language', 'rh20t_task_description.json') if os.path.exists(json_path): with open(json_path) as f: _TASK_DESC = json.load(f) else: _TASK_DESC = {} return (_TASK_DESC.get(task_id) or {}).get('task_description_english') def _build_language(task_desc, omniworld_texts): """Build the language metadata dict. Priority: native task description (imperative) > OmniWorld captions (fallback). orig_task - raw native task description string (if available) prompt - best imperative description: orig_task if available, else omniworld texts source - 'orig' | 'omniworld' omniworld - list of OmniWorld VLM captions (if available, always stored separately) """ if not task_desc and not omniworld_texts: return None lang = {} if task_desc: lang['orig_task'] = task_desc lang['prompt'] = task_desc lang['source'] = 'orig' else: lang['prompt'] = omniworld_texts lang['source'] = 'omniworld' if omniworld_texts: lang['omniworld'] = omniworld_texts return lang def extract_depth_from_rh20t_mp4(mp4_path, dst_dir): """Extract and decode depth frames from RH20T special-encoded depth.mp4. Adapted from rh20t_api/extract.py::convert_depth() -- same encoding scheme, but uses ffmpeg+PIL instead of cv2 (no ffmpeg backend in this env), and outputs float32 meters in npz rather than uint16 PNG. Omits the L515 camera 4x multiplier (is_l515 check) present in the API. Encoding: each video frame is 2*H x W (grayscale stored as BGR, all channels equal). - Top H rows = LOW byte of uint16 depth - Bottom H rows = HIGH byte of uint16 depth depth_mm = high * 256 + low (millimeters; 0 = invalid, 65535 = invalid) Uses ffmpeg for decoding (OpenCV cannot open this h264 variant). Saves float32 meters as {frame_name(i)}.npz with key 'depth' (0.0 = invalid). Returns ((H, W), n_frames) or (None, 0) if no frames. """ import subprocess, tempfile, shutil from PIL import Image tmp_dir = tempfile.mkdtemp(prefix='rh20t_depth_') try: pattern = os.path.join(tmp_dir, '%010d.png') cmd = ['ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', mp4_path, pattern] subprocess.run(cmd, check=True, capture_output=True, timeout=120) raw_files = sorted(glob(os.path.join(tmp_dir, '*.png'))) if not raw_files: return None, 0 os.makedirs(dst_dir, exist_ok=True) hw = None for n, src in enumerate(raw_files): img = np.array(Image.open(src).convert('L'), dtype=np.uint16) h2 = img.shape[0] // 2 low = img[:h2] high = img[h2:] depth_mm = high * np.uint16(256) + low depth_m = depth_mm.astype(np.float32) / 1000.0 depth_m[(depth_mm == 0) | (depth_mm == 65535)] = 0.0 np.savez_compressed(os.path.join(dst_dir, f'{frame_name(n)}.npz'), depth=depth_m) if hw is None: hw = (h2, img.shape[1]) return hw, len(raw_files) finally: shutil.rmtree(tmp_dir, ignore_errors=True) _calib_cache = {} def load_calib(cfg_dir, calib_ts, robot): """Load and cache cfg-level calibration for one timestamp. Adapted from rh20t_api/scene.py::_load_calib() + _load_calib_tcp(), with added robustness for invalid/None entries and empty-string serials. Returns (intrinsics_dict, ext_marker_dict, calib_tcp_7d) or (None, None, None). """ key = (cfg_dir, calib_ts) if key in _calib_cache: return _calib_cache[key] calib_dir = os.path.join(cfg_dir, 'calib', str(calib_ts)) intr_path = os.path.join(calib_dir, 'intrinsics.npy') extr_path = os.path.join(calib_dir, 'extrinsics.npy') tcp_path = os.path.join(calib_dir, 'tcp.npy') if not os.path.exists(intr_path): result = (None, None, None) _calib_cache[key] = result return result intr_raw = np.load(intr_path, allow_pickle=True).item() # {serial: (3,4)} extr_raw = np.load(extr_path, allow_pickle=True).item() # {serial: list[(4,4)]} raw_tcp = np.load(tcp_path) # (7,) or (6,) intrinsics = {str(s): np.array(v, dtype=np.float32) for s, v in intr_raw.items()} ext_marker = {} for s, v in extr_raw.items(): if not str(s): continue # skip empty-string serial keys (data artifact) arr = np.array(v, dtype=np.float64) if arr.ndim == 3: arr = arr[0] # (1,4,4) to (4,4) if arr.shape != (4, 4): continue # skip None/invalid entries (stored as NaN scalar in some cfgs) ext_marker[str(s)] = arr calib_tcp_7d = preprocess_calib_tcp(raw_tcp, robot) result = (intrinsics, ext_marker, calib_tcp_7d) _calib_cache[key] = result return result ####################################################### def load_tcp_and_ft_lookup(ep_dir): """Load transformed/tcp_base.npy and return both tcp and force-torque lookups. Returns: tcp_lookup: {serial: [(timestamp, tcp_array)]} ft_lookup: {serial: [(timestamp, ft_6d_array)]} -- empty if robot_ft absent Falls back to tcp.npy (no ft) if tcp_base.npy is not present. """ for name in ('tcp_base.npy', 'tcp.npy'): path = os.path.join(ep_dir, 'transformed', name) if os.path.exists(path): break else: return {}, {} data = np.load(path, allow_pickle=True).item() tcp_lookup = {} ft_lookup = {} for serial, frames in data.items(): tcp_lookup[str(serial)] = [(float(f['timestamp']), np.array(f['tcp'])) for f in frames] if frames and 'robot_ft' in frames[0]: ft_lookup[str(serial)] = [ (float(f['timestamp']), np.array(f['robot_ft'], dtype=np.float32)) for f in frames ] # TODO: also load joint.npy for joint-angle proprioception. # Structure: {serial: {timestamp_int: ndarray}} # cfg3/ur5 = 6D joint angles (rad); cfg1/cfg5 (flexiv/franka) = 14D (7 pos + 7 vel). # cfg1/cfg2 joint.npy comes from patch/ (merged into transformed/ by the download script). # Needs timestamp matching analogous to gripper_lookup. return tcp_lookup, ft_lookup def load_gripper_lookup(ep_dir): """Load transformed/gripper.npy as {serial: [(timestamp, float_value)]}.""" path = os.path.join(ep_dir, 'transformed', 'gripper.npy') if not os.path.exists(path): return {} data = np.load(path, allow_pickle=True).item() result = {} for serial, ts_dict in data.items(): pairs = [] for ts, d in ts_dict.items(): cmd = d.get('gripper_command', [0]) val = float(cmd[0]) if isinstance(cmd, (list, np.ndarray)) else float(cmd) pairs.append((float(ts), val)) pairs.sort() result[str(serial)] = pairs return result def nearest_value(pairs, target_ts): """Return value from [(timestamp, value)] nearest to target_ts. None if empty.""" if not pairs: return None return min(pairs, key=lambda x: abs(x[0] - target_ts))[1] def load_omniworld_text(ep_dir, serial, omniworld_root=None): """Load OmniWorld VLM captions for a camera. Returns list of strings. Supports two layouts: - episode-local (debug download): {ep_dir}/omniworld_text/cam_{serial}/*.txt - OmniWorld root (full download): {omniworld_root}/{shard}/RH20T/RH20T_{cfg_dir}/{ep_name}/cam_{serial}/text/*.txt where cfg_dir is e.g. 'cfg3', so RH20T_cfg3; ep_name is the episode folder name. """ def _read_txts(paths): out = [] for p in sorted(paths): with open(p) as f: t = f.read().strip() if t: out.append(t) return out # Episode-local layout (debug download or per-episode download) text_dir = os.path.join(ep_dir, 'omniworld_text', f'cam_{serial}') if os.path.exists(text_dir): return _read_txts(glob(os.path.join(text_dir, '*.txt'))) # OmniWorld-root layout (bulk sync from S3) if omniworld_root: cfg_dir = os.path.basename(os.path.dirname(ep_dir)) # e.g. 'cfg3' ep_name = os.path.basename(ep_dir) cfg_s3 = f'RH20T_{cfg_dir}' # e.g. 'RH20T_cfg3' pattern = os.path.join(omniworld_root, '*', 'RH20T', cfg_s3, ep_name, f'cam_{serial}', 'text', '*.txt') return _read_txts(glob(pattern)) return [] ####################################################### def get_camera_names(serials, in_hand_serials): """Map serial to canonical name: in_hand_N or ext1, ext2, ... Always uses in_hand_1, in_hand_2, ... (never bare 'in_hand') so that single and dual in-hand configs share a consistent naming scheme. """ ih_set = set(in_hand_serials) ihs = [s for s in serials if s in ih_set] exts = sorted(s for s in serials if s not in ih_set) name_map = {} for i, s in enumerate(ihs): name_map[s] = f'in_hand_{i + 1}' for i, s in enumerate(exts): name_map[s] = f'ext{i + 1}' return name_map ####################################################### def get_sequences(args): """Find all episode dirs (containing metadata.json) under src/cfg*/task_*/.""" pattern = os.path.join(args.src, 'cfg*', 'task_*', 'metadata.json') return sorted(os.path.dirname(p) for p in glob(pattern)) def parse_sequence(seq, args): return parse_dst_seq(seq, args) ####################################################### def load_camera_timestamps(seq, serial): """Load color timestamps for a camera. Returns float64 array (ms) or None.""" ts_path = os.path.join(seq, f'cam_{serial}', 'timestamps.npy') if not os.path.exists(ts_path): return None ts_data = np.load(ts_path, allow_pickle=True) if ts_data.dtype == object: ts_data = ts_data.item() return np.array(ts_data.get('color', []), dtype=np.float64) return ts_data.astype(np.float64) def build_time_grid(all_cam_ts, fps=10.0): """Build a uniform time grid over the strict intersection of all cameras. t_start = latest first-timestamp across cameras (all cameras have started). t_end = earliest last-timestamp across cameras (no camera has ended yet). Output indices always start at 0. Returns grid_ts (float64 array of timestamps in ms), or empty if no overlap. """ t_start = max(ts[0] for ts in all_cam_ts) t_end = min(ts[-1] for ts in all_cam_ts) if t_end < t_start: return np.array([], dtype=np.float64) interval_ms = 1000.0 / fps grid_ts = np.arange(t_start, t_end + interval_ms / 2, interval_ms) return grid_ts def map_camera_to_grid(cam_ts, grid_ts): """For each grid point, find the nearest source frame. Since the grid is built over the strict time intersection of all cameras, every camera is guaranteed to have data near every grid point. Returns a list of (grid_idx, src_frame_idx) pairs, one per grid point. """ if cam_ts is None or len(cam_ts) == 0: return [] mapping = [] for gi, gt in enumerate(grid_ts): idx = np.searchsorted(cam_ts, gt) if idx == 0: best = 0 elif idx >= len(cam_ts): best = len(cam_ts) - 1 else: if abs(cam_ts[idx - 1] - gt) <= abs(cam_ts[idx] - gt): best = idx - 1 else: best = idx mapping.append((gi, best)) return mapping def process_sequence(i, seq, dst, args): """Convert one RH20T episode to AnyData unified format. Uses timestamp-based temporal synchronization: a global 10Hz time grid is built from all cameras' timestamps, and each camera's frames are placed at the grid index matching their real-world time. Cameras that start late or end early have no files at the corresponding grid indices. """ import shutil # Config cfg_dir = os.path.dirname(seq) cfg_num = int(os.path.basename(cfg_dir).replace('cfg', '')) cfg = RH20T_CONFIGS[cfg_num] robot = cfg['robot'] in_hand_serials = cfg['in_hand'] # Human episodes have no robot action data; they get different tags is_human = os.path.basename(seq).endswith('_human') # Episode metadata with open(os.path.join(seq, 'metadata.json')) as f: ep_meta = json.load(f) calib_ts = ep_meta.get('calib') # Calibration (all cfgs; cfg1/cfg2 calib comes from the patch/ download merged into local dir) intrinsics_calib = None ext_base = None if calib_ts is not None: intr, ext_marker, calib_tcp_7d = load_calib(cfg_dir, calib_ts, robot) if intr is not None: intrinsics_calib = intr ext_base = compute_extrinsics_base_aligned( ext_marker, in_hand_serials, cfg['tc_mat'], calib_tcp_7d, cfg['align_mat_base'], ) # Cameras cam_dirs = sorted(glob(os.path.join(seq, 'cam_*'))) serials = [os.path.basename(c).replace('cam_', '') for c in cam_dirs] if not serials: raise ValueError(f'[yellow] SKIP[/yellow] no cameras: {seq}') name_map = get_camera_names(serials, in_hand_serials) # Action data (robot episodes only; human episodes have no tcp/gripper) tcp_lookup = {} ft_lookup = {} gripper_lookup = {} action_serial = None if not is_human: tcp_lookup, ft_lookup = load_tcp_and_ft_lookup(seq) gripper_lookup = load_gripper_lookup(seq) action_serial = next((s for s in in_hand_serials if s in tcp_lookup), None) if action_serial is None: action_serial = next(iter(tcp_lookup), None) # Language: native task description + OmniWorld VLM captions ep_name = os.path.basename(seq) task_id = ep_name.split('_user_')[0] # e.g. 'task_0001' task_desc = get_task_description(task_id) lang_texts = load_omniworld_text(seq, in_hand_serials[0], omniworld_root=getattr(args, 'omniworld_root', None)) # ===== Phase 1: Extract all frames to staging, collect timestamps ===== staging_dir = os.path.join(dst, '_staging') cam_info = {} # serial -> {cam_name, hw, n_rgb, cam_ts, staging_rgb, ...} ### ---- Process each camera ---- cameras = [name_map[cam] for cam in serials] num_frames = {cam: dict() for cam in cameras} resolution = {cam: dict() for cam in cameras} labels, lowdim = [], {} dense_labels = ['rgb','depth'] for serial in serials: cam_name = name_map[serial] color_mp4 = os.path.join(seq, f'cam_{serial}', 'color.mp4') if not os.path.exists(color_mp4): continue cam_ts = load_camera_timestamps(seq, serial) # Extract RGB to staging stg_rgb = os.path.join(staging_dir, cam_name, 'rgb') hw, n_rgb = extract_frames_from_mp4(color_mp4, stg_rgb, ext='jpg') if cam_ts is not None and len(cam_ts) != n_rgb: # Timestamps and frame count mismatch -- truncate to shorter n_rgb = min(n_rgb, len(cam_ts)) cam_ts = cam_ts[:n_rgb] # Extract depth to staging (if available) depth_mp4 = os.path.join( seq.replace(f'cfg{cfg_num}', f'cfg{cfg_num}_depth'), f'cam_{serial}', 'depth.mp4', ) stg_depth = None hw_d = None n_depth = 0 if os.path.exists(depth_mp4): stg_depth = os.path.join(staging_dir, cam_name, 'depth') hw_d, n_depth = extract_depth_from_rh20t_mp4(depth_mp4, stg_depth) # Intrinsics from calib, scaled to match extracted RGB resolution is_in_hand = serial in set(in_hand_serials) intr_mat = None if intrinsics_calib and serial in intrinsics_calib: intr_raw = intrinsics_calib[serial][:3, :3].astype(np.float64) cx_calib, cy_calib = intr_raw[0, 2], intr_raw[1, 2] calib_w, calib_h = round(2 * cx_calib), round(2 * cy_calib) rgb_h, rgb_w = hw if calib_w > 0 and calib_h > 0 and (calib_w != rgb_w or calib_h != rgb_h): sx, sy = rgb_w / calib_w, rgb_h / calib_h intr_raw[0, 0] *= sx intr_raw[1, 1] *= sy intr_raw[0, 2] *= sx intr_raw[1, 2] *= sy intr_mat = intr_raw.astype(np.float32) # Static extrinsics for external cameras extr_mat = None if not is_in_hand and ext_base and serial in ext_base: extr_mat = ext_base[serial].astype(np.float32) cam_info[serial] = dict( cam_name=cam_name, hw=hw, n_rgb=n_rgb, cam_ts=cam_ts, stg_rgb=stg_rgb, stg_depth=stg_depth, hw_d=hw_d, n_depth=n_depth, is_in_hand=is_in_hand, intr_mat=intr_mat, extr_mat=extr_mat, ) if not cam_info: raise ValueError(f'[yellow] SKIP[/yellow] no valid cameras: {seq}') # ===== Phase 2: Build global time grid & per-camera mappings ===== # Collect all timestamp arrays (only cameras that have timestamps) all_cam_ts = [ci['cam_ts'] for ci in cam_info.values() if ci['cam_ts'] is not None and len(ci['cam_ts']) > 0] if not all_cam_ts: raise ValueError(f'[yellow] SKIP[/yellow] no timestamps: {seq}') grid_ts = build_time_grid(all_cam_ts, fps=10.0) n_grid = len(grid_ts) if n_grid == 0: raise ValueError(f'[yellow] SKIP[/yellow] no temporal overlap between cameras: {seq}') # Build per-camera grid mappings for serial, ci in cam_info.items(): ci['grid_map'] = map_camera_to_grid(ci['cam_ts'], grid_ts) # ===== Phase 3: Write temporally aligned frames ===== cameras_out = [] # Action lookup lists (shared across cameras) tcp_pairs = tcp_lookup.get(action_serial, []) if action_serial else [] ft_pairs = ft_lookup.get(action_serial, []) if action_serial else [] gripper_pairs = gripper_lookup.get(action_serial, []) if action_serial else [] for serial, ci in cam_info.items(): dense = {label: dict() for label in dense_labels} cam_name = ci['cam_name'] grid_map = ci['grid_map'] if not grid_map: continue cameras_out.append(cam_name) has_depth = ci['stg_depth'] is not None and ci['n_depth'] > 0 # Create output dirs rgb_out = os.path.join(dst, 'rgb', cam_name) for gi, src_fi in grid_map: grid_name = frame_name(gi) src_name = frame_name(src_fi) ts = float(grid_ts[gi]) # Link RGB frame from staging to final output src_rgb = os.path.join(ci['stg_rgb'], f'{src_name}.jpg') dst_rgb = os.path.join(rgb_out, f'{grid_name}.jpg') if os.path.exists(src_rgb): dense['rgb'][grid_name] = np.array(read_image(src_rgb)) # Link depth frame if has_depth and src_fi < ci['n_depth']: src_d = os.path.join(ci['stg_depth'], f'{src_name}.npz') if os.path.exists(src_d): dense['depth'][grid_name] = read_npz(src_d)['depth'] # Lowdim raw_ts = float(ci['cam_ts'][src_fi]) if ci['cam_ts'] is not None else ts filename_lowdim = add_key_to_dict(lowdim, f'{dst}/lowdim/{cam_name}/{grid_name}.npz') lowdim[filename_lowdim].update(dict( camera=cam_name, timestep=np.int32(gi), timestamp_grid_ms=np.float64(ts), timestamp_raw_ms=np.float64(raw_ts), )) if not is_human: ts_next = float(grid_ts[gi + 1]) if gi + 1 < n_grid else ts + 100.0 raw_tcp = nearest_value(tcp_pairs, ts) raw_tcp_next = nearest_value(tcp_pairs, ts_next) state_tcp = (tcp_to_7d_xyzqxyzw(raw_tcp, robot) if raw_tcp is not None else np.zeros(7, dtype=np.float32)) action_tcp = (tcp_to_7d_xyzqxyzw(raw_tcp_next, robot) if raw_tcp_next is not None else state_tcp.copy()) state_grip = np.float32(nearest_value(gripper_pairs, ts) or 0.0) action_grip = np.float32(nearest_value(gripper_pairs, ts_next) or state_grip) action_dict = dict( state_tcp=state_tcp, state_gripper=state_grip, action_tcp=action_tcp, action_gripper=action_grip, ) raw_ft = nearest_value(ft_pairs, ts) if raw_ft is not None: action_dict['state_ft'] = raw_ft lowdim[filename_lowdim]['action'] = action_dict if ci['intr_mat'] is not None: lowdim[filename_lowdim]['intrinsics'] = ci['intr_mat'] # Per-frame extrinsics: static for ext cameras, TCP-derived for in-hand if ci['is_in_hand'] and not is_human and ext_base is not None: raw_tcp_cur = nearest_value(tcp_pairs, ts) extr_frame = compute_in_hand_extrinsic( raw_tcp_cur, robot, cfg['tc_mat'], cfg['align_mat_base']) # Fall back to identity if TCP not available (e.g. padded edge frames) lowdim[filename_lowdim]['extrinsics'] = (extr_frame if extr_frame is not None else np.eye(4, dtype=np.float32)) elif ci['extr_mat'] is not None: lowdim[filename_lowdim]['extrinsics'] = ci['extr_mat'] else: lowdim[filename_lowdim]['extrinsics'] = np.eye(4, dtype=np.float32) # Write dense labels write_labels(dst, cam_name, args.storage, dense, labels, resolution, num_frames) # Clean up staging directory shutil.rmtree(staging_dir, ignore_errors=True) # num_frames = total grid length (individual cameras cover subsets of this range) num_frames_out = n_grid # Per-camera timing info: raw range vs grid intersection grid_start_ms = float(grid_ts[0]) grid_end_ms = float(grid_ts[-1]) grid_dur_ms = grid_end_ms - grid_start_ms timing = {} for serial, ci in cam_info.items(): if ci['cam_ts'] is None or len(ci['cam_ts']) == 0: continue cam_start = float(ci['cam_ts'][0]) cam_end = float(ci['cam_ts'][-1]) cam_dur = cam_end - cam_start pct = round(100.0 * grid_dur_ms / cam_dur, 1) if cam_dur > 0 else 100.0 timing[ci['cam_name']] = dict( raw_start_ms=round(cam_start, 1), raw_end_ms=round(cam_end, 1), raw_duration_ms=round(cam_dur, 1), grid_pct=pct, ) ######## WRITE LOWDIM write_lowdim(args, dst, labels, num_frames, lowdim) # Episode metadata.json ep_tags = ['real', 'manipulation', 'human'] if is_human else ['real', 'manipulation', 'robotics'] ep_tags += [f'cfg{cfg_num}', robot] seq_meta = fill_metadata( args=args, info=dict( name='RH20T', tags=ep_tags, raw_id=os.path.relpath(seq, args.src), cfg=cfg_num, robot=robot, episode_type='human' if is_human else 'robot', ), labels=labels, cameras=cameras_out, resolution=resolution, num_frames=num_frames_out, framerate=10, rgb=dict(extension='jpg') if 'rgb' in labels else None, intrinsics=dict(model='pinhole') if intrinsics_calib else None, extrinsics=dict(metric=True, world='robot_base', transform='cam2world') if ext_base else None, depth=dict(extension='npz', metric=True, sparse=False) if 'depth' in labels else None, semantic=None, action=(dict(format='absolute', tcp='[x,y,z,qx,qy,qz,qw]', ft='[fx,fy,fz,tx,ty,tz] wrist frame (robot_ft from tcp_base.npy)') if ('action' in labels) else None), language=(_build_language(task_desc, lang_texts)), specific=dict( rating=ep_meta.get('rating'), calib_quality=ep_meta.get('calib_quality'), timing=dict( grid_start_ms=round(grid_start_ms, 1), grid_end_ms=round(grid_end_ms, 1), grid_duration_ms=round(grid_dur_ms, 1), grid_frames=n_grid, cameras=timing, ), ), ) write_json(os.path.join(dst, 'metadata.json'), seq_meta) return dst ####################################################### def _worker(item, args): i, seq = item try: process_sequence(i, seq, args) except Exception as e: import traceback console.print(f'[red]ERROR[/red] {os.path.basename(seq)}: {e}') console.print(traceback.format_exc()) ####################################################### # def main(): # parser = argparse.ArgumentParser(description='RH20T to AnyData unified format converter') # parser.add_argument('--src', required=True, # help='Source root (contains cfg1/, cfg2/, ...)') # parser.add_argument('--dst', required=True, help='Output root') # parser.add_argument('--workers', type=int, default=4) # parser.add_argument('--overwrite', action='store_true', # help='Reprocess already-done episodes') # parser.add_argument('--first', type=int, default=None, # help='Limit to first N episodes (for debugging)') # parser.add_argument('--depth-root', default=None, # help='Root for bulk-synced depth data. Contains cfg{N}_depth/ dirs. ' # 'Defaults to None (uses episode-local depth/ if present).') # parser.add_argument('--omniworld-root', default=None, # help='Root for bulk-synced OmniWorld text. Contains {shard}/RH20T/... ' # 'Defaults to None (uses episode-local omniworld_text/ if present).') # args = parser.parse_args() # seqs = get_sequences(args) # if args.first: # seqs = seqs[:args.first] # n_robot = sum(1 for s in seqs if not os.path.basename(s).endswith('_human')) # n_human = len(seqs) - n_robot # console.rule('[bold]RH20T Converter[/bold]') # console.print(f' src: {args.src}') # console.print(f' dst: {args.dst}') # console.print(f' workers: {args.workers}') # console.print(f' episodes: {len(seqs)} ({n_robot} robot, {n_human} human)') # if not seqs: # console.print('[red]No episodes found.[/red]') # return # if args.workers <= 1: # for i, seq in enumerate(seqs): # console.print(f' [{i + 1}/{len(seqs)}] {os.path.relpath(seq, args.src)}') # try: # process_sequence(i, seq, args) # except Exception as e: # import traceback # console.print(f'[red]ERROR[/red] {seq}: {e}\n{traceback.format_exc()}') # else: # from multiprocessing import Pool # items = list(enumerate(seqs)) # fn = partial(_worker, args=args) # with Pool(args.workers) as pool: # pool.map(fn, items) # # Generate metadata_shared.json, coverage.json, and split_all.json # from anydata.converters.utils import create_metadata_shared, create_coverage # from anydata.converters.misc.create_split import create_split # create_metadata_shared(args.dst, pattern='cfg*/task_*/metadata.json', # dataset_name='RH20T') # create_coverage(args.dst) # create_split(Namespace( # src=args.dst, name='split_all', subfolder=None, # download=False, upload=False, delete=False, # webbed=False, quiet=False, # subset=None, subset_slice=None, subset_random=None, # )) # console.rule('[bold green]Done[/bold green]') # if __name__ == '__main__': # main() ####################################################### if __name__ == '__main__': converter = os.path.basename(__file__) run(converter, get_sequences, parse_sequence, process_sequence) #######################################################