diff --git a/.gitignore b/.gitignore index 8e09472b..faa53dbd 100644 --- a/.gitignore +++ b/.gitignore @@ -29,4 +29,6 @@ tmp/ vmaf/ services/scoring/vmaf services/upscaling/models/ -test.py \ No newline at end of file +test.py +ignore/ +test_outputs/ \ No newline at end of file diff --git a/neurons/miner.py b/neurons/miner.py index 3428f920..707993a0 100644 --- a/neurons/miner.py +++ b/neurons/miner.py @@ -62,30 +62,33 @@ async def forward_compression_requests(self, synapse: VideoCompressionProtocol) Processes a video compression request by downloading, compressing, uploading, and returning a sharing link. """ - + start_time = time.time() - + payload_url: str = synapse.miner_payload.reference_video_url vmaf_threshold: float = synapse.miner_payload.vmaf_threshold + target_codec: str = getattr(synapse.miner_payload, 'target_codec', 'av1') + codec_mode: str = getattr(synapse.miner_payload, 'codec_mode', 'CRF') + target_bitrate: float = getattr(synapse.miner_payload, 'target_bitrate', 10.0) validator_uid: int = self.metagraph.hotkeys.index(synapse.dendrite.hotkey) - - logger.info(f"šŸ›œšŸ›œšŸ›œ Receiving CompressionRequest from validator: {synapse.dendrite.hotkey} with uid: {validator_uid} šŸ›œšŸ›œšŸ›œ") + + logger.info(f"šŸ›œšŸ›œšŸ›œ Receiving CompressionRequest from validator: {synapse.dendrite.hotkey} with uid: {validator_uid} | VMAF: {vmaf_threshold} | Codec: {target_codec} | Mode: {codec_mode} | Bitrate: {target_bitrate} Mbps šŸ›œšŸ›œšŸ›œ") check_version(synapse.version) - + try: - processed_video_url = await video_compressor(payload_url, vmaf_threshold) - + processed_video_url = await video_compressor(payload_url, vmaf_threshold, target_codec, codec_mode, target_bitrate) + if processed_video_url is None: logger.info(f"šŸ’” Failed to compress video šŸ’”") return synapse - + synapse.miner_response.optimized_video_url = processed_video_url - + processed_time = time.time() - start_time logger.info(f"šŸ’œ Returning Response, Processed in {processed_time:.2f} seconds šŸ’œ") - + return synapse except Exception as e: diff --git a/requirements.txt b/requirements.txt index bb31c3a0..d8e1e33c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,7 +7,7 @@ numpy==2.0.2 minio==7.2.16 opencv-python==4.12.0.88 loguru==0.7.3 -bittensor==9.9.0 +bittensor>=9.9.0 redis==6.4.0 sqlalchemy==2.0.43 pandas==2.3.2 @@ -23,8 +23,8 @@ wandb==0.22.0 apscheduler==3.11.0 boto3==1.40.34 google-auth-oauthlib==1.2.2 -google-auth==2.40.3 -google-api-python-client==2.182.0 +google-auth>=2.40.3 +google-api-python-client>=2.182.0 yt-dlp==2025.9.5 selenium==4.35.0 scikit-learn==1.7.2 @@ -33,7 +33,7 @@ scipy==1.16.2 torch==2.8.0 torchvision==0.23.0 torchaudio==2.8.0 -tensorflow==2.19.1 +tensorflow>=2.19.1 tensorboard==2.19.0 # Video and Image Processing diff --git a/services/compress/encoder.py b/services/compress/encoder.py index f13bf930..23f4a618 100644 --- a/services/compress/encoder.py +++ b/services/compress/encoder.py @@ -581,21 +581,39 @@ def safe_positive_float(value, default=1.0): target_vmaf = safe_float(target_vmaf or original_video_metadata.get('target_vmaf') or config.get('video_processing', {}).get('target_vmaf', 0.0), 0.0) - original_codec = original_video_metadata.get('original_codec', 'unknown') + # Get codec from config (set by the request) + # Priority: config['video_processing']['target_codec'] > metadata > fallback + target_codec_from_config = config.get('video_processing', {}).get('target_codec') target_codec_from_part1 = original_video_metadata.get('target_codec', 'auto') + original_codec = original_video_metadata.get('original_codec', 'unknown') current_codec = original_video_metadata.get('codec', original_codec) - config_codec = config.get('video_processing', {}).get('codec', 'auto') - - if target_codec_from_part1 and target_codec_from_part1 != 'auto': + + # Use target_codec from config (highest priority - comes from the request) + if target_codec_from_config and target_codec_from_config != 'auto': + codec = target_codec_from_config + if logging_enabled: + print(f" šŸŽÆ Using target codec from request: {codec}") + # Fallback to Part 1 metadata + elif target_codec_from_part1 and target_codec_from_part1 != 'auto': codec = target_codec_from_part1 - elif config_codec != 'auto': - codec = config_codec + if logging_enabled: + print(f" šŸ“‹ Using target codec from metadata: {codec}") + # Final fallback to upgrade map else: codec_upgrade_map = {'h264': 'av1_nvenc', 'hevc': 'av1_nvenc', 'vp9': 'av1_nvenc', 'av1': 'av1_nvenc'} codec = codec_upgrade_map.get(current_codec.lower(), 'av1_nvenc') + if logging_enabled: + print(f" āš” Using fallback codec mapping: {current_codec} ā†’ {codec}") + + if logging_enabled: + print(f" šŸŽ„ Final Selected Codec: {codec}") + + # Get codec_mode and target_bitrate from config (set by the request) + codec_mode = config.get('video_processing', {}).get('codec_mode', 'CRF') + target_bitrate = config.get('video_processing', {}).get('target_bitrate', 10.0) if logging_enabled: - print(f" šŸŽ„ Selected Codec: {codec}") + print(f" āš™ļø Codec Mode: {codec_mode}, Target Bitrate: {target_bitrate} Mbps") # STEP 1: BASIC ANALYSIS - Classify scene and extract video features in one call if logging_enabled: @@ -741,7 +759,9 @@ def safe_positive_float(value, default=1.0): codec=codec, adjusted_cq=final_cq, content_type=scene_type, - contrast_value=0.5, + contrast_value=0.5, + codec_mode=codec_mode, + target_bitrate=target_bitrate, max_retries=max_encoding_retries, logging_enabled=logging_enabled ) @@ -758,7 +778,9 @@ def safe_positive_float(value, default=1.0): codec=codec, rate=final_cq, scene_type=scene_type, - contrast_value=0.5, # Default contrast + contrast_value=0.5, # Default contrast + codec_mode=codec_mode, + target_bitrate=target_bitrate, logging_enabled=logging_enabled ) encoded_path = output_scene_path diff --git a/services/compress/server.py b/services/compress/server.py index ea403ded..e7c9983b 100644 --- a/services/compress/server.py +++ b/services/compress/server.py @@ -23,6 +23,17 @@ from validator_merger import validation_and_merging from vidaio_subnet_core.utilities import storage_client, download_video from vidaio_subnet_core import CONFIG +from utils.video_utils import get_video_duration, get_video_codec + + +# ============================================================================ +# Configuration Variables +# ============================================================================ + +# VMAF threshold to quality level mapping (configurable for miner flow) +VMAF_THRESHOLD_HIGH = 95.0 +VMAF_THRESHOLD_MEDIUM = 90.0 +VMAF_THRESHOLD_LOW = 85.0 # ============================================================================ @@ -40,7 +51,10 @@ class CompressPayload(BaseModel): """Payload for video compression requests.""" payload_url: str vmaf_threshold: float - target_quality: str = 'Medium' # High, Medium, Low + target_codec: str = 'av1' # Target codec: av1, hevc, h264, vp9 + codec_mode: str = 'CRF' # Codec mode: CRF (Constant Rate Factor), CBR (Constant Bitrate), VBR (Variable Bitrate) + target_bitrate: float = 10.0 # Target bitrate in Mbps (for CBR/VBR modes) + target_quality: str = 'Medium' # High, Medium, Low (legacy, derived from VMAF) max_duration: int = 3600 # Maximum allowed video duration in seconds output_dir: str = './output' # Output directory for final files @@ -50,6 +64,119 @@ class TestCompressPayload(BaseModel): video_path: str +# ============================================================================ +# Helper Functions +# ============================================================================ + +def create_lightweight_metadata(input_file: str, target_quality: str, target_codec: str, max_duration: int = 3600) -> Optional[dict]: + """ + Create video metadata without preprocessing (for already-compressed videos). + + This is a lightweight alternative to pre_processing() that only extracts + metadata without re-encoding. Perfect for miner chunks that are already compressed. + + Args: + input_file: Path to input video file + target_quality: Target quality level ('High', 'Medium', 'Low') + target_codec: Target codec for encoding + max_duration: Maximum allowed duration + + Returns: + dict: Video metadata or None if validation fails + """ + print(f"\nāš” === Part 1: Pre-processing (SKIPPED - Lightweight Metadata) ===") + print(f" šŸ“ Extracting metadata from already-compressed video") + + # Map quality to VMAF using configurable thresholds + quality_vmaf_mapping = { + 'High': VMAF_THRESHOLD_HIGH, + 'Medium': VMAF_THRESHOLD_MEDIUM, + 'Low': VMAF_THRESHOLD_LOW + } + target_vmaf = quality_vmaf_mapping.get(target_quality, VMAF_THRESHOLD_MEDIUM) + + # Get video duration + duration = get_video_duration(input_file) + if duration is None: + print(" āŒ Could not determine video duration") + return None + + if duration > max_duration: + print(f" āŒ Video duration {duration}s exceeds limit of {max_duration}s") + return None + + # Get video codec + original_codec = get_video_codec(input_file) + if not original_codec: + print(" āŒ Could not determine video codec") + return None + + print(f" āœ… Duration: {duration:.1f}s") + print(f" āœ… Codec: {original_codec}") + print(f" šŸŽÆ Target: {target_quality} (VMAF: {target_vmaf})") + print(f" šŸŽ„ Target codec: {target_codec}") + + # Return lightweight metadata (same format as pre_processing) + return { + 'path': input_file, + 'codec': original_codec, + 'original_codec': original_codec, + 'duration': duration, + 'was_reencoded': False, + 'encoding_time': 0.0, + 'target_vmaf': target_vmaf, + 'target_quality': target_quality, + 'target_codec': target_codec, + 'processing_info': { + 'lossless_conversion': False, + 'skipped_preprocessing': True + } + } + + +# ============================================================================ +# Codec Mapping +# ============================================================================ + +def map_codec_name(target_codec: str, prefer_gpu: bool = True) -> str: + """ + Map user-facing codec names (from ffprobe format) to ffmpeg encoder names. + + The protocol uses standard codec names (av1, hevc, h264, vp9) which match + ffprobe output, but ffmpeg requires specific encoder names. + + Args: + target_codec: Codec name from protocol (av1, hevc, h264, vp9) + prefer_gpu: Whether to prefer GPU encoders (NVENC) when available + + Returns: + FFmpeg encoder name (e.g., av1_nvenc, libx265, libx264, libvpx-vp9) + + Examples: + map_codec_name('av1', prefer_gpu=True) ā†’ 'av1_nvenc' + map_codec_name('av1', prefer_gpu=False) ā†’ 'libsvtav1' + map_codec_name('hevc', prefer_gpu=True) ā†’ 'hevc_nvenc' + map_codec_name('h264', prefer_gpu=False) ā†’ 'libx264' + """ + codec_map = { + 'av1': 'av1_nvenc' if prefer_gpu else 'libsvtav1', + 'hevc': 'hevc_nvenc' if prefer_gpu else 'libx265', + 'h264': 'h264_nvenc' if prefer_gpu else 'libx264', + 'vp9': 'libvpx_vp9', # No NVENC encoder for VP9 + } + + # Normalize to lowercase and get mapped codec + normalized_codec = target_codec.lower().strip() + ffmpeg_codec = codec_map.get(normalized_codec) + + if not ffmpeg_codec: + logger.warning(f"Unknown codec '{target_codec}', defaulting to av1_nvenc") + return 'av1_nvenc' + + logger.info(f"Mapped codec '{target_codec}' ā†’ '{ffmpeg_codec}' (GPU={prefer_gpu})") + return ffmpeg_codec + + # ============================================================================ # API Endpoints # ============================================================================ @@ -67,21 +194,30 @@ async def compress_video(video: CompressPayload): """ print(f"video url: {video.payload_url}") print(f"vmaf threshold: {video.vmaf_threshold}") - + print(f"target codec: {video.target_codec}") + print(f"codec mode: {video.codec_mode}") + print(f"target bitrate: {video.target_bitrate} Mbps") + # Download video from URL input_path = await download_video(video.payload_url) input_file = Path(input_path) vmaf_threshold = video.vmaf_threshold - # Map VMAF threshold to target quality - if vmaf_threshold == 85: + # Map VMAF threshold to target quality using configurable thresholds + if vmaf_threshold == VMAF_THRESHOLD_LOW: target_quality = 'Low' - elif vmaf_threshold == 89: + elif vmaf_threshold == VMAF_THRESHOLD_MEDIUM: target_quality = 'Medium' - elif vmaf_threshold == 93: + elif vmaf_threshold == VMAF_THRESHOLD_HIGH: target_quality = 'High' else: - raise HTTPException(status_code=400, detail="Invalid VMAF threshold.") + raise HTTPException( + status_code=400, + detail=f"Invalid VMAF threshold. Expected {VMAF_THRESHOLD_LOW}, {VMAF_THRESHOLD_MEDIUM}, or {VMAF_THRESHOLD_HIGH}, got {vmaf_threshold}" + ) + + # Map codec name from protocol format to ffmpeg encoder name + ffmpeg_codec = map_codec_name(video.target_codec, prefer_gpu=True) # Validate input file if not input_file.is_file(): @@ -96,8 +232,13 @@ async def compress_video(video: CompressPayload): compressed_video_path = video_compressor( input_file=str(input_file), target_quality=target_quality, + target_codec=ffmpeg_codec, + codec_mode=video.codec_mode, + target_bitrate=video.target_bitrate, max_duration=video.max_duration, - output_dir=str(output_dir) + output_dir=str(output_dir), + skip_scene_detection=True, # Skip for miner chunks (already pre-split) + skip_preprocessing=True # Skip for miner chunks (already compressed) ) print(f"compressed_video_path: {compressed_video_path}") @@ -192,20 +333,30 @@ async def test_compress_video(test_payload: TestCompressPayload): # ============================================================================ def video_compressor( - input_file: str, - target_quality: str = 'Medium', - max_duration: int = 3600, - output_dir: str = './output' + input_file: str, + target_quality: str = 'Medium', + target_codec: str = 'av1_nvenc', + codec_mode: str = 'CRF', + target_bitrate: float = 10.0, + max_duration: int = 3600, + output_dir: str = './output', + skip_scene_detection: bool = True, + skip_preprocessing: bool = True ) -> Optional[str]: """ Main video compression pipeline orchestrator. - + Args: input_file: Path to input video file target_quality: Target quality level ('High', 'Medium', 'Low') + target_codec: FFmpeg encoder name (e.g., 'av1_nvenc', 'hevc_nvenc', 'libx264') + codec_mode: Encoding mode - 'CRF' (Constant Rate Factor), 'CBR' (Constant Bitrate), 'VBR' (Variable Bitrate) + target_bitrate: Target bitrate in Mbps (used for CBR/VBR modes) max_duration: Maximum allowed video duration in seconds output_dir: Output directory for final files - + skip_scene_detection: If True, treats entire video as single scene (default: True for miner chunks) + skip_preprocessing: If True, skips lossless re-encoding (default: True for already-compressed miner chunks) + Returns: str: Path to compressed video file, or None if failed """ @@ -223,6 +374,9 @@ def video_compressor( if 'video_processing' not in config: config['video_processing'] = {} config['video_processing']['target_quality'] = target_quality + config['video_processing']['target_codec'] = target_codec + config['video_processing']['codec_mode'] = codec_mode + config['video_processing']['target_bitrate'] = target_bitrate # Create temp directory temp_dir = Path(config['directories']['temp_dir']) @@ -231,25 +385,53 @@ def video_compressor( # Display pipeline information _display_pipeline_info(input_file, target_quality, max_duration, output_dir) - # PART 1: Pre-processing - part1_result = _execute_preprocessing(input_file, target_quality, max_duration, output_dir_path) - if not part1_result: - print("āŒ Part 1 failed. Pipeline terminated.") - return False - - part1_time = time.time() - pipeline_start_time - _display_preprocessing_results(part1_result, part1_time) - - # PART 2: Scene Detection + # PART 1: Pre-processing (optional - can be skipped for already-compressed videos) + if skip_preprocessing: + # Skip full preprocessing - use lightweight metadata extraction + part1_result = create_lightweight_metadata(input_file, target_quality, target_codec, max_duration) + if not part1_result: + print("āŒ Part 1 failed. Pipeline terminated.") + return False + part1_time = time.time() - pipeline_start_time + print(f" ā±ļø Metadata extraction: {part1_time:.2f}s") + else: + # Run full preprocessing (checks for lossless codecs and re-encodes if needed) + part1_result = _execute_preprocessing(input_file, target_quality, max_duration, output_dir_path) + if not part1_result: + print("āŒ Part 1 failed. Pipeline terminated.") + return False + part1_time = time.time() - pipeline_start_time + _display_preprocessing_results(part1_result, part1_time) + + # PART 2: Scene Detection (optional - can be skipped for pre-chunked videos) part2_start_time = time.time() - scenes_metadata = scene_detection(part1_result) - if not scenes_metadata: - print("āŒ Part 2 failed. Pipeline terminated.") - return False - - part2_time = time.time() - part2_start_time - _display_scene_detection_results(scenes_metadata, part2_time) + if skip_scene_detection: + # Skip scene detection - treat entire video as single scene + print(f"\nāš” === Part 2: Scene Detection (SKIPPED) ===") + print(f" šŸ“ Treating entire video as single scene (pre-chunked input)") + + scenes_metadata = [{ + 'path': part1_result['path'], + 'scene_number': 1, + 'start_time': 0.0, + 'end_time': part1_result['duration'], + 'duration': part1_result['duration'], + 'original_video_metadata': part1_result + }] + part2_time = time.time() - part2_start_time + print(f" ā±ļø Scene setup: {part2_time:.2f}s") + print(f" āœ… 1 scene created (0.0s - {part1_result['duration']:.1f}s)") + else: + # Run normal scene detection + scenes_metadata = scene_detection(part1_result) + if not scenes_metadata: + print("āŒ Part 2 failed. Pipeline terminated.") + return False + + part2_time = time.time() - part2_start_time + _display_scene_detection_results(scenes_metadata, part2_time) + # PART 3: AI Encoding part3_result = _execute_ai_encoding(scenes_metadata, config, target_quality) if not part3_result: @@ -366,7 +548,10 @@ def _get_default_config() -> dict: 'video_processing': { 'SHORT_VIDEO_THRESHOLD': 20, 'target_vmaf': 93.0, - 'codec': 'auto', + 'codec': 'auto', # Legacy parameter, overridden by target_codec + 'target_codec': 'av1_nvenc', # Will be overridden by request + 'codec_mode': 'CRF', # Will be overridden by request + 'target_bitrate': 10.0, # Will be overridden by request 'size_increase_protection': True, 'conservative_cq_adjustment': 2, 'max_output_size_ratio': 1.15, diff --git a/services/compress/utils/encode_video.py b/services/compress/utils/encode_video.py index 37dc36a6..a79991fb 100644 --- a/services/compress/utils/encode_video.py +++ b/services/compress/utils/encode_video.py @@ -9,6 +9,123 @@ import numpy as np from .encoder_configs import ENCODER_SETTINGS, SCENE_SPECIFIC_PARAMS, MODEL_CQ_REFERENCE_CODEC, QUALITY_MAPPING_ANCHORS +def cleanup_quality_params(settings, keep_param=None): + """ + Remove conflicting quality parameters (CRF, CQ, QP). + + Args: + settings (dict): Current settings dictionary + keep_param (str, optional): Which parameter to keep ('crf', 'cq', or 'qp') + If None, removes all quality parameters + + Returns: + dict: Updated settings with cleaned quality parameters + """ + quality_params = ['crf', 'cq', 'qp'] + + if keep_param: + # Remove all except the one we want to keep + for param in quality_params: + if param != keep_param: + settings.pop(param, None) + else: + # Remove all quality parameters + for param in quality_params: + settings.pop(param, None) + + return settings + + +def apply_rate_mapping(codec, rate, current_settings, logging_enabled=True): + """ + Apply rate (CQ/CRF) mapping to current settings based on codec type. + + Args: + codec (str): Video codec name + rate (int/float): Model-predicted CQ value + current_settings (dict): Current encoder settings + logging_enabled (bool): Enable logging + + Returns: + dict: Updated settings with rate parameter applied + """ + model_predicted_ref_cq = int(rate) + + if codec == MODEL_CQ_REFERENCE_CODEC: + # Use the predicted CQ directly for the reference codec + if 'cq' in current_settings: + current_settings['cq'] = model_predicted_ref_cq + cleanup_quality_params(current_settings, keep_param='cq') + if logging_enabled: + print(f"Applying model CQ directly for {MODEL_CQ_REFERENCE_CODEC}: {current_settings['cq']}") + elif 'crf' in current_settings: + current_settings['crf'] = model_predicted_ref_cq + cleanup_quality_params(current_settings, keep_param='crf') + if logging_enabled: + print(f"Applying model CQ as CRF for {MODEL_CQ_REFERENCE_CODEC}: {current_settings['crf']}") + else: + if logging_enabled: + print(f"Warning: Neither 'cq' nor 'crf' in base settings for {MODEL_CQ_REFERENCE_CODEC}. Using CRF fallback.") + current_settings['crf'] = model_predicted_ref_cq + + elif codec in QUALITY_MAPPING_ANCHORS: + # Use quality mapping for non-reference codecs + mapping_config = QUALITY_MAPPING_ANCHORS[codec] + + model_anchor_cqs = [p[0] for p in mapping_config['anchor_points']] + target_anchor_params = [p[1] for p in mapping_config['anchor_points']] + + # Interpolate with clamping + clamped_model_cq = np.clip(model_predicted_ref_cq, + mapping_config['model_ref_cq_range'][0], + mapping_config['model_ref_cq_range'][1]) + if logging_enabled and clamped_model_cq != model_predicted_ref_cq: + print(f"Clamped model_predicted_ref_cq from {model_predicted_ref_cq} to {clamped_model_cq}") + + mapped_param_float = np.interp(clamped_model_cq, model_anchor_cqs, target_anchor_params) + + # Clamp to target parameter range + min_target_param, max_target_param = mapping_config['target_param_range'] + mapped_param_clamped = np.clip(mapped_param_float, min_target_param, max_target_param) + mapped_param_int = int(round(mapped_param_clamped)) + + target_param_type = mapping_config['target_param_type'] + if target_param_type == 'cq': + current_settings['cq'] = mapped_param_int + cleanup_quality_params(current_settings, keep_param='cq') + if logging_enabled: + print(f"Applying mapped CQ for {codec} from model ref CQ {model_predicted_ref_cq}: {current_settings['cq']}") + elif target_param_type == 'crf': + current_settings['crf'] = mapped_param_int + cleanup_quality_params(current_settings, keep_param='crf') + if logging_enabled: + print(f"Applying mapped CRF for {codec} from model ref CQ {model_predicted_ref_cq}: {current_settings['crf']}") + else: + if logging_enabled: + print(f"Warning: Unknown target_param_type '{target_param_type}' for {codec}. Using fallback.") + if 'cq' in current_settings: + current_settings['cq'] = model_predicted_ref_cq + elif 'crf' in current_settings: + current_settings['crf'] = model_predicted_ref_cq + + else: + # Fallback for codecs without mapping + if logging_enabled: + print(f"Warning: No quality mapping found for {codec}. Applying model ref CQ {model_predicted_ref_cq} directly.") + if 'cq' in current_settings: + current_settings['cq'] = model_predicted_ref_cq + cleanup_quality_params(current_settings, keep_param='cq') + elif 'crf' in current_settings: + current_settings['crf'] = model_predicted_ref_cq + cleanup_quality_params(current_settings, keep_param='crf') + else: + current_settings['crf'] = model_predicted_ref_cq + if logging_enabled: + print(f"Applied direct rate {model_predicted_ref_cq} to {codec}.") + + return current_settings + + def get_contrast_optimized_params(scene_type, contrast_value, codec): """ Get contrast-optimized encoding parameters for a specific scene type and codec. @@ -132,24 +249,25 @@ def get_contrast_optimized_params(scene_type, contrast_value, codec): return params -def encode_video(input_path, output_path, codec, rate=None, max_bit_rate=None, preset=None, scene_type=None, contrast_value=None, logging_enabled=True): +def encode_video(input_path, output_path, codec, rate=None, preset=None, scene_type=None, contrast_value=None, codec_mode=None, target_bitrate=None, logging_enabled=True): """ Encodes a video using specified codec settings, optimized for scene type and contrast. Audio is copied without re-encoding for efficiency. - + Note: encoding decisions should be made BEFORE calling this function. - + Args: input_path (str): Path to input video file output_path (str): Path for output video file codec (str): Video codec to use (e.g., 'av1_nvenc', 'libx264') rate (int/float, optional): Quality parameter (CQ value from model) - max_bit_rate (str, optional): Maximum bitrate (e.g., '5000k', '10m') preset (str, optional): Encoder preset override scene_type (str, optional): Scene classification for optimization contrast_value (float, optional): Perceptual contrast (0.0-1.0) + codec_mode (str, optional): Encoding mode - 'CRF', 'CBR', or 'VBR' + target_bitrate (float, optional): Target bitrate in Mbps (for CBR/VBR modes) logging_enabled (bool): Enable detailed logging - + Returns: tuple: (encoding_results_log, encoding_time) or (None, None) on failure """ @@ -186,172 +304,107 @@ def encode_video(input_path, output_path, codec, rate=None, max_bit_rate=None, p print(f"Applying contrast-specific params: {contrast_params}") current_settings.update(contrast_params) - # 4. Apply rate (CQ) parameter with codec-specific handling - if rate is not None: - model_predicted_ref_cq = int(rate) - - if codec == MODEL_CQ_REFERENCE_CODEC: - # Use the predicted CQ directly for the reference codec - if 'cq' in current_settings: - current_settings['cq'] = model_predicted_ref_cq - if 'crf' in current_settings: - del current_settings['crf'] - # Use VBR for NVENC/QSV with CQ - if codec.endswith("_nvenc") or codec.endswith("_qsv"): - current_settings['rc'] = 'vbr' - if 'qp' in current_settings: - del current_settings['qp'] - if logging_enabled: - print(f"Applying model CQ directly for {MODEL_CQ_REFERENCE_CODEC}: {current_settings['cq']}") - elif 'crf' in current_settings: - current_settings['crf'] = model_predicted_ref_cq - if 'cq' in current_settings: - del current_settings['cq'] - if logging_enabled: - print(f"Applying model CQ as CRF for {MODEL_CQ_REFERENCE_CODEC}: {current_settings['crf']}") - else: - if logging_enabled: - print(f"Warning: Neither 'cq' nor 'crf' in base settings for {MODEL_CQ_REFERENCE_CODEC}. Using CRF fallback.") - current_settings['crf'] = model_predicted_ref_cq + # 4. Apply codec_mode and target_bitrate if provided (BEFORE rate mapping) + # This determines whether we should apply CQ/CRF or use bitrate control + skip_rate_application = False + if codec_mode and target_bitrate: + if logging_enabled: + print(f"Applying codec_mode='{codec_mode}' with target_bitrate={target_bitrate} Mbps") - elif codec in QUALITY_MAPPING_ANCHORS: - # Use quality mapping for non-reference codecs - mapping_config = QUALITY_MAPPING_ANCHORS[codec] - - model_anchor_cqs = [p[0] for p in mapping_config['anchor_points']] - target_anchor_params = [p[1] for p in mapping_config['anchor_points']] + if codec_mode.upper() == 'CBR': + # Constant Bitrate Mode - fixed bitrate throughout + skip_rate_application = True # Don't apply CQ/CRF in CBR mode + bitrate_kbps = int(target_bitrate * 1000) # Convert Mbps to kbps + bitrate_value = f"{bitrate_kbps}k" - # Interpolate with clamping - clamped_model_cq = np.clip(model_predicted_ref_cq, - mapping_config['model_ref_cq_range'][0], - mapping_config['model_ref_cq_range'][1]) - if logging_enabled and clamped_model_cq != model_predicted_ref_cq: - print(f"Clamped model_predicted_ref_cq from {model_predicted_ref_cq} to {clamped_model_cq}") + current_settings['bitrate'] = bitrate_value + current_settings['maxrate'] = bitrate_value + current_settings['bufsize'] = f"{bitrate_kbps * 2}k" # Buffer = 2x bitrate - mapped_param_float = np.interp(clamped_model_cq, model_anchor_cqs, target_anchor_params) - - # Clamp to target parameter range - min_target_param, max_target_param = mapping_config['target_param_range'] - mapped_param_clamped = np.clip(mapped_param_float, min_target_param, max_target_param) - mapped_param_int = int(round(mapped_param_clamped)) - - target_param_type = mapping_config['target_param_type'] - if target_param_type == 'cq': - current_settings['cq'] = mapped_param_int - if 'crf' in current_settings: - del current_settings['crf'] - # Use VBR for NVENC and QSV when using CQ - if codec.endswith("_nvenc") or codec.endswith("_qsv"): - current_settings['rc'] = 'vbr' - if 'qp' in current_settings: - del current_settings['qp'] - if logging_enabled: - print(f"Applying mapped CQ (VBR) for {codec} from model ref CQ {model_predicted_ref_cq}: {current_settings['cq']}") - elif target_param_type == 'crf': - current_settings['crf'] = mapped_param_int - if 'cq' in current_settings: - del current_settings['cq'] - if logging_enabled: - print(f"Applying mapped CRF for {codec} from model ref CQ {model_predicted_ref_cq}: {current_settings['crf']}") - else: - if logging_enabled: - print(f"Warning: Unknown target_param_type '{target_param_type}' for {codec}. Using fallback.") - if 'cq' in current_settings: - current_settings['cq'] = model_predicted_ref_cq - elif 'crf' in current_settings: - current_settings['crf'] = model_predicted_ref_cq + # Remove quality parameters (incompatible with CBR) + cleanup_quality_params(current_settings) + + # Set rate control mode for NVENC/QSV + if codec.endswith('_nvenc'): + current_settings['rc'] = 'cbr' + if logging_enabled: + print(f"Set NVENC rate control to CBR") + elif codec.endswith('_qsv'): + current_settings['rc'] = 'cbr' + + if logging_enabled: + print(f"CBR mode: bitrate={bitrate_value}, maxrate={bitrate_value}, bufsize={bitrate_kbps * 2}k") + + elif codec_mode.upper() == 'VBR': + # Variable Bitrate Mode - allows bitrate to vary but caps at target + # We will still apply CQ/CRF for quality, but add maxrate constraint + bitrate_kbps = int(target_bitrate * 1000) + bitrate_value = f"{bitrate_kbps}k" + + current_settings['maxrate'] = bitrate_value + current_settings['bufsize'] = f"{bitrate_kbps * 2}k" + + # Set rate control mode for NVENC/QSV + if codec.endswith('_nvenc') or codec.endswith('_qsv'): + current_settings['rc'] = 'vbr' + if logging_enabled: + print(f"Set rate control to VBR") + + if logging_enabled: + print(f"VBR mode: maxrate={bitrate_value}, bufsize={bitrate_kbps * 2}k, will apply CQ/CRF for quality") + + elif codec_mode.upper() == 'CRF': + # CRF mode is the default - will apply rate parameter below + if logging_enabled: + print(f"CRF mode: Will apply quality-based encoding (rate={rate})") else: - # Fallback for codecs without mapping if logging_enabled: - print(f"Warning: No quality mapping found for {codec}. Applying model ref CQ {model_predicted_ref_cq} directly.") - if 'cq' in current_settings: - current_settings['cq'] = model_predicted_ref_cq - if 'crf' in current_settings: - del current_settings['crf'] - # Use VBR for NVENC and QSV with CQ - if codec.endswith("_nvenc") or codec.endswith("_qsv"): - current_settings['rc'] = 'vbr' - if 'qp' in current_settings: - del current_settings['qp'] - elif 'crf' in current_settings: - current_settings['crf'] = model_predicted_ref_cq - if 'cq' in current_settings: - del current_settings['cq'] - else: - current_settings['crf'] = model_predicted_ref_cq - if logging_enabled: - print(f"Applied direct rate {model_predicted_ref_cq} to {codec}.") + print(f"Warning: Unknown codec_mode '{codec_mode}', defaulting to CRF behavior") - # 5. Apply preset override if provided + # 5. Apply rate (CQ) parameter with codec-specific handling (skip if CBR mode) + if rate is not None and not skip_rate_application: + current_settings = apply_rate_mapping(codec, rate, current_settings, logging_enabled) + + # 6. Apply preset override if provided if preset is not None: current_settings['preset'] = preset if logging_enabled: print(f"Applying preset override: {preset}") - # 6. Apply max_bit_rate and enable VBR for NVENC/QSV when maxrate is provided - if max_bit_rate is not None: - current_settings['maxrate'] = max_bit_rate - try: - # Extract numeric part and unit (k or m) - numeric_maxrate = int(re.sub(r'\D', '', max_bit_rate)) - unit = re.sub(r'\d', '', max_bit_rate).lower() - if unit not in ['k', 'm']: - unit = 'k' - bufsize_val = numeric_maxrate * 2 - current_settings['bufsize'] = f"{bufsize_val}{unit}" - - # Enable VBR when maxrate is provided for NVENC/QSV - if codec.endswith("_nvenc") or codec.endswith("_qsv"): - current_settings['rc'] = 'vbr' - if logging_enabled: - print(f"Enabled VBR rate control for {codec} due to maxrate setting") - - if logging_enabled: - print(f"Applying maxrate: {current_settings['maxrate']}, bufsize: {current_settings['bufsize']}") - except (ValueError, Exception) as e: - print(f"Warning: Could not parse max_bit_rate '{max_bit_rate}': {e}. Ignoring.") - # 7. Handle CRF usage - disable constqp when CRF is used try: uses_crf = 'crf' in current_settings has_maxrate = 'maxrate' in current_settings - + if uses_crf: # Remove constqp explicitly when CRF is used if str(current_settings.get('rc', '')).lower() == 'constqp': if logging_enabled: print("Disabling 'constqp' because CRF is in use") del current_settings['rc'] - - # For NVENC/QSV with CRF + maxrate, use VBR - if (codec.endswith('_nvenc') or codec.endswith('_qsv')) and has_maxrate: + + # For NVENC/QSV with CRF + maxrate, use VBR (only if rc not already set by codec_mode) + if (codec.endswith('_nvenc') or codec.endswith('_qsv')) and has_maxrate and 'rc' not in current_settings: current_settings['rc'] = 'vbr' if logging_enabled: print(f"Using 'vbr' with CRF due to specified maxrate") except Exception: pass - # 8. CQ policy for NVENC/QSV - use VBR with CQ unless maxrate + CQ requires constqp + # 8. CQ policy for NVENC/QSV - use VBR with CQ (only if rc not already set by codec_mode) try: has_cq = 'cq' in current_settings and isinstance(current_settings.get('cq'), (int, float)) has_maxrate = 'maxrate' in current_settings - - if (codec.endswith('_nvenc') or codec.endswith('_qsv')) and has_cq: - if has_maxrate: - # Use VBR with both CQ and maxrate for rate-limited quality encoding - current_settings['rc'] = 'vbr' - if 'qp' in current_settings: - del current_settings['qp'] - if logging_enabled: - print(f"Using VBR with CQ and maxrate for {codec}") - else: - # Standard CQ encoding with VBR - current_settings['rc'] = 'vbr' - if 'qp' in current_settings: - del current_settings['qp'] - if logging_enabled: - print(f"Using VBR with CQ for {codec}") + rc_already_set = 'rc' in current_settings + + if (codec.endswith('_nvenc') or codec.endswith('_qsv')) and has_cq and not rc_already_set: + # Standard CQ encoding with VBR (fallback when codec_mode not specified) + current_settings['rc'] = 'vbr' + cleanup_quality_params(current_settings, keep_param='cq') + if logging_enabled: + maxrate_info = " and maxrate" if has_maxrate else "" + print(f"Using VBR with CQ{maxrate_info} for {codec}") except Exception: pass diff --git a/services/compress/utils/processing_utils.py b/services/compress/utils/processing_utils.py index d349c364..c422a9d8 100644 --- a/services/compress/utils/processing_utils.py +++ b/services/compress/utils/processing_utils.py @@ -94,7 +94,7 @@ def should_skip_encoding(input_path, estimated_cq, target_vmaf, logging_enabled= return False, adjusted_cq -def encode_scene_with_size_check(scene_path, output_path, codec, adjusted_cq, content_type, contrast_value, max_retries=2, logging_enabled=True): +def encode_scene_with_size_check(scene_path, output_path, codec, adjusted_cq, content_type, contrast_value, codec_mode=None, target_bitrate=None, max_retries=2, logging_enabled=True): original_size = os.path.getsize(scene_path) input_analysis = analyze_input_compression(scene_path) @@ -137,6 +137,8 @@ def encode_scene_with_size_check(scene_path, output_path, codec, adjusted_cq, co rate=retry_cq, # FIXED: Use retry_cq instead of adjusted_cq scene_type=content_type if attempt == 0 else None, contrast_value=contrast_value if attempt == 0 else None, + codec_mode=codec_mode, + target_bitrate=target_bitrate, logging_enabled=logging_enabled ) diff --git a/services/miner_utilities/miner_utils.py b/services/miner_utilities/miner_utils.py index ca565252..c48cd47a 100644 --- a/services/miner_utilities/miner_utils.py +++ b/services/miner_utilities/miner_utils.py @@ -88,16 +88,31 @@ async def video_upscaler(payload_url: str, task_type: str) -> str | None: logger.error(f"Upscaling service error: {response.status}") return None -async def video_compressor(payload_url: str, vmaf_threshold: float) -> str | None: +async def video_compressor(payload_url: str, vmaf_threshold: float, target_codec: str = "av1", + codec_mode: str = "CRF", target_bitrate: float = 10.0) -> str | None: """ Sends a video file path to the compression service and retrieves the processed video path. + + Args: + payload_url (str): The URL of the video to be compressed. + vmaf_threshold (float): The VMAF threshold for quality control. + target_codec (str): The target codec for compression (default: "av1"). + codec_mode (str): Codec mode - CBR, VBR, or CRF (default: "CRF"). + target_bitrate (float): Target bitrate in Mbps (default: 10.0). + + Returns: + str | None: The URL of the compressed video or None if an error occurs. """ url = f"http://{CONFIG.video_compressor.host}:{CONFIG.video_compressor.port}/compress-video" headers = {"Content-Type": "application/json"} data = { "payload_url": payload_url, "vmaf_threshold": vmaf_threshold, + "target_codec": target_codec, + "codec_mode": codec_mode, + "target_bitrate": target_bitrate, } + logger.info(f"šŸŽ¬ Sending compression request: VMAF={vmaf_threshold}, Codec={target_codec}, Mode={codec_mode}, Bitrate={target_bitrate} Mbps") async with aiohttp.ClientSession() as session: async with session.post(url, headers=headers, data=json.dumps(data)) as response: if response.status == 200: diff --git a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py index 58f8470c..b1bc3008 100644 --- a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py +++ b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py @@ -180,19 +180,23 @@ async def build_synthetic_protocol(self, content_lengths: list[int], version, ro async def build_compression_protocol(self, vmaf_thresholds: List[float], num_miners: int, version, round_id, recent_counts: List[int]) -> Tuple[list[str], list[str], list[str], list[VideoCompressionProtocol]]: """Fetches synthetic video chunks and builds the video compression protocols. - + Args: - vmaf_thresholds: List of VMAF thresholds for compression quality control + vmaf_threshold: VMAF threshold for compression quality control + num_miners: Number of miners to create protocols for version: Version of the protocol to use round_id: Unique identifier for this round - + target_codec: Target codec for compression (default: "av1") + codec_mode: Codec mode - CBR, VBR, or CRF (default: "CRF") + target_bitrate: Target bitrate in Mbps (default: 10.0) + Returns: Tuple containing lists of: - payload URLs - video IDs - uploaded object names - VideoCompressionProtocol instances - + Raises: httpx.HTTPStatusError: If the request to the video scheduler fails RuntimeError: If max retries exceeded without valid response @@ -263,7 +267,8 @@ async def build_compression_protocol(self, vmaf_thresholds: List[float], num_min payload_urls.append(chunk["sharing_link"]) video_ids.append(chunk["video_id"]) uploaded_object_names.append(chunk["uploaded_object_name"]) - + + # Use the provided compression parameters synapse = VideoCompressionProtocol( miner_payload=CompressionMinerPayload( reference_video_url=chunk["sharing_link"], @@ -396,10 +401,18 @@ async def build_organic_compression_protocol(self, needed: int): logger.info(f"Invalid compression type: {chunk['compression_type']}") continue + # Get compression parameters from chunk with defaults + target_codec = chunk.get("target_codec", "av1") + codec_mode = chunk.get("codec_mode", "CRF") + target_bitrate = chunk.get("target_bitrate", 10.0) + synapse = VideoCompressionProtocol( miner_payload=CompressionMinerPayload( reference_video_url=chunk["url"], - vmaf_threshold=vmaf_threshold + vmaf_threshold=vmaf_threshold, + target_codec=target_codec, + codec_mode=codec_mode, + target_bitrate=target_bitrate ), ) task_ids.append(chunk["task_id"])