vidaio-subnet / services /compress /video_preprocessor.py

Arpan Tripathi

revise new compression miner difficulty (#100)

aa0f372 unverified 11 months ago

6.78 kB

import os
from utils.video_utils import get_video_duration, get_video_codec
from utils.encode_video import encode_lossless_video

def pre_processing(video_path, target_quality='Medium',codec='auto', max_duration=60, output_dir='./output'):
    """
    Part 1: Initial video checks and lossless encoding if necessary.
    This function performs the first stage of the video processing pipeline.
    It checks the video's duration and codec to determine if it's a candidate
    for processing. If the video is in a lossless format, it's re-encoded
    into a standardized lossless format (FFV1 in an MKV container) to ensure
    compatibility with later stages. Compressed videos are passed through
    unmodified, while videos exceeding the maximum duration are rejected.
    Args:
        video_path (str): The full path to the input video file.
        target_quality (str): Target quality level - 'High', 'Medium', or 'Low'.
                             Gets converted to VMAF scores: High=93, Medium=89, Low=85.
        max_duration (int): Maximum allowed video duration in seconds. Default: 3600 (1 hour).
        output_dir (str): Directory for final output files. Default: './output'.
        codec (str): Target encoding codec. Default: 'auto' (auto-detect best available).
                    Options: 'auto', 'av1_nvenc', 'libx264', 'libx265', 'h264_nvenc', etc.
       
    Returns:
        dict or None: Dictionary containing video metadata and processing info if successful, otherwise None.
                     Keys: 'path', 'codec', 'original_codec', 'duration', 'was_reencoded', 
                           'encoding_time', 'target_vmaf', 'target_quality', 'directories'
    """
    
    # ✅ CODEC HANDLING: Resolve auto codec selection
    if codec.lower() == 'auto':
        target_codec = 'av1_nvenc'  # Default to AV1 for best quality
        
    else:
        target_codec = codec
        print(f"🎯 Using specified codec: {target_codec}")
    
    # ✅ QUALITY MAPPING: Convert target quality to VMAF score
    quality_vmaf_mapping = {
        'High': 93.0,
        'Medium': 89.0,  # Updated from 93.0 to 90.0
        'Low': 85.0      # Updated from 90.0 to 85.0
    }
    
    # Validate and convert target quality
    if target_quality not in quality_vmaf_mapping:
        print(f"⚠️ Invalid target quality '{target_quality}'. Using 'Medium' as fallback.")
        target_quality = 'Medium'
    
    target_vmaf = quality_vmaf_mapping[target_quality]
    
    print(f"🎯 Target quality: {target_quality} (VMAF: {target_vmaf})")
    print(f"🎥 Target codec: {target_codec}")
    
    # ✅ DIRECTORY SETUP: Create output and temp directories
    os.makedirs(output_dir, exist_ok=True)
    
    print(f"📁 Output directory: {output_dir}")
    
    # ✅ DURATION CHECK: Validate video duration
    print(f"⏱️ Checking video duration...")
    duration = get_video_duration(video_path)
    
    if duration is None:
        print("❌ Could not determine video duration. Aborting.")
        return None
    
    if duration > max_duration:
        print(f"❌ Video duration {duration}s exceeds maximum allowed duration of {max_duration}s. Rejecting video.")
        return None
    
    print(f"✅ Video duration: {duration}s (within limit: {max_duration}s)")
    
    # CODEC CHECK: Determine video codec
    print(f"🎥 Checking video codec...")
    original_codec = get_video_codec(video_path)
    
    if not original_codec:
        print("❌ Could not determine video codec. Aborting.")
        return None
    
    print(f"🎥 Detected codec: {original_codec}")
    
    # LOSSLESS CODEC HANDLING: Define lossless codecs and special cases
    lossless_codecs = ['ffv1', 'h264_lossless', 'utvideo', 'rawvideo', 'prores_ks', 'dnxhd', 'cineform']
    lossless_extensions = ['.y4m', '.yuv', '.raw']
    
    is_lossless = (
        original_codec.lower() in lossless_codecs or 
        any(video_path.lower().endswith(ext) for ext in lossless_extensions)
    )
    
    if is_lossless:
        print(f"🔄 Video is in a lossless format ({original_codec}). Re-encoding with standardized lossless compression...")
        
        # Generate output filename with timestamp to avoid conflicts
        from datetime import datetime
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        base_name = os.path.splitext(os.path.basename(video_path))[0]
        output_filename = f"{base_name}_lossless_{timestamp}.mkv"
        output_path = os.path.join(output_dir, output_filename)
        
        print(f"📁 Lossless output: {output_filename}")
        
        # ✅ IMPROVED: Use direct FFmpeg command for lossless encoding
        # This bypasses the main encode_video function which expects lossy codecs
        encode_log, encode_time = encode_lossless_video(
            input_path=video_path,
            output_path=output_path,
            logging_enabled=True
        )
        
        if encode_log:
            print(f"✅ Lossless encoding successful in {encode_time:.1f}s")
            print(f"📁 Output at: {output_path}")
            
            # Return comprehensive metadata
            return {
                'path': output_path,
                'codec': 'ffv1',
                'original_codec': original_codec,
                'duration': duration,
                'was_reencoded': True,
                'encoding_time': encode_time,
                'target_vmaf': target_vmaf,
                'target_quality': target_quality,
                'target_codec': target_codec,
                'processing_info': {
                    'lossless_conversion': True,
                    'original_format': original_codec,
                    'standardized_format': 'ffv1',
                    'container': 'mkv'
                }
            }
        else:
            print("❌ Lossless encoding failed.")
            return None
    else:
        print(f"✅ Video is already compressed with codec: {original_codec}. Proceeding with original file.")
        
        # Return metadata for compressed video
        return {
            'path': video_path,
            'codec': original_codec,
            'original_codec': original_codec,
            'duration': duration,
            'was_reencoded': False,
            'encoding_time': 0,
            'target_vmaf': target_vmaf,
            'target_quality': target_quality,
            'target_codec': target_codec,  # ✅ Added target codec
            'processing_info': {
                'lossless_conversion': False,
                'original_format': original_codec,
                'standardized_format': original_codec,
                'container': os.path.splitext(video_path)[1][1:]  # Extension without dot
            }
        }

	import os
	from utils.video_utils import get_video_duration, get_video_codec
	from utils.encode_video import encode_lossless_video

	def pre_processing(video_path, target_quality='Medium',codec='auto', max_duration=60, output_dir='./output'):
	"""
	Part 1: Initial video checks and lossless encoding if necessary.

	This function performs the first stage of the video processing pipeline.
	It checks the video's duration and codec to determine if it's a candidate
	for processing. If the video is in a lossless format, it's re-encoded
	into a standardized lossless format (FFV1 in an MKV container) to ensure
	compatibility with later stages. Compressed videos are passed through
	unmodified, while videos exceeding the maximum duration are rejected.

	Args:
	video_path (str): The full path to the input video file.
	target_quality (str): Target quality level - 'High', 'Medium', or 'Low'.
	Gets converted to VMAF scores: High=93, Medium=89, Low=85.
	max_duration (int): Maximum allowed video duration in seconds. Default: 3600 (1 hour).
	output_dir (str): Directory for final output files. Default: './output'.
	codec (str): Target encoding codec. Default: 'auto' (auto-detect best available).
	Options: 'auto', 'av1_nvenc', 'libx264', 'libx265', 'h264_nvenc', etc.


	Returns:
	dict or None: Dictionary containing video metadata and processing info if successful, otherwise None.
	Keys: 'path', 'codec', 'original_codec', 'duration', 'was_reencoded',
	'encoding_time', 'target_vmaf', 'target_quality', 'directories'
	"""

	# ✅ CODEC HANDLING: Resolve auto codec selection
	if codec.lower() == 'auto':
	target_codec = 'av1_nvenc' # Default to AV1 for best quality

	else:
	target_codec = codec
	print(f"🎯 Using specified codec: {target_codec}")

	# ✅ QUALITY MAPPING: Convert target quality to VMAF score
	quality_vmaf_mapping = {
	'High': 93.0,
	'Medium': 89.0, # Updated from 93.0 to 90.0
	'Low': 85.0 # Updated from 90.0 to 85.0
	}

	# Validate and convert target quality
	if target_quality not in quality_vmaf_mapping:
	print(f"⚠️ Invalid target quality '{target_quality}'. Using 'Medium' as fallback.")
	target_quality = 'Medium'

	target_vmaf = quality_vmaf_mapping[target_quality]

	print(f"🎯 Target quality: {target_quality} (VMAF: {target_vmaf})")
	print(f"🎥 Target codec: {target_codec}")

	# ✅ DIRECTORY SETUP: Create output and temp directories
	os.makedirs(output_dir, exist_ok=True)

	print(f"📁 Output directory: {output_dir}")

	# ✅ DURATION CHECK: Validate video duration
	print(f"⏱️ Checking video duration...")
	duration = get_video_duration(video_path)

	if duration is None:
	print("❌ Could not determine video duration. Aborting.")
	return None

	if duration > max_duration:
	print(f"❌ Video duration {duration}s exceeds maximum allowed duration of {max_duration}s. Rejecting video.")
	return None

	print(f"✅ Video duration: {duration}s (within limit: {max_duration}s)")

	# CODEC CHECK: Determine video codec
	print(f"🎥 Checking video codec...")
	original_codec = get_video_codec(video_path)

	if not original_codec:
	print("❌ Could not determine video codec. Aborting.")
	return None

	print(f"🎥 Detected codec: {original_codec}")

	# LOSSLESS CODEC HANDLING: Define lossless codecs and special cases
	lossless_codecs = ['ffv1', 'h264_lossless', 'utvideo', 'rawvideo', 'prores_ks', 'dnxhd', 'cineform']
	lossless_extensions = ['.y4m', '.yuv', '.raw']

	is_lossless = (
	original_codec.lower() in lossless_codecs or
	any(video_path.lower().endswith(ext) for ext in lossless_extensions)
	)

	if is_lossless:
	print(f"🔄 Video is in a lossless format ({original_codec}). Re-encoding with standardized lossless compression...")

	# Generate output filename with timestamp to avoid conflicts
	from datetime import datetime
	timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
	base_name = os.path.splitext(os.path.basename(video_path))[0]
	output_filename = f"{base_name}_lossless_{timestamp}.mkv"
	output_path = os.path.join(output_dir, output_filename)

	print(f"📁 Lossless output: {output_filename}")

	# ✅ IMPROVED: Use direct FFmpeg command for lossless encoding
	# This bypasses the main encode_video function which expects lossy codecs
	encode_log, encode_time = encode_lossless_video(
	input_path=video_path,
	output_path=output_path,
	logging_enabled=True
	)

	if encode_log:
	print(f"✅ Lossless encoding successful in {encode_time:.1f}s")
	print(f"📁 Output at: {output_path}")

	# Return comprehensive metadata
	return {
	'path': output_path,
	'codec': 'ffv1',
	'original_codec': original_codec,
	'duration': duration,
	'was_reencoded': True,
	'encoding_time': encode_time,
	'target_vmaf': target_vmaf,
	'target_quality': target_quality,
	'target_codec': target_codec,
	'processing_info': {
	'lossless_conversion': True,
	'original_format': original_codec,
	'standardized_format': 'ffv1',
	'container': 'mkv'
	}
	}
	else:
	print("❌ Lossless encoding failed.")
	return None
	else:
	print(f"✅ Video is already compressed with codec: {original_codec}. Proceeding with original file.")

	# Return metadata for compressed video
	return {
	'path': video_path,
	'codec': original_codec,
	'original_codec': original_codec,
	'duration': duration,
	'was_reencoded': False,
	'encoding_time': 0,
	'target_vmaf': target_vmaf,
	'target_quality': target_quality,
	'target_codec': target_codec, # ✅ Added target codec
	'processing_info': {
	'lossless_conversion': False,
	'original_format': original_codec,
	'standardized_format': original_codec,
	'container': os.path.splitext(video_path)[1][1:] # Extension without dot
	}
	}