RSK World - Video Classification Dataset - Project Files Browser | RSK World

organize_dataset.py transcript_samples.json process_videos.py

scripts/organize_dataset.py

#!/usr/bin/env python3
"""
Dataset Organization Script for Video Classification Dataset
Author: Molla Samser
Designer & Tester: Rima Khatun
Website: https://rskworld.in
Email: help@rskworld.in, support@rskworld.in
Phone: +91 93305 39277
Organization: RSK World

This script organizes video files into train/test/validation splits.
"""

import os
import argparse
import yaml
import shutil
from pathlib import Path
import random


def load_config(config_path='config.yaml'):
    """Load configuration from YAML file."""
    if not os.path.exists(config_path):
        raise FileNotFoundError(f"Configuration file not found: {config_path}")
    with open(config_path, 'r') as f:
        config = yaml.safe_load(f)
        if config is None:
            raise ValueError(f"Configuration file is empty or invalid: {config_path}")
        return config


def get_video_files(directory, video_formats):
    """
    Get all video files organized by category.
    
    Args:
        directory: Root directory containing category folders
        video_formats: List of video format extensions
    """
    videos_by_category = {}
    
    for category_dir in Path(directory).iterdir():
        if not category_dir.is_dir():
            continue
        
        category = category_dir.name
        videos = []
        
        for format in video_formats:
            videos.extend(category_dir.rglob(f"*.{format}"))
            videos.extend(category_dir.rglob(f"*.{format.upper()}"))
        
        if videos:
            videos_by_category[category] = [str(v) for v in videos]
    
    return videos_by_category


def split_dataset(videos_by_category, train_ratio, test_ratio, validation_ratio, random_seed=42):
    """
    Split dataset into train/test/validation sets.
    
    Args:
        videos_by_category: Dictionary of category -> video files
        train_ratio: Ratio for training set
        test_ratio: Ratio for test set
        validation_ratio: Ratio for validation set
        random_seed: Random seed for reproducibility
    """
    random.seed(random_seed)
    
    train_files = {}
    test_files = {}
    validation_files = {}
    
    for category, videos in videos_by_category.items():
        # Shuffle videos
        random.shuffle(videos)
        
        # Calculate split sizes
        total = len(videos)
        train_size = int(total * train_ratio)
        test_size = int(total * test_ratio)
        
        # Split videos
        train = videos[:train_size]
        remaining = videos[train_size:]
        
        test = remaining[:test_size]
        validation = remaining[test_size:]
        
        train_files[category] = train
        test_files[category] = test
        validation_files[category] = validation
        
        print(f"Category '{category}': Train={len(train)}, Test={len(test)}, Validation={len(validation)}")
    
    return train_files, test_files, validation_files


def copy_files(file_dict, output_dir):
    """
    Copy files to output directory maintaining category structure.
    
    Args:
        file_dict: Dictionary of category -> file paths
        output_dir: Output directory
    """
    for category, files in file_dict.items():
        category_dir = os.path.join(output_dir, category)
        os.makedirs(category_dir, exist_ok=True)
        
        for file_path in files:
            if not os.path.exists(file_path):
                print(f"Warning: Source file not found: {file_path}")
                continue
            
            filename = os.path.basename(file_path)
            dest_path = os.path.join(category_dir, filename)
            try:
                shutil.copy2(file_path, dest_path)
            except Exception as e:
                print(f"Error copying {file_path} to {dest_path}: {e}")


