#!/usr/bin/env python3 """ Quick validation script to check sanity check contents of processed format (ex: frame counts across rgb and lowdim folders) This version has minimal dependencies, and has been tested on AssemblyHands_V2 dataset. You can change `dataset_path` and `camera_names` to point to your dataset. Usage: python3 quick_validate.py """ import os import re # Dataset paths dataset_path = "/workspace/cv_datasets/processed/AssemblyHands_V2/" # List of camera folder names camera_names = ['camera_0', 'camera_1', 'camera_2', 'camera_3', 'camera_4', 'camera_5', 'camera_6', 'camera_7'] def extract_frame_number(filename): """Extract frame number from filename like frame_000123.jpg""" match = re.match(r'frame_(\d+)\.(jpg|npz)', filename) if match: return int(match.group(1)) return None def validate_episode(episode_path): """Validate a single episode for frame consistency""" episode_name = os.path.basename(episode_path) errors = [] warnings = [] rgb_path = os.path.join(episode_path, 'rgb') lowdim_path = os.path.join(episode_path, 'lowdim') # Check basic structure if not os.path.exists(rgb_path): errors.append(f"Missing rgb directory in {episode_name}") return errors, warnings, 0 if not os.path.exists(lowdim_path): errors.append(f"Missing lowdim directory in {episode_name}") return errors, warnings, 0 camera_data = {} # Process each camera for camera in camera_names: camera_rgb_path = os.path.join(rgb_path, camera) camera_lowdim_path = os.path.join(lowdim_path, camera) if not os.path.exists(camera_rgb_path): errors.append(f"Missing {episode_name}/rgb/{camera}") continue if not os.path.exists(camera_lowdim_path): errors.append(f"Missing {episode_name}/lowdim/{camera}") continue # Get frame files rgb_files = [f for f in os.listdir(camera_rgb_path) if f.endswith('.jpg')] lowdim_files = [f for f in os.listdir(camera_lowdim_path) if f.endswith('.npz')] # Sort files rgb_files.sort() lowdim_files.sort() # Extract frame numbers rgb_frame_nums = [] lowdim_frame_nums = [] for f in rgb_files: frame_num = extract_frame_number(f) if frame_num is not None: rgb_frame_nums.append(frame_num) else: errors.append(f"Invalid RGB filename: {episode_name}/{camera}/{f}") for f in lowdim_files: frame_num = extract_frame_number(f) if frame_num is not None: lowdim_frame_nums.append(frame_num) else: errors.append(f"Invalid lowdim filename: {episode_name}/{camera}/{f}") camera_data[camera] = { 'rgb_count': len(rgb_files), 'lowdim_count': len(lowdim_files), 'rgb_frames': set(rgb_frame_nums), 'lowdim_frames': set(lowdim_frame_nums) } # Check RGB vs lowdim consistency for this camera if len(rgb_files) != len(lowdim_files): errors.append(f"{episode_name}/{camera}: Frame count mismatch - RGB: {len(rgb_files)}, lowdim: {len(lowdim_files)}") if set(rgb_frame_nums) != set(lowdim_frame_nums): missing_rgb = set(lowdim_frame_nums) - set(rgb_frame_nums) missing_lowdim = set(rgb_frame_nums) - set(lowdim_frame_nums) if missing_rgb: errors.append(f"{episode_name}/{camera}: Missing RGB frames: {sorted(list(missing_rgb))}") if missing_lowdim: errors.append(f"{episode_name}/{camera}: Missing lowdim frames: {sorted(list(missing_lowdim))}") # Check consistency across cameras if camera_data: reference_camera = camera_names[0] if reference_camera in camera_data: ref_count = camera_data[reference_camera]['rgb_count'] ref_frames = camera_data[reference_camera]['rgb_frames'] for camera in camera_names[1:]: if camera in camera_data: if camera_data[camera]['rgb_count'] != ref_count: errors.append(f"{episode_name}: Frame count mismatch across cameras - {reference_camera}: {ref_count}, {camera}: {camera_data[camera]['rgb_count']}") if camera_data[camera]['rgb_frames'] != ref_frames: errors.append(f"{episode_name}: Frame indices mismatch between {reference_camera} and {camera}") return errors, warnings, ref_count return errors, warnings, 0 def main(): print(f"Validating AssemblyHands_V2 dataset at: {dataset_path}") if not os.path.exists(dataset_path): print(f"ERROR: Dataset path does not exist: {dataset_path}") return 1 # Get all episodes episodes = [d for d in os.listdir(dataset_path) if os.path.isdir(os.path.join(dataset_path, d)) and d.startswith('episode_')] episodes.sort() if not episodes: print("ERROR: No episodes found in dataset") return 1 print(f"Found {len(episodes)} episodes to validate") total_errors = [] total_warnings = [] episode_stats = {} # Validate each episode for i, episode in enumerate(episodes): print(f"Validating {i+1}/{len(episodes)}: {episode}") episode_path = os.path.join(dataset_path, episode) errors, warnings, frame_count = validate_episode(episode_path) print (errors, warnings, frame_count) total_errors.extend(errors) total_warnings.extend(warnings) episode_stats[episode] = frame_count if errors: print(f"{len(errors)} errors found") else: print(f"No errors ({frame_count} frames)") # Print summary print("\n" + "="*60) print("VALIDATION SUMMARY") print("="*60) print(f"Total episodes: {len(episodes)}") if episode_stats: frame_counts = list(episode_stats.values()) avg_frames = sum(frame_counts) / len(frame_counts) if frame_counts else 0 min_frames = min(frame_counts) if frame_counts else 0 max_frames = max(frame_counts) if frame_counts else 0 print(f"Average frames per episode: {avg_frames:.1f}") print(f"Frame count range: {min_frames} - {max_frames}") print(f"Total errors: {len(total_errors)}") print(f"Total warnings: {len(total_warnings)}") if total_errors: print("\nFirst 20 ERRORS:") for error in total_errors[:20]: print(f" - {error}") if len(total_errors) > 20: print(f" ... and {len(total_errors) - 20} more errors") if total_warnings: print("\nWARNINGS:") for warning in total_warnings[:10]: print(f" - {warning}") if len(total_warnings) > 10: print(f" ... and {len(total_warnings) - 10} more warnings") if len(total_errors) == 0: print("\nVALIDATION PASSED - No errors detected!") return 0 else: print(f"\nVALIDATION FAILED - {len(total_errors)} errors detected!") return 1 if __name__ == "__main__": exit(main())