convert_equirect.py

import subprocess
import os
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

#FFPROBE = r"ffmpeg\ffprobe.exe"
#FFMPEG = r"ffmpeg\ffmpeg.exe"
FFPROBE = r"ffprobe"
FFMPEG = r"ffmpeg"
VIDEOS_DIR = "/nvme/2025_fred/panflow/"
NUM_WORKERS = 5


def check_and_convert(video_path: Path) -> str:
    """Check if video has spherical metadata type 3 and convert if needed."""
    try:
        # Run ffprobe to check for spherical metadata
        result = subprocess.run(
            [FFPROBE, str(video_path)],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace"
        )
        
        # Check stderr for the spherical metadata message (ffprobe outputs info to stderr)
        #output = result.stderr
        output = result.stderr

        if "Unknown spherical metadata type 3" not in output:
            return f"[SKIP] {video_path.name}: No spherical metadata type 3 found"
        
        # Build output filename
        stem = video_path.stem
        ext = video_path.suffix
        output_path = video_path.parent / f"{stem}_equirect{ext}"
        
        # Skip if output already exists
        if output_path.exists():
            return f"[SKIP] {video_path.name}: Output already exists"
        
        # Run ffmpeg conversion
        cmd = [
            FFMPEG,
            "-i", str(video_path),
            "-vf", "v360=eac:equirect:yaw=180:pitch=0:roll=0",
            "-an",
            str(output_path)
        ]
        
        convert_result = subprocess.run(
            cmd,
            capture_output=False,
            text=True,
            encoding="utf-8",
            errors="replace"
        )
        
        if convert_result.returncode == 0:
            return f"[OK] {video_path.name} -> {output_path.name}"
        else:
            return f"[ERROR] {video_path.name}: ffmpeg failed"
            
    except Exception as e:
        return f"[ERROR] {video_path.name}: {str(e)}"


def main():
    videos_path = Path(VIDEOS_DIR)
    
    # Get all video files
    video_extensions = {".mp4", ".webm", ".mkv", ".avi", ".mov"}
    video_files = [
        f for f in videos_path.iterdir()
        if f.is_file() and f.suffix.lower() in video_extensions and "_equirect" not in f.stem
    ]
    
    # Sort files by filename in ascending order
    video_files.sort(key=lambda x: x.name)
    
    print(f"Found {len(video_files)} video files to process")
    print(f"Using {NUM_WORKERS} workers")
    print("-" * 60)

    with ProcessPoolExecutor(max_workers=NUM_WORKERS) as executor:
        futures = {executor.submit(check_and_convert, video): video for video in video_files}
        
        for future in as_completed(futures):
            result = future.result()
            print(result)
    
    print("-" * 60)
    print("Done!")


if __name__ == "__main__":
    main()