def organize_dataset(input_dir, output_dir, config):
    """
    Organize dataset into train/test/validation splits.
    
    Args:
        input_dir: Input directory containing videos organized by category
        output_dir: Output directory for organized dataset
        config: Configuration dictionary
    """
    video_formats = config['video']['formats']
    splits = config['dataset']['splits']
    
    # Get all video files by category
    print("Scanning for video files...")
    videos_by_category = get_video_files(input_dir, video_formats)
    
    if not videos_by_category:
        print("No video files found!")
        return
    
    print(f"Found {len(videos_by_category)} categories")
    
    # Split dataset
    print("\nSplitting dataset...")
    train_files, test_files, validation_files = split_dataset(
        videos_by_category,
        splits['train_ratio'],
        splits['test_ratio'],
        splits['validation_ratio']
    )
    
    # Create output directories
    train_dir = os.path.join(output_dir, 'train')
    test_dir = os.path.join(output_dir, 'test')
    validation_dir = os.path.join(output_dir, 'validation')
    
    # Copy files to respective directories
    print("\nCopying files to train directory...")
    copy_files(train_files, train_dir)
    
    print("Copying files to test directory...")
    copy_files(test_files, test_dir)
    
    print("Copying files to validation directory...")
    copy_files(validation_files, validation_dir)
    
    print("\nDataset organization completed!")
    print(f"Train: {sum(len(files) for files in train_files.values())} videos")
    print(f"Test: {sum(len(files) for files in test_files.values())} videos")
    print(f"Validation: {sum(len(files) for files in validation_files.values())} videos")


def main():
    parser = argparse.ArgumentParser(description='Organize dataset into train/test/validation splits')
    parser.add_argument('--input', type=str, required=True,
                       help='Input directory containing videos organized by category')
    parser.add_argument('--output', type=str, default='data',
                       help='Output directory for organized dataset')
    parser.add_argument('--config', type=str, default='config.yaml',
                       help='Path to configuration file')
    parser.add_argument('--train-ratio', type=float, default=None,
                       help='Training set ratio (overrides config)')
    parser.add_argument('--test-ratio', type=float, default=None,
                       help='Test set ratio (overrides config)')
    parser.add_argument('--validation-ratio', type=float, default=None,
                       help='Validation set ratio (overrides config)')
    
    args = parser.parse_args()
    
    # Load configuration
    config = load_config(args.config)
    
    # Override ratios if provided
    if args.train_ratio is not None:
        config['dataset']['splits']['train_ratio'] = args.train_ratio
    if args.test_ratio is not None:
        config['dataset']['splits']['test_ratio'] = args.test_ratio
    if args.validation_ratio is not None:
        config['dataset']['splits']['validation_ratio'] = args.validation_ratio
    
    # Organize dataset
    organize_dataset(args.input, args.output, config)


if __name__ == '__main__':
    main()

214 lines•7.2 KB

python

scripts/process_videos.py

Raw Download

#!/usr/bin/env python3
"""
Video Processing Script for Video Classification Dataset
Author: Molla Samser
Designer & Tester: Rima Khatun
Website: https://rskworld.in
Email: help@rskworld.in, support@rskworld.in
Phone: +91 93305 39277
Organization: RSK World

This script processes video files: resizing, format conversion, and validation.
"""

import cv2
import os
import argparse
import yaml
from pathlib import Path
from tqdm import tqdm
import subprocess


def load_config(config_path='config.yaml'):
    """Load configuration from YAML file."""
    if not os.path.exists(config_path):
        raise FileNotFoundError(f"Configuration file not found: {config_path}")
    with open(config_path, 'r') as f:
        config = yaml.safe_load(f)
        if config is None:
            raise ValueError(f"Configuration file is empty or invalid: {config_path}")
        return config


def get_video_info(video_path):
    """Get video information."""
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        return None
    
    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    
    # Avoid division by zero
    duration = frame_count / fps if fps > 0 else 0.0
    
    info = {
        'fps': fps,
        'width': int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
        'height': int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
        'frame_count': frame_count,
        'duration': duration
    }
    cap.release()
    return info


def resize_video(input_path, output_path, target_size=(224, 224)):
    """
    Resize video to target dimensions.
    
    Args:
        input_path: Input video path
        output_path: Output video path
        target_size: Target (width, height)
    """
    cap = cv2.VideoCapture(input_path)
    if not cap.isOpened():
        return False
    
    fps = cap.get(cv2.CAP_PROP_FPS)
    # Use default FPS if invalid
    if fps <= 0:
        fps = 30.0
    
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    out = cv2.VideoWriter(output_path, fourcc, fps, target_size)
    
    if not out.isOpened():
        cap.release()
        return False
    
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        
        resized_frame = cv2.resize(frame, target_size)
        out.write(resized_frame)
    
    cap.release()
    out.release()
    return True


