# Copyright 2026 Toyota Research Institute. All rights reserved. import os import sys import torch import ffmpeg import argparse import subprocess import numpy as np import pandas as pd import multiprocessing from glob import glob from tqdm import tqdm from argparse import Namespace from scipy.spatial.transform import Rotation from copy import deepcopy from anydata.utils.write import create_folder from anydata.sync.sync_utils import rm_sync S3_PATH = 's3://tri-ml-sandbox-16011-us-west-2-datasets' # TODO @vitor remove hardcoded path # Canonical metadata label sort order (loosely the IKEDOCLABSN grouping, but this is the # metadata ordering, NOT the webdataset naming string) CANONICAL_LABEL_ORDER = ['rgb', 'depth', 'pointmap', 'raymap', 'intrinsics', 'extrinsics', 'language', 'action', 'semantics', 'masks', 'lowdim'] # TODO: write_jpg_from_mp4() also exists, maybe merge to avoid code duplication def extract_frames_from_mp4(video_path, dst_folder, ext='jpg'): """Extract all frames from an MP4 via ffmpeg. Returns (resolution, num_frames). ffmpeg outputs 1-indexed files; renames to 0-indexed ``frame_name()`` format. Supports ext='jpg' (lossy, default) or ext='png' (lossless, for depth). """ from PIL import Image os.makedirs(dst_folder, exist_ok=True) pattern = f'{dst_folder}/%010d.{ext}' ffmpeg_opts = ['-q:v', '2'] if ext == 'jpg' else [] cmd = ['ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', video_path] + ffmpeg_opts + [pattern] subprocess.run(cmd, check=True, capture_output=True, timeout=600) raw_files = sorted(glob(f'{dst_folder}/*.{ext}')) n_frames = len(raw_files) if n_frames == 0: raise RuntimeError(f'Zero frames extracted from: {video_path}') for idx, src in enumerate(raw_files): target = f'{dst_folder}/{frame_name(idx)}.{ext}' if src != target: os.rename(src, target) img = Image.open(f'{dst_folder}/{frame_name(0)}.{ext}') resolution = (img.height, img.width) return resolution, n_frames def extract_mp4_shape(video): probe = ffmpeg.probe(video) video_info = next(s for s in probe['streams'] if s['codec_type'] == 'video') width, height = int(video_info['width']), int(video_info['height']) num_frames = int(video_info['nb_frames']) return (num_frames, height, width) def extract_mp4(video, start_frame=0, end_frame=1e12, copy=False): """Extracts a numpy array from an mp4 file (NHW3, uint8)""" probe = ffmpeg.probe(video) video_info = next(s for s in probe['streams'] if s['codec_type'] == 'video') width, height = int(video_info['width']), int(video_info['height']) out, _ = (ffmpeg .input(video) .filter('trim', start_frame=start_frame, end_frame=end_frame) .output('pipe:', format='rawvideo', pix_fmt='rgb24') .run(capture_stdout=True, capture_stderr=True) ) video = (np .frombuffer(out, np.uint8) .reshape([-1, height, width, 3]) ) if copy: return np.copy(video) else: return video from anydata.utils.misc import frame_name from anydata.utils.read import read_yaml, read_json from anydata.utils.write import write_json from anydata.utils.colmap import read_cameras_binary, read_images_binary, read_points3d_binary, qvec2rotmat from anydata.utils.colmap import read_cameras_text, read_images_text, read_points3D_text from anydata.utils.geometry import invert_extrinsics from anydata.sync.sync_utils import get_seqs_subset, create_tar, aws_s3_cp, aws_s3_rm, aws_s3_sync, remove_path from anydata.geometry.camera import Camera def clean_string(data): for key in ':-"$&%^*#()[]\' ': data = data.replace(key, '_') # Replace certain characters with _ return data def add_key_to_dict(data, key1, key2=None, key3=None): """Adds a new key to a dictionary (the value of that key is an empty dictionary)""" if key1 not in data: data[key1] = dict() if key2 is not None: if key2 not in data[key1]: data[key1][key2] = dict() if key3 is not None: if key3 not in data[key1][key2]: data[key1][key2][key3] = dict() return key1 def loop_over(src): """Iterate over a folder recursively until it finds metadata files""" folders = sorted(glob(f'{src}/*')) if f'{src}/metadata.json' not in folders: new_folders = [] for f in list(folders): new_folders.extend(loop_over(f)) return new_folders else: return [f'{src}/metadata.json'] def list_s3(s3_path, suffix=None): """Lists files from s3, given a path and an optional suffix for filtering""" proc = subprocess.Popen(f'aws s3 ls {s3_path}/', stdout=subprocess.PIPE, shell=True) (data, _) = proc.communicate() data = data.decode().replace('/\n', ' ').replace('\n', ' ').replace(' ', ' ').split(' ') if suffix is not None: data = [d for d in data if d.endswith(suffix)] else: data = [d for d in data if d not in ['','PRE']] proc.kill() return data def list_s3_recursive(s3_path, suffix=None): """Lists files from s3 recursively, given a path and an optional suffix for filtering""" proc = subprocess.Popen(f'aws s3 ls {s3_path}/ --recursive', stdout=subprocess.PIPE, shell=True) (data, _) = proc.communicate() data = data.decode().replace('/\n', ' ').replace('\n', ' ').replace(' ', ' ').split(' ') if suffix is not None: if isinstance(suffix, list): data = [d for d in data if any([d.endswith(s) for s in suffix])] else: data = [d for d in data if d.endswith(suffix)] proc.kill() return data def get_splits(seqs, args): """Returns splits for multi-thread processing, given a sequence list""" n_seqs = len(seqs) if n_seqs < args.num_procs: args.num_procs = len(seqs) splits = np.linspace(0, n_seqs, args.num_procs + 1) return [int(s) for s in splits] def simplify(data, remove): """Simplify dictionary if all values are the same (recursively)""" while isinstance(data, dict): keys = list(data.keys()) for key in keys: if data[key] == 0 and data[key] is not False: remove.append(key) data.pop(key) valid = all([data[keys[i]] == data[keys[0]] for i in range(len(keys))]) if valid: data = data[keys[0]] else: return data return data def check_dict(data, remove=[], simplify_first=False): """Check dictionary for simplification""" if data is None: return data if not isinstance(data, dict): return data # Return data if not dictionary keys = list(data.keys()) if len(keys) == 0: return data # Return data if empty if isinstance(data[keys[0]], dict): data = {key: simplify(data[key], remove) for key in keys} # Simplify root dict if requested return simplify(data, remove) if simplify_first else data def remove_none(data): """Remove None values from a dictionary""" if isinstance(data, dict): return {key: remove_none(val) for key, val in data.items() if val is not None} else: return data def parse_dst_seq(seq, args, remove=[], remext=False, separate=None, derepeat=True): # Change path from src to dst seq = seq.replace(args.src, args.dst) seq = clean_string(seq) # Replace certain characters with _ if separate is not None: seq = '/'.join(seq.split(separate)) # Remove certain nested folders from the name if requested if len(remove) > 0: base_len = len(args.dst.split('/')) seq_split = seq.split('/') remove = [len(seq_split) + r if r < 0 else r + base_len for r in remove] seq_split = [seq_split[i] for i in range(len(seq_split)) if i not in remove] seq = '/'.join(seq_split) # Remove extension if requested if remext: seq = os.path.splitext(seq)[0] if derepeat: split = seq.split('/') if split[-1] == split[-2]: split = split[:-1] seq = '/'.join(split) # If overwriting, always process if args.overwrite: return seq, None, 'process' # Create tmp file for bookeeping (maximum file name length of 128) seq_tmp = '/'.join(seq.split('/')[5:]).replace('/', '__') if len(seq_tmp) > 128: seq_tmp = seq_tmp[-128:] tmp = f'{args.dst}/tmp/{seq_tmp}.txt' # If tmp file exists, something went wrong, process again if os.path.exists(tmp): return seq, tmp, 'process' # If metadata seq exists, it has been processed already if os.path.exists(seq): # Single sequence per episode if os.path.exists(f'{seq}/metadata.json'): return [seq], tmp, 'done' # Multiple sequences per episode else: folders = glob(f'{seq}/*') meta = glob(f'{seq}/**/metadata.json', recursive=True) if len(folders) == len(meta): return folders, tmp, 'done' # New sequence, create temp file and start processing os.makedirs(os.path.dirname(tmp), exist_ok=True) with open(tmp, 'w') as fp: pass return seq, tmp, 'process' def read_colmap_binary(path): cameras_binary = read_cameras_binary(f'{path}/sparse/0/cameras.bin') points_binary = read_points3d_binary(f'{path}/sparse/0/points3D.bin') images_binary = read_images_binary(f'{path}/sparse/0/images.bin') return cameras_binary, points_binary, images_binary def read_dense_colmap_binary(path): cameras_binary = read_cameras_binary(f'{path}/cameras.bin') points_binary = read_points3d_binary(f'{path}/points3D.bin') images_binary = read_images_binary(f'{path}/images.bin') return cameras_binary, points_binary, images_binary def read_colmap_text(path): cameras = read_cameras_text(f'{path}/cameras.txt') points = read_points3D_text(f'{path}/points3D.txt') images = read_images_text(f'{path}/images.txt') return cameras, points, images def geometry_from_colmap(val, cameras_binary, points_binary, downsample=1, geometry=None): camera_binary = cameras_binary[val.camera_id] intrinsics = np.array(camera_binary.params) if geometry == 'pinhole': intrinsics = intrinsics[:4] # Force pinhole model if camera_binary.model == 'SIMPLE_RADIAL': f, cx, cy = intrinsics[:3] intrinsics = np.array([[f, 0.0, cx], [0.0, f, cy], [0.0, 0.0, 1.0]]) hw = camera_binary.height, camera_binary.width if downsample > 1: hw = [v // downsample for v in hw] intrinsics[:4] /= downsample rot = qvec2rotmat(val.qvec).astype(np.float32) tvec = np.array(val.tvec, dtype=np.float32) extrinsics = np.vstack([ np.hstack((rot, np.expand_dims(tvec, axis=1))), np.array([0.0, 0.0, 0.0, 1.0]) ]).astype(np.float32) points = np.stack([pt.xyz for id, pt in points_binary.items() if id in val.point3D_ids]) cam = Camera( K=torch.tensor(intrinsics).unsqueeze(0).float(), Twc=torch.tensor(extrinsics).unsqueeze(0).float(), hw=hw, ) points = torch.tensor(points).permute(1, 0).unsqueeze(0).float() depth = cam.project_pointcloud(points)[0,0].numpy() extrinsics = invert_extrinsics(extrinsics) return intrinsics, extrinsics, depth, hw def prepare_lowdim(lowdim, dst, cam, frame): """Include basic information to lowdim dict""" filename_lowdim = add_key_to_dict(lowdim, f'{dst}/lowdim/{cam}/{frame}.npz') lowdim[filename_lowdim]['camera'] = cam lowdim[filename_lowdim]['timestep'] = int(frame) ####################################################### def parse_args(converter): """Shared functionality for parsing arguments""" parser = argparse.ArgumentParser() parser.add_argument("path", type=str) parser.add_argument('--official', action='store_true') parser.add_argument("--num_procs", type=int, default=16) parser.add_argument("--language", type=str, default=None) parser.add_argument('--upload', action='store_true') parser.add_argument('--upload_and_delete', action='store_true') parser.add_argument('--restart', action='store_true') parser.add_argument('--overwrite', action='store_true') parser.add_argument('--local_path', type=str, default=os.environ.get('ANYDATA_LOCAL_ROOT', '/data')) parser.add_argument('--s3_path', type=str, default=S3_PATH) parser.add_argument("--folder", type=str, default='cv_unified') parser.add_argument('--subset', type=str, default=None) parser.add_argument('--first', type=int, default=None) parser.add_argument('--cleanup', action='store_true') parser.add_argument('--resume', action='store_true') parser.add_argument('--refresh', action='store_true') parser.add_argument('--webbed', action='store_true') parser.add_argument('--download', action='store_true') sto = parser.add_mutually_exclusive_group(required=True) sto.add_argument('--frames', action='store_const', dest='storage', const='frames') sto.add_argument('--videos', action='store_const', dest='storage', const='videos') args = parser.parse_args() # Cleaning up a webbed dataset if args.webbed: args.folder = 'cv_webbed' # Refresh requires restart if args.refresh: args.restart = True if args.subset is not None and not args.subset.endswith('0'): args.refresh = False # Cleanup requires no subset if args.cleanup: if args.restart: sys.exit() args.subset = None args.restart = True # Cleanup requires no subset if args.resume: args.restart = True # Remove subset if it's a single part if args.subset == '1/0': args.subset = None # Determine storage mode args.folder += f'/{args.storage}' # Add debug to folder if not official (be careful when using official!) if not args.official: args.folder += '_debug' # Processed datasets are saved somewhere else if converter == 'processed.py': args.path = args.path.replace('cv_downloaded', 'cv_processed') # Remove trailing '/' from path if args.path.endswith('/'): args.path = args.path[:-1] # Prepare language dictionary if converter is not None and args.language is None: # Check if there is a yaml language file default_language = f'anydata/converters/language/{os.path.splitext(converter)[0]}.yaml' if os.path.exists(default_language): args.language = default_language if converter is not None and args.language is None: # Check if there is a csv language file default_language = f'anydata/converters/language/{os.path.splitext(converter)[0]}.csv' if os.path.exists(default_language): args.language = default_language # Prepare source and destination paths if args.webbed: args.src = args.path args.dst = f'{args.local_path}/{args.folder}/{args.path}' else: args.src, args.dst = args.path, os.path.basename(args.path) args.dst = f'{args.local_path}/{args.folder}/{args.dst}' if args.cleanup: tmp = glob(f'{args.dst}/tmp_split_all_*.json') if len(tmp) > 0: print('#######################') print('### Temporary files found, exiting....') print('#######################') return # Load language dictionary if converter is not None and args.language is not None: if args.language.endswith('.yaml'): # yaml file args.language = read_yaml(args.language)['language_dict'] elif args.language.endswith('.csv'): # csv file args.language = pd.read_csv(args.language) language = dict() for _, row in args.language.iterrows(): d = row.to_dict() language[d['Unnamed: 2']] = d['Unnamed: 4'] args.language = language return args def post_process(dst, tmp, args, delete_tmp=True): """Handles uploading and temporary files""" if args.upload or args.upload_and_delete: ### Temporary and s3 locations tmp_folder = f'{args.dst}/tmp' os.makedirs(tmp_folder, exist_ok=True) s3_dst = dst.replace(args.local_path, args.s3_path) if args.storage == 'videos': ### Sync the entire folder as is aws_s3_sync(dst, s3_dst, robust=True) else: ### Get label folders label_folders = glob(f'{dst}/*') label_folders = [f for f in label_folders if not f.endswith('.json')] ### Loop over label folders for label_folder in label_folders: ### Get filenames tarname = label_folder.replace('/','__') + '.tar.gz' if len(tarname) > 128: tarname = tarname[-128:] # Limit filename length label = os.path.basename(label_folder) ### Create tarfile create_tar(f'{args.dst}/tmp/{tarname}', label_folder) ### Upload tarfile to s3 aws_s3_cp(f'{args.dst}/tmp/{tarname}', f'{s3_dst}/{label}.tar.gz', robust=True) ### Delete tarfile os.remove(f'{args.dst}/tmp/{tarname}') ### Upload metadata to s3 aws_s3_cp(f'{dst}/metadata.json', f'{s3_dst}/metadata.json', robust=True) # Done, delete temporary file if tmp is not None and delete_tmp: os.remove(tmp) ####################################################### def process_sequences(i, seqs, args): """Shared functionality for processing sequences""" suc, err, don = 0, 0, 0 progress = tqdm(seqs, ncols=96, leave=False) dataset = args.dst.split('/')[-1] for seq in progress: dst, tmp, reason = args.parse_sequence_fn(seq, args) try: # Prepare description description = f'| SED: {suc}/{err}/{don}' if args.subset is not None: description = f': {args.subset} {description}' progress.set_description(f'# {dataset} {description}') # Skip if has been processed already if reason == 'done': for d in dst: args.log_suc[os.path.relpath(d, args.dst)] = 'done' don += 1 continue dst_orig = dst if args.download: root = os.path.dirname(os.path.dirname(args.src)) remote = args.s3_path + seq[len(root):] aws_s3_sync(remote, seq, quiet=True, robust=True) dst = args.process_sequence_fn(i, seq, dst, args) if isinstance(dst, list) and len(dst) == 0: args.log_err[os.path.relpath(dst_orig, args.dst)] = 'NO VALID SEQUENCES WERE CREATED' err += 1 continue # Post-process sequence or sequences if isinstance(dst, list): for j, d in enumerate(dst): post_process(d, tmp, args, delete_tmp=(j==len(dst) - 1)) args.log_suc[os.path.relpath(d, args.dst)] = 'success' suc += 1 else: post_process(dst, tmp, args) args.log_suc[os.path.relpath(dst, args.dst)] = 'success' suc += 1 if args.download: remove_path(seq) except Exception as e: args.log_err[os.path.relpath(dst, args.dst)] = str(e) err += 1 ####################################################### def clean_metadata(args, scratch=False, suc=None, err=None): """Extract shared metadata and write metadata_shared.json. Per-sequence metadata files are left intact (redundancy provides clarity; shared metadata serves as defaults that local fields may override). """ # Based share metadata metadata_shared_path = f'{args.dst}/metadata_shared.json' # Get all sequence-specific metadata files files = crawl(args.dst, 'metadata.json') seqs = [file[len(args.dst)+1:-len('/metadata.json')] for file in files] # Check if successes and metadata match if suc is not None: suc_match = set(suc.keys()) == set(seqs) if not suc_match: # Each episode as multiple sequences and some failed mid-unification if err is None: err = dict() for d in list(set(seqs).difference(set(suc.keys()))): suc[d] = 'success' # Some were successful all_metadata = {file: read_json(file) for file in files} all_metadata_seqs = {key[len(args.dst)+1:-len('/metadata.json')]: val for key, val in all_metadata.items()} if not scratch and os.path.exists(metadata_shared_path) and not args.overwrite: # Start from available shared metadata metadata_shared = read_json(metadata_shared_path) else: # Start from first sequence metadata metadata_shared = deepcopy(all_metadata[files[0]]) metadata_shared.pop('specific', None) tags_available = [] # List of all available tags labels_available = [] # List of all available lables cameras_available = [] # List of all available cameras removed = {key: [] for key in metadata_shared.keys()} # Keep track of removed keys shared_tags = None shared_labels = None shared_cameras = None # Check if all metadata files share the same value and keep those in shared metadata for key in list(metadata_shared.keys()): if key not in metadata_shared: continue for val1 in all_metadata.values(): if val1 is None: continue # Skip if value is None if key == 'info': tags_available.extend(val1[key]['tags']) tags_available = list(set(tags_available)) if shared_tags is None: shared_tags = val1[key]['tags'] else: shared_tags = [l for l in shared_tags if l in val1[key]['tags']] if key == 'cameras': # Add cameras to availables cameras_available.extend(val1[key]) cameras_available = list(set(cameras_available)) if shared_cameras is None: shared_cameras = val1[key] else: shared_cameras = [c for c in shared_cameras if c in val1[key]] if key == 'labels': # Add labels to availables labels_available.extend(val1[key]) labels_available = list(set(labels_available)) if shared_labels is None: shared_labels = val1[key] else: shared_labels = [l for l in shared_labels if l in val1[key]] if key not in metadata_shared: continue if key not in val1: metadata_shared.pop(key); continue if isinstance(metadata_shared[key], dict): # Recursive check if is dict if not isinstance(val1[key], dict): metadata_shared.pop(key) else: for key2 in list(val1[key].keys()): # Add key to shared if hasn't been seen before (and track log to avoid repetition) if key2 not in removed[key] and key2 not in metadata_shared[key]: metadata_shared[key][key2] = val1[key][key2] if key not in metadata_shared: continue if key2 not in metadata_shared[key]: continue # Remove key if doesn't match previous (and log so it's not added again) if metadata_shared[key][key2] != val1[key][key2]: removed[key].append(key2) metadata_shared[key].pop(key2) else: # Direct check if not dict if key not in metadata_shared: continue if key not in val1: continue if metadata_shared[key] != val1[key]: metadata_shared.pop(key) # Remove empty keys after filtering for key in list(metadata_shared.keys()): if isinstance(metadata_shared[key], dict) and len(metadata_shared[key]) == 0: metadata_shared.pop(key) # Store labels and cameras shared across all samples ordered_shared_labels = [l for l in CANONICAL_LABEL_ORDER if l in shared_labels] ordered_shared_labels += sorted(l for l in shared_labels if l not in CANONICAL_LABEL_ORDER) metadata_shared['shared_tags'] = sorted(shared_tags) metadata_shared['shared_labels'] = ordered_shared_labels metadata_shared['shared_cameras'] = sorted(shared_cameras) # Order: canonical label order (rgb first per design doc), cameras alphabetical ordered_labels = [l for l in CANONICAL_LABEL_ORDER if l in labels_available] ordered_labels += sorted(l for l in labels_available if l not in CANONICAL_LABEL_ORDER) metadata_shared['info']['tags'] = sorted(tags_available) metadata_shared['labels'] = ordered_labels metadata_shared['cameras'] = sorted(cameras_available) # Reorder keys to match design doc: info, labels, cameras, resolution, # framerate, language, num_frames, then label conventions, then rest. _KEY_ORDER = [ 'info', 'labels', 'cameras', 'resolution', 'framerate', 'language', 'num_frames', 'rgb', 'depth', 'lowdim', 'intrinsics', 'extrinsics', 'action', 'semantic', ] metadata_shared_ordered = {} for k in _KEY_ORDER: if k in metadata_shared: metadata_shared_ordered[k] = metadata_shared.pop(k) metadata_shared_ordered.update(metadata_shared) # Write shared metadata write_json(metadata_shared_path, metadata_shared_ordered) # Write sequence stats args.path = args.dst write_stats(args, metadata_shared_path, all_metadata_seqs, metadata_shared_ordered, err) return files, all_metadata ####################################################### def suffix_subset(path, args): if args.subset is not None: if '/' in args.subset: path += f'_{args.subset.replace("/","_")}' # Add subset value else: path += f'_{args.subset}' # Add subset value return path ####################################################### def run(converter, get_sequences_fn, parse_sequence_fn, process_sequence_fn, seqs=None): """Shared functionality for running the converter""" args = parse_args(converter) args.process_sequence_fn = process_sequence_fn args.parse_sequence_fn = parse_sequence_fn # Create split names split_name = suffix_subset('split_all', args) errors_name = suffix_subset('errors_all', args) stats_name = suffix_subset('stats_all', args) + '.txt' # Full split path for checking split_path = f'{args.dst}/{split_name}.json' tmp_split_path = f'{args.dst}/tmp_{split_name}.json' # Check if split already exists, and stop if not restarting if not args.restart: if os.path.exists(split_path): print(f'### {split_path} already exists, exiting...') return if os.path.exists(tmp_split_path): print(f'### {tmp_split_path} already exists, exiting...') return # Restarting, so warn if files already exist for path in [split_path, tmp_split_path]: if os.path.exists(path): print(f'### {path} already exists, restarting!!!') if not args.cleanup: rm_sync(path, args) rm_sync(path.replace('split_all', 'errors_all'), args) if args.subset is not None: rm_sync(f'{args.dst}/split_all.json', args) rm_sync(f'{args.dst}/errors_all.json', args) rm_sync(f'{args.dst}/stats_all.txt', args) if args.refresh: # Delete root json files (splits and metadata) json_files = glob(f'{args.dst}/*.json') txt_files = glob(f'{args.dst}/*.txt') for f in json_files + txt_files: remove_path(f) # Refresh s3 data as well if args.upload: # Delete root json files (splits and metadata) s3_path = args.dst.replace(args.local_path, args.s3_path) json_files_s3 = [f'{s3_path}/{f}' for f in list_s3(s3_path, suffix='.json')] txt_files_s3 = [f'{s3_path}/{f}' for f in list_s3(s3_path, suffix='.txt')] for f in json_files_s3 + txt_files_s3: aws_s3_rm(f, robust=True) # Write temporary file to get things started write_json(tmp_split_path, dict()) if args.upload: tmp_split_path_s3 = tmp_split_path.replace(args.local_path, args.s3_path) aws_s3_cp(tmp_split_path, tmp_split_path_s3, robust=True) # Get sequences to unify seqs = get_sequences_fn(args) if seqs is None else seqs # For debugging, only convert the first few if args.first is not None: seqs = seqs[:args.first] # Break conversion into subsets (e.g., 10/0 breaks in 10 subsets and process first) if args.subset == 'failed': failed = glob(f'{args.dst}/tmp/*.txt') failed = [f[len(f'{args.dst}/tmp')+1:-len('.txt')].replace('__','/') for f in failed] filtered_seqs = [] for seq in seqs: if any([seq.endswith(f) for f in failed]): filtered_seqs.append(seq) seqs = filtered_seqs else: seqs = get_seqs_subset(seqs, args.subset) print(f'#################################################') print(f'################### CONVERTING ##################') print(f'#################################################') print(f'### CONVERTER: {converter}') print(f'### NUM SEQUENCES: {len(seqs)}') if args.subset is not None: print(f'### SUBSET: {args.subset}') print(f'#################################################') print(f'### SRC: {args.src}') print(f'### DST: {args.dst}') print(f'#################################################') if len(seqs) == 0: print('##### THERE ARE NO SEQUENCES TO PROCESS!! Exiting...') return if not args.cleanup: ### Unify the data as requested args.log_suc = multiprocessing.Manager().dict() args.log_err = multiprocessing.Manager().dict() if args.num_procs == 0: # No processes, single thread process_sequences(0, seqs, args) else: # Multiple threads jobs = [] splits = get_splits(seqs, args) for i in range(args.num_procs): jobs.append(multiprocessing.Process( target=process_sequences, args=(i, seqs[splits[i]:splits[i+1]], args), )) for j in jobs: j.start() for j in jobs: j.join() else: ### Cleaning up, collect errors for logging and move on errors_path = f'{args.dst}/errors_all.json' if os.path.exists(errors_path): errors_path = [errors_path] else: errors_path = glob(f'{args.dst}/errors_all_*.json') args.log_err = dict() for error_path in errors_path: error_json = read_json(error_path) args.log_err.update(error_json['errors']) ### Cleaning up, collect successes for logging and move on splits_path = f'{args.dst}/split_all.json' if os.path.exists(splits_path): splits_path = [splits_path] else: splits_path = glob(f'{args.dst}/split_all_*.json') args.log_suc = dict() for split_path in splits_path: split_json = read_json(split_path) args.log_suc.update(split_json['sequences']) ### Log errors in the unified folder errors_path = f'{args.dst}/errors_all.json' if args.subset is not None: errors_path = errors_path.replace('.json', f'_{args.subset.replace("/","_")}.json') errors_dict = {'total': len(args.log_err), 'errors': {}} for key, val in args.log_err.items(): errors_dict['errors'][key] = val write_json(errors_path, errors_dict) ### Only consolidate metadata if processing without subsets if args.subset is None: ### Clean metadata for shared entries print('##### CLEANING METADATA ...') files, metadata_all = clean_metadata(args, scratch=True, suc=args.log_suc, err=args.log_err) else: files, metadata_all = None, None ### Create split_all (or part of it if subset is being used) print('##### CREATING SPLIT ...') from anydata.converters.misc.create_split import parse_create_split sys.argv = [sys.argv[0], args.dst, '--name', split_name, '--subset', args.subset] if args.official: sys.argv.append('--official') if args.storage == 'frames': sys.argv.append('--frames') if args.storage == 'videos': sys.argv.append('--videos') reason = parse_create_split(files=files, metadata_all=metadata_all, suc=args.log_suc, err=args.log_err) # Create final split if reason == 'failed': return remove_path(tmp_split_path) # Delete temporary split file ### Uploading results to s3 if args.upload or args.upload_and_delete: for file in [split_name, 'metadata_shared', stats_name, errors_name]: if not file.endswith('txt'): file += '.json' split = f'{args.dst}/{file}' if os.path.exists(split): s3_split = split.replace(args.local_path, args.s3_path) aws_s3_cp(split, s3_split, robust=True) s3_tmp_split = tmp_split_path.replace(args.local_path, args.s3_path) aws_s3_rm(s3_tmp_split, robust=True) # Upload tmp folder only if there are no subsets (could upload ongoing files as well) if args.subset is None: tmp_folder = f'{args.dst}/tmp' if os.path.exists(tmp_folder): tmp_folder_s3 = tmp_folder.replace(args.local_path, args.s3_path) aws_s3_sync(tmp_folder, tmp_folder_s3, robust=True) if args.cleanup: ### Deleting subset errors (better at the end, to avoid premature deletion) errors_path = glob(f'{args.dst}/errors_all_*.json') errors_path = [e for e in errors_path if not e.endswith(errors_name)] rm_sync(errors_path, args) ### Deleting subset splits (better at the end, to avoid premature deletion) splits_path = glob(f'{args.dst}/split_all_*.json') splits_path = [e for e in splits_path if not e.endswith(split_name)] rm_sync(splits_path, args) print(f'#################################################') print(f'################ CONVERTING DONE ################') print(f'#################################################') print(f'### CONVERTER: {converter}') print(f'### NUM SEQUENCES: {len(seqs)}') if args.subset is not None: print(f'### SUBSET: {args.subset}') if args.restart: print(f'### *** RESTART ***') if args.overwrite: print(f'### *** OVERWRITE ***') if args.cleanup: print(f'### *** CLEANUP ***') if args.refresh: print(f'### *** REFRESH ***') if args.webbed: print(f'### *** WEBBED ***') print(f'#################################################') print(f'### SRC: {args.src}') print(f'### DST: {args.dst}') print(f'#################################################') ####################################################### def remove_labels(data, labels): """Remove keys from list of labels if they don't match the rgb entry (e.g., different sequence length)""" if isinstance(data, dict): # If it's not a dict, it's correct for key1 in list(data.keys()): if isinstance(data[key1], dict) and 'rgb' in data[key1].keys(): # Check nested dict for key2 in list(data[key1].keys()): if data[key1][key2] != data[key1]['rgb']: if key2 in labels: labels.remove(key2) # Treat RGB value as the correct one elif 'rgb' in data.keys(): # Check dict if data[key1] != data['rgb']: if key1 in labels: labels.remove(key1) # Treat RGB value as the correct one return labels def remove_cameras(data, cameras): """Remove keys from list of cameras if they don't match the rgb entry (e.g., different sequence length)""" if isinstance(data, dict): # If it's not a dict, it's correct num = {key: val['rgb'] if isinstance(val, dict) else val for key, val in data.items() if key in cameras} if len(num) > 0: max_num = max(num.values()) invalid_cams = [key for key, val in num.items() if val < max_num] cameras = [c for c in cameras if c not in invalid_cams] return cameras def floor_resolution_even(resolution, rounding=4): """Floor resolution to be even (used for sequences encoded as mp4)""" if isinstance(resolution, dict): for key1, val1 in resolution.items(): if isinstance(val1, dict): for key2, val2 in val1.items(): resolution[key1][key2] = [v // rounding * rounding for v in val2] else: resolution[key1] = [v // rounding * rounding for v in val1] else: resolution = [v // rounding * rounding for v in resolution] return resolution def fill_metadata(*, args, info, labels, cameras, resolution, framerate, language, num_frames, rgb, depth, lowdim=None, intrinsics, extrinsics, action, semantic, specific): """Returns a unified metadata dictionary, for storage""" ### Sort tags alphabetically, labels by the canonical metadata order (CANONICAL_LABEL_ORDER above) info['tags'] = sorted(info['tags']) info['storage'] = args.storage # Add storage mode to info if language is not None and 'language' not in labels: # episode-level language labels.append('language') # may already be present as a per-frame lowdim label labels = [label for label in labels if label not in ['timestep','camera']] # Remove redundant labels labels = sorted(labels, key=lambda l: (CANONICAL_LABEL_ORDER.index(l) if l in CANONICAL_LABEL_ORDER else len(CANONICAL_LABEL_ORDER), l)) # Check and simplify certain entries resolution = check_dict(resolution, simplify_first=True) num_frames = check_dict(num_frames, simplify_first=True) framerate = check_dict(framerate, simplify_first=True) # If entries don't agree across labels, remove those from the labels list labels = remove_labels(num_frames, labels) # If entries don't agree across cameras, remove those cameras cameras = remove_cameras(num_frames, cameras) # Include default extensions if lowdim is None: lowdim = dict(extension='npz') # If storing sequences, switch rgb, depth, and lowdim extensions accordingly if args.storage == 'videos': resolution = floor_resolution_even(resolution) if rgb is not None: rgb['extension'] = 'mp4' if depth is not None: depth['extension'] = 'zarr' if lowdim is not None: lowdim['extension'] = 'npz' ### Return metadata dictionary (design doc key order) metadata = dict( info=info, labels=labels, cameras=sorted(cameras), resolution=resolution, framerate=framerate, language=check_dict(language) if 'language' in labels else None, num_frames=num_frames, ### LABEL CONVENTIONS (top-level per design doc; conventions wrapper is dataloader-only) rgb=check_dict(rgb) if 'rgb' in labels else None, depth=check_dict(depth) if 'depth' in labels else None, lowdim=check_dict(lowdim) if 'lowdim' in labels else None, intrinsics=check_dict(intrinsics) if 'intrinsics' in labels else None, extrinsics=check_dict(extrinsics) if 'extrinsics' in labels else None, action=check_dict(action) if 'action' in labels else None, semantic=check_dict(semantic) if 'semantic' in labels else None, ### DATASET SPECIFIC specific=specific, ) return remove_none(metadata) ####################################################### def prepare_pattern(pattern): """Prepare a pattern dict to check for matches in a nested folder""" # Nothing to do if pattern is None: return pattern # Checks for wildcards starts = pattern.endswith('*') ends = pattern.startswith('*') middle = '*' in pattern and not starts and not ends equal = '*' not in pattern # Create dictioanry with 'starts', 'ends', and 'equal' keys for comparison pattern_dict = dict() if starts: pattern_dict['starts'] = pattern[:-1] if ends: pattern_dict['ends'] = pattern[1:] if middle: split = pattern.split('*') pattern_dict['starts'] = split[0] pattern_dict['ends'] = split[1] if equal: pattern_dict['equal'] = pattern # Return pattern dictionary return pattern_dict def is_crawl_match(f, pattern): """Checks if a filename is a match for a given pattern""" bf = os.path.basename(f) match_starts = 'starts' not in pattern or ('starts' in pattern and bf.startswith(pattern['starts'])) match_ends = 'ends' not in pattern or ('ends' in pattern and bf.endswith(pattern['ends'])) match_equals = 'equal' not in pattern or (bf == pattern['equal']) return match_starts and match_ends and match_equals def crawl(path, pattern, avoid=None, first=True): """Crawl over a nested folder searching for a pattern while avoiding others (faster alternative to glob)""" if first: # Prepare patterns for first loop pattern = prepare_pattern(pattern) avoid = prepare_pattern(avoid) # Folders in the current level files = glob(f'{path}/*') matches = [] found = False for f in files: # If it's a match, store if is_crawl_match(f, pattern): matches.append(f) found = True # If there were no matches, keep going deeper if not found: for f in files: if avoid is None or not is_crawl_match(f, avoid): # Check if a folder should be avoided matches.extend(crawl(f, pattern=pattern, avoid=avoid, first=False)) return matches def create_metadata_shared(dst_dir, pattern='**/metadata.json', dataset_name=None, label_order=None, camera_order=None): """Build metadata_shared.json from per-episode metadata files. Scans all per-episode metadata.json files under dst_dir, computes: - cameras: union over episodes (design doc: "global = union over episodes") - labels: union over episodes - tags: intersection over episodes - resolution/framerate/num_frames: shared value or 'variable' - label convention keys (rgb, depth, intrinsics, ...): shared value or 'variable' Writes result in design-doc key order. Parameters ---------- dst_dir : str Root of the unified dataset. pattern : str Glob pattern for per-episode metadata.json files. dataset_name : str, optional Override for info.name (default: read from first episode). label_order : list, optional Canonical ordering for labels (extras appended alphabetically). camera_order : list, optional Canonical ordering for cameras (extras appended alphabetically). """ import json from anydata.utils.write import write_json from rich.console import Console console = Console() console.rule(f'[bold]Creating metadata_shared.json for {dst_dir}[/bold]') meta_paths = sorted(glob(os.path.join(dst_dir, pattern), recursive=True)) n_episodes = len(meta_paths) console.print(f' Found {n_episodes} episodes') if n_episodes == 0: console.print('[red]No episodes found, skipping metadata_shared.json[/red]') return None all_labels = set() all_cameras = set() all_tags = None all_resolutions = set() all_framerates = set() all_num_frames = [] # Label convention keys: collect unique JSON-serialized values per key _LABEL_KEYS = ('rgb', 'depth', 'intrinsics', 'extrinsics', 'action', 'semantic', 'language') label_meta_vals = {k: set() for k in _LABEL_KEYS} first_info = None for mp in meta_paths: with open(mp) as f: meta = json.load(f) if first_info is None: first_info = meta.get('info', {}) all_labels.update(meta.get('labels', [])) all_cameras.update(meta.get('cameras', [])) ep_tags = set(meta.get('info', {}).get('tags', [])) all_tags = ep_tags if all_tags is None else (all_tags & ep_tags) res = meta.get('resolution') if isinstance(res, dict): all_resolutions.add(json.dumps(res, sort_keys=True)) elif isinstance(res, list): all_resolutions.add(tuple(res)) elif res is not None: all_resolutions.add(res) fr = meta.get('framerate') if fr is not None: all_framerates.add(fr) nf = meta.get('num_frames') if isinstance(nf, (int, float)): all_num_frames.append(nf) for lk in _LABEL_KEYS: if lk in meta and isinstance(meta[lk], dict): label_meta_vals[lk].add(json.dumps(meta[lk], sort_keys=True)) # Resolve scalars: single unique value or 'variable' def _resolve_set(s, as_list=False): if len(s) == 1: val = next(iter(s)) if as_list and isinstance(val, tuple): return list(val) if isinstance(val, str) and val.startswith('{'): return json.loads(val) return val elif len(s) > 1: return 'variable' return None resolved_resolution = _resolve_set(all_resolutions, as_list=True) resolved_framerate = _resolve_set(all_framerates) resolved_num_frames = 'variable' if len(set(all_num_frames)) > 1 else ( all_num_frames[0] if all_num_frames else None) # Resolve label convention keys resolved_label_meta = {} for lk, vals in label_meta_vals.items(): if len(vals) == 1: resolved_label_meta[lk] = json.loads(next(iter(vals))) elif len(vals) > 1: resolved_label_meta[lk] = 'variable' # Order labels and cameras by canonical order, then alphabetical for extras. # Default label order follows design doc (highdim first, then lowdim-derived). _DEFAULT_LABEL_ORDER = CANONICAL_LABEL_ORDER def _ordered(items, order): if order is None: order = [] result = [x for x in order if x in items] result += sorted(x for x in items if x not in order) return result if label_order is None: label_order = _DEFAULT_LABEL_ORDER name = dataset_name or first_info.get('name', os.path.basename(dst_dir)) # Tags: intersection gives only tags present in ALL episodes tags = sorted(all_tags) if all_tags else [] shared = dict( info=dict(name=name, tags=tags), labels=_ordered(all_labels, label_order), cameras=_ordered(all_cameras, camera_order), ) if resolved_resolution is not None: shared['resolution'] = resolved_resolution if resolved_framerate is not None: shared['framerate'] = resolved_framerate # language at top level (shared across dataset) -- use resolved or empty dict if 'language' in resolved_label_meta: shared['language'] = resolved_label_meta.pop('language') elif any('language' in meta.get('labels', []) for mp in meta_paths[:1] for meta in [json.load(open(mp))]): shared['language'] = {} if resolved_num_frames is not None: shared['num_frames'] = resolved_num_frames # Label conventions (top-level per design doc) for lk in _LABEL_KEYS: if lk in resolved_label_meta: shared[lk] = resolved_label_meta[lk] out_path = os.path.join(dst_dir, 'metadata_shared.json') write_json(out_path, shared) console.print(f' [green]{out_path} written[/green]') console.print(f' labels: {shared["labels"]}') console.print(f' cameras: {shared["cameras"]}') return shared def create_coverage(dst_dir, pattern='cfg*/task_*/metadata.json'): """Create coverage.json with label/camera/tag frequency stats for a unified dataset. Works for any converter that writes per-episode metadata.json with 'labels', 'cameras', 'info.tags', and 'num_frames' keys. """ from collections import Counter from anydata.utils.write import write_json from rich.console import Console import json console = Console() console.rule(f'[bold]Creating coverage.json for {dst_dir}[/bold]') meta_paths = sorted(glob(os.path.join(dst_dir, pattern), recursive=True)) n_episodes = len(meta_paths) console.print(f' Found {n_episodes} episodes') if n_episodes == 0: console.print('[red]No episodes found, skipping[/red]') return None label_counts = Counter() camera_counts = Counter() tag_counts = Counter() frame_counts = [] cameras_per_ep = [] labels_per_ep = [] for mp in meta_paths: with open(mp) as f: meta = json.load(f) label_counts.update(meta.get('labels', [])) camera_counts.update(meta.get('cameras', [])) tag_counts.update(meta.get('info', {}).get('tags', [])) cameras_per_ep.append(len(meta.get('cameras', []))) labels_per_ep.append(len(meta.get('labels', []))) nf = meta.get('num_frames', 0) while isinstance(nf, dict) and nf: nf = next(iter(nf.values())) if isinstance(nf, (int, float)) and nf > 0: frame_counts.append(int(nf)) def _cov_dict(counter): return {k: {'count': v, 'pct': round(100 * v / n_episodes, 1)} for k, v in sorted(counter.items(), key=lambda x: (-x[1], x[0]))} coverage = dict( n_episodes=n_episodes, label_coverage=_cov_dict(label_counts), camera_coverage=_cov_dict(camera_counts), tag_coverage=_cov_dict(tag_counts), frames=dict( mean=round(float(np.mean(frame_counts)), 2) if frame_counts else 0, std=round(float(np.std(frame_counts)), 2) if frame_counts else 0, min=int(np.min(frame_counts)) if frame_counts else 0, max=int(np.max(frame_counts)) if frame_counts else 0, total=int(np.sum(frame_counts)) if frame_counts else 0, ), cameras_per_episode=dict( mean=round(float(np.mean(cameras_per_ep)), 2), std=round(float(np.std(cameras_per_ep)), 2), ), labels_per_episode=dict( mean=round(float(np.mean(labels_per_ep)), 2), std=round(float(np.std(labels_per_ep)), 2), ), ) out_path = os.path.join(dst_dir, 'coverage.json') write_json(out_path, coverage) console.print(f' [green]{out_path} written[/green]') for section in ['label_coverage', 'camera_coverage', 'tag_coverage']: console.print(f' {section}:') for k, v in coverage[section].items(): console.print(f' {k}: {v["count"]}/{n_episodes} ({v["pct"]}%)') return coverage def get_num_frames(val, cams): if not isinstance(val, dict): return val elif 'rgb' in val.keys(): return val['rgb'] else: return get_num_frames(val[cams[0]], cams[1:]) def get_shared_info(metadata): first = list(metadata.values())[0] shared_tags = first['info']['tags'] shared_labels = first['labels'] shared_cameras = first['cameras'] for key, val in metadata.items(): shared_tags = [s for s in shared_tags if s in val['info']['tags']] shared_labels = [s for s in shared_labels if s in val['labels']] shared_cameras = [s for s in shared_cameras if s in val['cameras']] return shared_tags, shared_labels, shared_cameras def write_stats(args, base, metadata, shared, errors=None, name='split_all', removed=None): name = name.replace('split_', 'stats_') stats_path = base.replace('metadata_shared.json', name) stats_name = suffix_subset(stats_path, args) + '.txt' shared_tags, shared_labels, shared_cameras = get_shared_info(metadata) with open(stats_name, 'w') as f: seqs, tags, labels, cameras, nums = [], [], [], [], [] num_seqs = len(metadata) num_scameras = len(shared_cameras) num_acameras = len(shared['cameras']) num_errors = 0 if errors is None else len(errors) num_samples = 0 num_frames = 0 num_shours = 0 num_fhours = 0 atags = '[]' if 'tags' not in shared['info'] else str(shared['info']['tags']).replace(' ', '').replace("'","") alabels = str(shared['labels']).replace(' ', '').replace("'","") acameras = str(shared['cameras']).replace(' ', '').replace("'","") tags_perc = {key: 0 for key in shared['info']['tags']} labels_perc = {key: 0 for key in shared['labels']} cameras_perc = {key: 0 for key in shared['cameras']} resolution = {key: {} for key in shared['cameras']} dist_cameras = {} keys = sorted(list(metadata.keys())) for key in keys: val = metadata[key] seqs.append(str(key) if len(key) < 96 else key[-96:]) tags.append(str(val['info']['tags'])) labels.append(str(val['labels'])) cameras.append(str(val['cameras'])) nums.append(str(get_num_frames(val['num_frames'], val['cameras']))) if isinstance(val['resolution'], dict): for k, v in val['resolution'].items(): v = str(v).replace(' ', '') if v not in resolution[k]: resolution[k][v] = 0 resolution[k][v] += 1 else: for k in val['cameras']: v = str(val['resolution']).replace(' ', '') if v not in resolution[k]: resolution[k][v] = 0 resolution[k][v] += 1 ncams = len(val['cameras']) if ncams not in dist_cameras: dist_cameras[ncams] = 0 dist_cameras[ncams] += 1 length = int(nums[-1]) num_samples += length num_frames += length * len(val['cameras']) fph = val['framerate'] * 3600 if fph > 0: num_shours += length / fph num_fhours += length * len(val['cameras']) / fph for t in val['info']['tags']: tags_perc[t] += 1 for l in val['labels']: labels_perc[l] += 1 for c in val['cameras']: cameras_perc[c] += 1 if len(dist_cameras) > 0: dist_cameras = {key: dist_cameras[key] for key in sorted(dist_cameras.keys())} tags = [s.replace(' ', '').replace("'","") for s in tags] labels = [s.replace(' ', '').replace("'","") for s in labels] cameras = [s.replace(' ', '').replace("'","") for s in cameras] int_nums = [int(n) for n in nums] max_seqs = max(max(len(s) for s in seqs), len('Sequence')) max_tags = max(max(len(s) for s in tags), len('Tags')) max_labels = max(max(len(s) for s in labels), len('Labels')) max_cameras = max(max(len(s) for s in cameras), len('Cameras')) max_nums = max(max(len(s) for s in nums), len('Frames')) max_seqs = max(max_seqs, len('AVAILABLE')) max_tags = max(max_tags, len(atags)) max_labels = max(max_labels, len(alabels)) max_cameras = max(max_cameras, len(acameras)) stags = str(shared_tags).replace(' ', '').replace("'","") slabels = str(shared_labels).replace(' ', '').replace("'","") scameras = str(shared_cameras).replace(' ', '').replace("'","") max_seqs = max(max_seqs, len('SHARED')) max_tags = max(max_tags, len(stags)) max_labels = max(max_labels, len(slabels)) max_cameras = max(max_cameras, len(scameras)) if errors is not None and len(errors) > 0: max_errs = max(len(e) for e in errors) max_seqs = max(max_seqs, max_errs) dataset = shared['info']['name'] path = args.path.replace(args.local_path, args.s3_path) num_shours_str = str('{:,.2f}'.format(num_shours)) num_fhours_str = str('{:,.2f}'.format(num_fhours)) total = max_seqs + max_tags + max_labels + max_cameras + max_nums + 19 f.write('-' * total + '\n') f.write(f' | ### {dataset} : {path} \n') f.write('-' * total + '\n') if name == 'stats_all': # Error logging is only for stats_all sequence_str = f'{num_seqs:,}/{num_seqs + num_errors:,} ({num_errors:,} errors)' if 'framerate' in shared.keys(): sequence_str += f' ({shared["framerate"]} fps)' else: sequence_str = f'{num_seqs:,}/{num_seqs + removed:,} ({removed:,} removed)' samples_str = f'{num_samples:,}' if num_shours > 0: samples_str += f' ({num_shours_str} hours)' frames_str = f'{num_frames:,}' if num_fhours > 0: frames_str += f' ({num_fhours_str} hours)' f.write(f' | ### SEQUENCES: {sequence_str} \n') f.write(f' | ### SAMPLES: {samples_str} \n') f.write(f' | ### FRAMES: {frames_str} \n') f.write(f' | ### CAMERAS: {num_scameras}/{num_acameras} \n') f.write('-' * total + '\n') f.write(f' | ### NUM_FRAMES DENSITY: \n') f.write(f' | ###### Min: {np.min(int_nums):,} \n') f.write(f' | ###### Max: {np.max(int_nums):,} \n') f.write(f' | ###### Mean: {int(np.round(np.mean(int_nums))):,} \n') f.write(f' | ###### Std: {int(np.round(np.std(int_nums))):,} \n') f.write(f' | ### NUM_CAMERAS DENSITY: \n') for key, val in dist_cameras.items(): perc = float(np.round(10000 * val / num_seqs)) / 100 f.write(f' | ###### {key}: {val}/{num_seqs} ({perc}%) \n') f.write('-' * total + '\n') f.write(f' | ### TAG DENSITY: \n') for key, val in tags_perc.items(): perc = float(np.round(10000 * val / num_seqs)) / 100 f.write(f' | ###### {key}: {val:,}/{num_seqs:,} ({perc}%) \n') f.write(f' | ### LABEL DENSITY: \n') for key, val in labels_perc.items(): perc = float(np.round(10000 * val / num_seqs)) / 100 f.write(f' | ###### {key}: {val:,}/{num_seqs:,} ({perc}%) \n') f.write(f' | ### CAMERA DENSITY: \n') for key, val in cameras_perc.items(): perc = float(np.round(10000 * val / num_seqs)) / 100 f.write(f' | ###### {key}: {val:,}/{num_seqs:,} ({perc}%) \n') f.write('-' * total + '\n') f.write(f' | ### CAMERA RESOLUTION: \n') for key, val in resolution.items(): f.write(f' | ###### {key}: \n') tot = sum(val.values()) for c, v in val.items(): perc = float(np.round(10000 * v / tot)) / 100 f.write(f' | ######### {c}: {v:,}/{tot:,} ({perc}%) \n') f.write('-' * total + '\n') s = '{m: ^{width}}'.format(m='Sequence', width=max_seqs) t = '{m: ^{width}}'.format(m='Tags', width=max_tags) l = '{m: ^{width}}'.format(m='Labels', width=max_labels) c = '{m: ^{width}}'.format(m='Cameras', width=max_cameras) n = '{m: ^{width}}'.format(m='Frames', width=max_nums) data = f' | {s} | {t} | {l} | {c} | {n} |' f.write(f'{data}\n') f.write('-' * total + '\n') s = '{m: ^{width}}'.format(m='AVAILABLE', width=max_seqs) t = '{m: ^{width}}'.format(m=atags, width=max_tags) l = '{m: ^{width}}'.format(m=alabels, width=max_labels) c = '{m: ^{width}}'.format(m=acameras, width=max_cameras) n = '{m: ^{width}}'.format(m='-', width=max_nums) data = f' | {s} | {t} | {l} | {c} | {n} |' f.write(f'{data}\n') s = '{m: ^{width}}'.format(m='SHARED', width=max_seqs) t = '{m: ^{width}}'.format(m=stags, width=max_tags) l = '{m: ^{width}}'.format(m=slabels, width=max_labels) c = '{m: ^{width}}'.format(m=scameras, width=max_cameras) n = '{m: ^{width}}'.format(m='-', width=max_nums) data = f' | {s} | {t} | {l} | {c} | {n} |' f.write(f'{data}\n') f.write('-' * total + '\n') for i in range(len(metadata)): s = '{m: <{width}}'.format(m=seqs[i], width=max_seqs) t = '{m: ^{width}}'.format(m=tags[i], width=max_tags) l = '{m: ^{width}}'.format(m=labels[i], width=max_labels) c = '{m: ^{width}}'.format(m=cameras[i], width=max_cameras) n = '{m: >{width}}'.format(m=nums[i], width=max_nums) data = f' | {s} | {t} | {l} | {c} | {n} |' f.write(f'{data}\n') f.write('-' * total + '\n') if errors is not None and len(errors) > 0: f.write('-' * total + '\n') for seq, reason in errors.items(): if len(seq) < 96: seq = seq[-96:] s = '{m: <{width}}'.format(m=seq, width=max_seqs) t = '{m: <}'.format(m=reason) data = f' | {s} | {t}' f.write(f'{data}\n') f.write('-' * total + '\n') return stats_name