def convert_video_format(input_path, output_path, output_format='mp4'):
    """
    Convert video format using FFmpeg.
    
    Args:
        input_path: Input video path
        output_path: Output video path
        output_format: Output format (mp4, mov, etc.)
    """
    try:
        cmd = [
            'ffmpeg', '-i', str(input_path),
            '-c:v', 'libx264',
            '-c:a', 'aac',
            '-y',  # Overwrite output file
            str(output_path)
        ]
        subprocess.run(cmd, check=True, capture_output=True)
        return True
    except (subprocess.CalledProcessError, FileNotFoundError):
        print(f"FFmpeg not found or error converting {input_path}")
        return False


def validate_video(video_path, config):
    """
    Validate video file meets requirements.
    
    Args:
        video_path: Path to video file
        config: Configuration dictionary
    """
    info = get_video_info(video_path)
    if info is None:
        return False, "Could not open video"
    
    max_duration = config['video']['processing'].get('max_duration', None)
    if max_duration and info['duration'] > max_duration:
        return False, f"Video duration {info['duration']:.2f}s exceeds max {max_duration}s"
    
    return True, "Valid"


def process_video(input_path, output_path, config, resize=True, convert=False):
    """
    Process a single video file.
    
    Args:
        input_path: Input video path
        output_path: Output video path
        config: Configuration dictionary
        resize: Whether to resize video
        convert: Whether to convert format
    """
    # Validate video
    is_valid, message = validate_video(input_path, config)
    if not is_valid:
        print(f"Warning: {input_path} - {message}")
        return False
    
    # Create output directory
    output_dir = os.path.dirname(output_path)
    if output_dir:  # Only create directory if path contains a directory
        os.makedirs(output_dir, exist_ok=True)
    
    # Get target resolution
    target_size = tuple(config['video']['processing']['resolution'])
    
    # Process video
    if resize:
        return resize_video(input_path, output_path, target_size)
    elif convert:
        return convert_video_format(input_path, output_path)
    else:
        # Just copy if no processing needed
        import shutil
        shutil.copy2(input_path, output_path)
        return True


def process_directory(input_dir, output_dir, config):
    """
    Process all videos in a directory.
    
    Args:
        input_dir: Directory containing video files
        output_dir: Directory to save processed videos
        config: Configuration dictionary
    """
    video_formats = config['video']['formats']
    
    # Find all video files
    video_files = []
    for format in video_formats:
        video_files.extend(Path(input_dir).rglob(f"*.{format}"))
        video_files.extend(Path(input_dir).rglob(f"*.{format.upper()}"))
    
    if not video_files:
        print(f"No video files found in {input_dir}")
        return
    
    print(f"Found {len(video_files)} video files")
    
    # Process each video
    for video_path in tqdm(video_files, desc="Processing videos"):
        # Maintain directory structure in output
        relative_path = video_path.relative_to(input_dir)
        output_path = os.path.join(output_dir, relative_path)
        
        # Ensure output format is mp4
        output_path = str(Path(output_path).with_suffix('.mp4'))
        
        # Process video
        process_video(str(video_path), output_path, config, resize=True)


def main():
    parser = argparse.ArgumentParser(description='Process video files')
    parser.add_argument('--input', type=str, required=True,
                       help='Input directory containing video files')
    parser.add_argument('--output', type=str, required=True,
                       help='Output directory for processed videos')
    parser.add_argument('--config', type=str, default='config.yaml',
                       help='Path to configuration file')
    parser.add_argument('--no-resize', action='store_true',
                       help='Skip resizing videos')
    parser.add_argument('--convert', action='store_true',
                       help='Convert video format')
    
    args = parser.parse_args()
    
    # Load configuration
    config = load_config(args.config)
    
    # Process videos
    process_directory(args.input, args.output, config)
    
    print("Video processing completed!")


if __name__ == '__main__':
    main()

238 lines•7 KB

python

Theme Settings

Color Scheme

Display Options

Font Size