From 85c2fa64a2295611f9dd7ee6ae62c77a6e0a940b Mon Sep 17 00:00:00 2001 From: Ahmad Ayad Date: Wed, 5 Nov 2025 21:57:45 +0100 Subject: [PATCH 01/10] added passing for target codec for both synthetic jobs and organics --- neurons/validator.py | 18 ++++++++++---- services/organic_gateway/models.py | 2 ++ services/organic_gateway/routes.py | 11 +++++---- services/organic_gateway/services.py | 10 ++++---- vidaio_subnet_core/protocol.py | 4 ++++ .../synthesizing/challenge_synthesizer.py | 24 ++++++++++++------- 6 files changed, 49 insertions(+), 20 deletions(-) diff --git a/neurons/validator.py b/neurons/validator.py index 2ad36843..0a76067f 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -34,6 +34,15 @@ 95, #High ] +TARGET_CODECS = [ + "av1_nvenc", # AV1 (NVIDIA GPU) + # "hevc_nvenc", # H.265 (NVIDIA GPU) + # "h264_nvenc", # H.264 (NVIDIA GPU) + # "libsvtav1", # AV1 (CPU - SVT-AV1) + # "libx265", # H.265 (CPU) + # "libx264", # H.264 (CPU) +] + SLEEP_TIME_LOW = 60 * 3 # 3 minutes SLEEP_TIME_HIGH = 60 * 4 # 4 minutes @@ -440,14 +449,15 @@ async def process_compression_miners(self, compression_miners, version): axons.append(miner[0]) vmaf_threshold = random.choice(VMAF_QUALITY_THRESHOLDS) - + target_codec = random.choice(TARGET_CODECS) + round_id = str(uuid.uuid4()) num_miners = len(uids) - payload_urls, video_ids, uploaded_object_names, synapses = await self.challenge_synthesizer.build_compression_protocol(vmaf_threshold, num_miners, version, round_id) - logger.debug(f"Built compression challenge protocol") - + payload_urls, video_ids, uploaded_object_names, synapses = await self.challenge_synthesizer.build_compression_protocol(vmaf_threshold, num_miners, version, round_id, target_codec) + logger.warning(f"Built compression challenge protocol with VMAF threshold {vmaf_threshold} and codec {target_codec}") + timestamp = datetime.now(timezone.utc).isoformat() logger.debug(f"Processing compression UIDs in batch: {uids}") diff --git a/services/organic_gateway/models.py b/services/organic_gateway/models.py index 216c18f2..f35e01b9 100644 --- a/services/organic_gateway/models.py +++ b/services/organic_gateway/models.py @@ -29,6 +29,7 @@ class CompressionRequest(BaseModel): chunk_id: str chunk_url: str compression_type: Literal["High", "Medium", "Low"] + target_codec: Optional[str] = "av1_nvenc" # Default to av1_nvenc for best compression class CompressionResponse(BaseModel): task_id: str @@ -63,6 +64,7 @@ class InsertOrganicCompressionRequest(BaseModel): chunk_id: str task_id: str compression_type: Literal["High", "Medium", "Low"] + target_codec: Optional[str] = "av1_nvenc" class InsertResultRequest(BaseModel): processed_video_url: str diff --git a/services/organic_gateway/routes.py b/services/organic_gateway/routes.py index c00a629b..75faf85e 100644 --- a/services/organic_gateway/routes.py +++ b/services/organic_gateway/routes.py @@ -39,7 +39,8 @@ async def upscale( chunk_id=request.chunk_id, chunk_url=request.chunk_url, resolution_type=request.resolution_type, - compression_type=None + compression_type=None, + target_codec=None ) # Submit to Redis service in background to avoid blocking @@ -93,16 +94,18 @@ async def compression( chunk_id=request.chunk_id, chunk_url=request.chunk_url, resolution_type=None, - compression_type=request.compression_type + compression_type=request.compression_type, + target_codec=request.target_codec ) - + # Submit to Redis service in background to avoid blocking background_tasks.add_task( redis_service.insert_organic_compression_chunk, url=request.chunk_url, chunk_id=request.chunk_id, task_id=task_id, - compression_type=request.compression_type + compression_type=request.compression_type, + target_codec=request.target_codec ) return CompressionResponse( diff --git a/services/organic_gateway/services.py b/services/organic_gateway/services.py index 3ea79919..0f65be4b 100644 --- a/services/organic_gateway/services.py +++ b/services/organic_gateway/services.py @@ -25,7 +25,7 @@ class TaskService: def __init__(self, redis_conn): self.redis = redis_conn - def create_task(self, task_id: str, chunk_id: str, chunk_url: str, resolution_type: Optional[str], compression_type: Optional[str]): + def create_task(self, task_id: str, chunk_id: str, chunk_url: str, resolution_type: Optional[str], compression_type: Optional[str], target_codec: Optional[str] = None): """Create a new task and store in Redis""" now = datetime.utcnow().isoformat() task_data = { @@ -34,6 +34,7 @@ def create_task(self, task_id: str, chunk_id: str, chunk_url: str, resolution_ty "chunk_url": chunk_url, "resolution_type": resolution_type or "", "compression_type": compression_type or "", + "target_codec": target_codec or "", "status": TaskStatus.QUEUED, "created_at": now, "updated_at": now @@ -148,16 +149,17 @@ async def insert_organic_upscaling_chunk(self, url: str, chunk_id: str, task_id: return await self._make_request("POST", api_url, payload.dict()) - async def insert_organic_compression_chunk(self, url: str, chunk_id: str, task_id: str, compression_type: str): + async def insert_organic_compression_chunk(self, url: str, chunk_id: str, task_id: str, compression_type: str, target_codec: str = "av1_nvenc"): """Insert chunk into organic compression queue via Redis service""" api_url = f"{self.endpoint}/api/insert_organic_compression_chunk" payload = InsertOrganicCompressionRequest( url=url, chunk_id=chunk_id, task_id=task_id, - compression_type=compression_type + compression_type=compression_type, + target_codec=target_codec ) - + return await self._make_request("POST", api_url, payload.dict()) async def get_result(self, original_video_url: str): diff --git a/vidaio_subnet_core/protocol.py b/vidaio_subnet_core/protocol.py index b3555dcf..4ad40a26 100644 --- a/vidaio_subnet_core/protocol.py +++ b/vidaio_subnet_core/protocol.py @@ -57,6 +57,10 @@ class CompressionMinerPayload(BaseModel): ge=0.0, le=100.0, ) + target_codec: str = Field( + description="The target codec for compression (e.g., H264, H265, VP9, AV1)", + default="av1_nvenc", + ) class MinerResponse(BaseModel): diff --git a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py index 40e11b89..0231186a 100644 --- a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py +++ b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py @@ -172,21 +172,23 @@ async def build_synthetic_protocol(self, content_lengths: list[int], version, ro raise RuntimeError(f"Failed to get synthetic chunks after {self.max_retries} attempts") - async def build_compression_protocol(self, vmaf_threshold: float, num_miners: int, version, round_id) -> Tuple[list[str], list[str], list[str], list[VideoCompressionProtocol]]: + async def build_compression_protocol(self, vmaf_threshold: float, num_miners: int, version, round_id, target_codec: str = "av1_nvenc") -> 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_nvenc") + 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 @@ -256,11 +258,13 @@ async def build_compression_protocol(self, vmaf_threshold: float, num_miners: in payload_urls.append(chunk["sharing_link"]) video_ids.append(chunk["video_id"]) uploaded_object_names.append(chunk["uploaded_object_name"]) - + + # Use the provided target_codec parameter synapse = VideoCompressionProtocol( miner_payload=CompressionMinerPayload( reference_video_url=chunk["sharing_link"], - vmaf_threshold=vmaf_threshold + vmaf_threshold=vmaf_threshold, + target_codec=target_codec ), version=version, round_id=round_id @@ -389,10 +393,14 @@ async def build_organic_compression_protocol(self, needed: int): logger.info(f"Invalid compression type: {chunk['compression_type']}") continue + # Get target_codec from chunk, default to av1_nvenc if not present + target_codec = chunk.get("target_codec", "av1_nvenc") + synapse = VideoCompressionProtocol( miner_payload=CompressionMinerPayload( reference_video_url=chunk["url"], - vmaf_threshold=vmaf_threshold + vmaf_threshold=vmaf_threshold, + target_codec=target_codec ), ) task_ids.append(chunk["task_id"]) From 2d485776eea504a02737e9ff66fa25f857b2b793 Mon Sep 17 00:00:00 2001 From: Ahmad Ayad Date: Tue, 11 Nov 2025 01:53:53 +0100 Subject: [PATCH 02/10] added miner logic to handle target codec --- .gitignore | 3 ++- neurons/miner.py | 21 ++++++++++--------- neurons/validator.py | 4 ++-- services/miner_utilities/miner_utils.py | 12 ++++++++++- .../synthesizing/challenge_synthesizer.py | 13 +++++++----- 5 files changed, 34 insertions(+), 19 deletions(-) diff --git a/.gitignore b/.gitignore index 8e09472b..19a6a45f 100644 --- a/.gitignore +++ b/.gitignore @@ -29,4 +29,5 @@ tmp/ vmaf/ services/scoring/vmaf services/upscaling/models/ -test.py \ No newline at end of file +test.py +ignore/ \ No newline at end of file diff --git a/neurons/miner.py b/neurons/miner.py index 3428f920..5c77d7a5 100644 --- a/neurons/miner.py +++ b/neurons/miner.py @@ -62,30 +62,31 @@ 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 = synapse.miner_payload.target_codec 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} šŸ›œšŸ›œšŸ›œ") 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) + 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/neurons/validator.py b/neurons/validator.py index 0a76067f..1c945011 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -454,10 +454,10 @@ async def process_compression_miners(self, compression_miners, version): round_id = str(uuid.uuid4()) num_miners = len(uids) - + payload_urls, video_ids, uploaded_object_names, synapses = await self.challenge_synthesizer.build_compression_protocol(vmaf_threshold, num_miners, version, round_id, target_codec) logger.warning(f"Built compression challenge protocol with VMAF threshold {vmaf_threshold} and codec {target_codec}") - + timestamp = datetime.now(timezone.utc).isoformat() logger.debug(f"Processing compression UIDs in batch: {uids}") diff --git a/services/miner_utilities/miner_utils.py b/services/miner_utilities/miner_utils.py index ca565252..bd5c7582 100644 --- a/services/miner_utilities/miner_utils.py +++ b/services/miner_utilities/miner_utils.py @@ -88,16 +88,26 @@ 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_nvenc") -> 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_nvenc"). + + 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, } + logger.info(f"šŸŽ¬ Sending compression request: VMAF={vmaf_threshold}, Codec={target_codec}") 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 0231186a..29d44841 100644 --- a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py +++ b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py @@ -172,23 +172,24 @@ async def build_synthetic_protocol(self, content_lengths: list[int], version, ro raise RuntimeError(f"Failed to get synthetic chunks after {self.max_retries} attempts") - async def build_compression_protocol(self, vmaf_threshold: float, num_miners: int, version, round_id, target_codec: str = "av1_nvenc") -> Tuple[list[str], list[str], list[str], list[VideoCompressionProtocol]]: + async def build_compression_protocol(self, vmaf_threshold: float, num_miners: int, version, round_id, + target_codec: str = "av1_nvenc") -> Tuple[list[str], list[str], list[str], list[VideoCompressionProtocol]]: """Fetches synthetic video chunks and builds the video compression protocols. - + Args: - vmaf_threshold: VMAF threshold for compression quality control + vmaf_thresholds: List of VMAF thresholds 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_nvenc") - + 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 @@ -201,6 +202,8 @@ async def build_compression_protocol(self, vmaf_threshold: float, num_miners: in num_needed = (num_protocols + miners_per_task - 1) // miners_per_task # Ceiling division logger.info(f"Vmaf_threshold: {vmaf_threshold}") + + logger.info(f"Using {miners_per_task} miners per task") logger.info(f"Optimized chunk request: {num_needed} chunks (reduced from {num_protocols} protocols)") From 9baa2ea153c9607f28a2aa3748e4ea4d5ff2aeda Mon Sep 17 00:00:00 2001 From: Ahmad Ayad Date: Fri, 14 Nov 2025 12:33:43 +0100 Subject: [PATCH 03/10] now validator received codec info, codec mode, and target bit rate and pass it to miners --- neurons/miner.py | 6 +- neurons/validator.py | 81 ++++++--- services/miner_utilities/miner_utils.py | 11 +- services/organic_gateway/models.py | 8 +- services/organic_gateway/routes.py | 8 +- services/organic_gateway/services.py | 10 +- services/scoring/server.py | 167 +++++++++++++----- vidaio_subnet_core/protocol.py | 13 +- .../synthesizing/challenge_synthesizer.py | 30 ++-- 9 files changed, 239 insertions(+), 95 deletions(-) diff --git a/neurons/miner.py b/neurons/miner.py index 5c77d7a5..633a37e1 100644 --- a/neurons/miner.py +++ b/neurons/miner.py @@ -68,14 +68,16 @@ async def forward_compression_requests(self, synapse: VideoCompressionProtocol) payload_url: str = synapse.miner_payload.reference_video_url vmaf_threshold: float = synapse.miner_payload.vmaf_threshold target_codec: str = synapse.miner_payload.target_codec + codec_mode: str = synapse.miner_payload.codec_mode + target_bitrate: float = synapse.miner_payload.target_bitrate validator_uid: int = self.metagraph.hotkeys.index(synapse.dendrite.hotkey) - logger.info(f"šŸ›œšŸ›œšŸ›œ Receiving CompressionRequest from validator: {synapse.dendrite.hotkey} with uid: {validator_uid} | VMAF: {vmaf_threshold} | Codec: {target_codec} šŸ›œšŸ›œšŸ›œ") + 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, target_codec) + 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 šŸ’”") diff --git a/neurons/validator.py b/neurons/validator.py index 1c945011..4675f20b 100644 --- a/neurons/validator.py +++ b/neurons/validator.py @@ -35,12 +35,24 @@ ] TARGET_CODECS = [ - "av1_nvenc", # AV1 (NVIDIA GPU) - # "hevc_nvenc", # H.265 (NVIDIA GPU) - # "h264_nvenc", # H.264 (NVIDIA GPU) - # "libsvtav1", # AV1 (CPU - SVT-AV1) - # "libx265", # H.265 (CPU) - # "libx264", # H.264 (CPU) + "av1", # AV1 + # "hevc", # H.265/HEVC (ffprobe returns "hevc", not "h265") + # "h264", # H.264/AVC + # "vp9", # VP9 +] + +CODEC_MODES = [ + "CRF", # Constant Rate Factor (default, quality-based) + # "CBR", # Constant Bitrate + # "VBR", # Variable Bitrate +] + +TARGET_BITRATES = [ + # 2.0, # 2 Mbps / SD + # 3.0, # 3 Mbps / HD + # 6.0, # 6 Mbps / FHD + 10.0, # 10 Mbps / QHD (default) + # 25.0, # 25 Mbps / 4K UHD ] SLEEP_TIME_LOW = 60 * 3 # 3 minutes @@ -435,7 +447,7 @@ async def process_compression_miners(self, compression_miners, version): batch_size = CONFIG.bandwidth.requests_per_synthetic_interval miner_batches = await self.create_miner_batches(compression_miners, batch_size, task_type="compression") - + logger.info(f"Created {len(miner_batches)} compression batches of size {batch_size}") for batch_idx, batch in enumerate(miner_batches): @@ -450,13 +462,17 @@ async def process_compression_miners(self, compression_miners, version): vmaf_threshold = random.choice(VMAF_QUALITY_THRESHOLDS) target_codec = random.choice(TARGET_CODECS) + codec_mode = random.choice(CODEC_MODES) + target_bitrate = random.choice(TARGET_BITRATES) round_id = str(uuid.uuid4()) num_miners = len(uids) - - payload_urls, video_ids, uploaded_object_names, synapses = await self.challenge_synthesizer.build_compression_protocol(vmaf_threshold, num_miners, version, round_id, target_codec) - logger.warning(f"Built compression challenge protocol with VMAF threshold {vmaf_threshold} and codec {target_codec}") + + payload_urls, video_ids, uploaded_object_names, synapses = await self.challenge_synthesizer.build_compression_protocol( + vmaf_threshold, num_miners, version, round_id, target_codec, codec_mode, target_bitrate + ) + logger.warning(f"Built compression challenge protocol with VMAF threshold {vmaf_threshold}, codec {target_codec}, mode {codec_mode}, bitrate {target_bitrate} Mbps") timestamp = datetime.now(timezone.utc).isoformat() @@ -478,7 +494,7 @@ async def process_compression_miners(self, compression_miners, version): logger.warning(f"āš ļø Reference video file missing for video_id {video_id}: {reference_video_path}") reference_video_paths.append(reference_video_path) - asyncio.create_task(self.score_compressions(uids, responses, payload_urls, reference_video_paths, timestamp, video_ids, uploaded_object_names, vmaf_threshold, round_id)) + asyncio.create_task(self.score_compressions(uids, responses, payload_urls, reference_video_paths, timestamp, video_ids, uploaded_object_names, vmaf_threshold, target_codec, codec_mode, target_bitrate, round_id)) batch_processed_time = time.time() - batch_start_time sleep_time = random.uniform(SLEEP_TIME_LOW, SLEEP_TIME_HIGH) - batch_processed_time @@ -632,15 +648,18 @@ async def score_upscalings( logger.info("Failed to send data to dashboard") async def score_compressions( - self, - uids: list[int], - responses: list[protocol.Synapse], - payload_urls: list[str], - reference_video_paths: list[str], - timestamp: str, - video_ids: list[str], - uploaded_object_names: list[str], - vmaf_threshold: float, + self, + uids: list[int], + responses: list[protocol.Synapse], + payload_urls: list[str], + reference_video_paths: list[str], + timestamp: str, + video_ids: list[str], + uploaded_object_names: list[str], + vmaf_threshold: float, + target_codec: str, + codec_mode: str, + target_bitrate: float, round_id: str ): distorted_urls = [] @@ -655,7 +674,10 @@ async def score_compressions( "reference_paths": reference_video_paths, "video_ids": video_ids, "uploaded_object_names": uploaded_object_names, - "vmaf_threshold": vmaf_threshold + "vmaf_threshold": vmaf_threshold, + "target_codec": target_codec, + "codec_mode": codec_mode, + "target_bitrate": target_bitrate }, timeout=240 ) @@ -815,21 +837,22 @@ async def score_organics_upscaling(self, uids: list[int], responses: list[protoc else: logger.info("Failed to send data to dashboard") - async def score_organics_compression(self, uids: list[int], responses: list[protocol.Synapse], reference_urls: list[str], vmaf_thresholds: list[float], timestamp: str): + async def score_organics_compression(self, uids: list[int], responses: list[protocol.Synapse], reference_urls: list[str], vmaf_thresholds: list[float], target_codecs: list[str], timestamp: str): """Score organic compression tasks.""" distorted_urls = [response.miner_response.optimized_video_url for response in responses] - combined = list(zip(uids, distorted_urls, reference_urls, vmaf_thresholds)) + combined = list(zip(uids, distorted_urls, reference_urls, vmaf_thresholds, target_codecs)) random.shuffle(combined) - uids, distorted_urls, reference_urls, vmaf_thresholds = map(list, zip(*combined)) + uids, distorted_urls, reference_urls, vmaf_thresholds, target_codecs = map(list, zip(*combined)) num_pairs_to_validate = min(5, len(combined)) selected_indices = random.sample(range(len(combined)), num_pairs_to_validate) - + selected_uids = [uids[i] for i in selected_indices] selected_distorted_urls = [distorted_urls[i] for i in selected_indices] selected_reference_urls = [reference_urls[i] for i in selected_indices] selected_vmaf_thresholds = [vmaf_thresholds[i] for i in selected_indices] + selected_target_codecs = [target_codecs[i] for i in selected_indices] logger.info(f"Randomly selected {len(selected_uids)} pairs out of {len(uids)} total pairs for compression validation") @@ -839,7 +862,8 @@ async def score_organics_compression(self, uids: list[int], responses: list[prot "uids": selected_uids, "distorted_urls": selected_distorted_urls, "reference_urls": selected_reference_urls, - "vmaf_thresholds": selected_vmaf_thresholds + "vmaf_thresholds": selected_vmaf_thresholds, + "target_codecs": selected_target_codecs }, timeout=115 ) @@ -1021,12 +1045,15 @@ async def process_organic_compression_chunks(self, num_organic_chunks): responses = [response[0] for response in raw_responses] processed_urls = [response.miner_response.optimized_video_url for response in responses] + # Extract target_codecs from synapses + target_codecs = [synapse.miner_payload.target_codec for synapse in synapses] + logger.info("Updating task status to 'completed' and pushing results for compression") for task_id, original_url, processed_url in zip(task_ids, original_urls, processed_urls): await self.update_task_status(task_id, original_url, "completed") await self.push_result(task_id, original_url, processed_url) - asyncio.create_task(self.score_organics_compression(forward_uids.tolist(), responses, original_urls, vmaf_thresholds, timestamp)) + asyncio.create_task(self.score_organics_compression(forward_uids.tolist(), responses, original_urls, vmaf_thresholds, target_codecs, timestamp)) end_time = time.time() total_time = end_time - organic_start_time diff --git a/services/miner_utilities/miner_utils.py b/services/miner_utilities/miner_utils.py index bd5c7582..c48cd47a 100644 --- a/services/miner_utilities/miner_utils.py +++ b/services/miner_utilities/miner_utils.py @@ -88,14 +88,17 @@ 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, target_codec: str = "av1_nvenc") -> 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_nvenc"). + 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. @@ -106,8 +109,10 @@ async def video_compressor(payload_url: str, vmaf_threshold: float, target_codec "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}") + 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/services/organic_gateway/models.py b/services/organic_gateway/models.py index f35e01b9..69cf2b52 100644 --- a/services/organic_gateway/models.py +++ b/services/organic_gateway/models.py @@ -29,7 +29,9 @@ class CompressionRequest(BaseModel): chunk_id: str chunk_url: str compression_type: Literal["High", "Medium", "Low"] - target_codec: Optional[str] = "av1_nvenc" # Default to av1_nvenc for best compression + target_codec: Optional[str] = "av1" # Default to av1 for best compression + codec_mode: Optional[str] = "CRF" # CBR, VBR, or CRF (default) + target_bitrate: Optional[float] = 10.0 # Target bitrate in Mbps class CompressionResponse(BaseModel): task_id: str @@ -64,7 +66,9 @@ class InsertOrganicCompressionRequest(BaseModel): chunk_id: str task_id: str compression_type: Literal["High", "Medium", "Low"] - target_codec: Optional[str] = "av1_nvenc" + target_codec: Optional[str] = "av1" + codec_mode: Optional[str] = "CRF" + target_bitrate: Optional[float] = 10.0 class InsertResultRequest(BaseModel): processed_video_url: str diff --git a/services/organic_gateway/routes.py b/services/organic_gateway/routes.py index 75faf85e..d916e218 100644 --- a/services/organic_gateway/routes.py +++ b/services/organic_gateway/routes.py @@ -95,7 +95,9 @@ async def compression( chunk_url=request.chunk_url, resolution_type=None, compression_type=request.compression_type, - target_codec=request.target_codec + target_codec=request.target_codec, + codec_mode=request.codec_mode, + target_bitrate=request.target_bitrate ) # Submit to Redis service in background to avoid blocking @@ -105,7 +107,9 @@ async def compression( chunk_id=request.chunk_id, task_id=task_id, compression_type=request.compression_type, - target_codec=request.target_codec + target_codec=request.target_codec, + codec_mode=request.codec_mode, + target_bitrate=request.target_bitrate ) return CompressionResponse( diff --git a/services/organic_gateway/services.py b/services/organic_gateway/services.py index 0f65be4b..a1df5c0c 100644 --- a/services/organic_gateway/services.py +++ b/services/organic_gateway/services.py @@ -25,7 +25,7 @@ class TaskService: def __init__(self, redis_conn): self.redis = redis_conn - def create_task(self, task_id: str, chunk_id: str, chunk_url: str, resolution_type: Optional[str], compression_type: Optional[str], target_codec: Optional[str] = None): + def create_task(self, task_id: str, chunk_id: str, chunk_url: str, resolution_type: Optional[str], compression_type: Optional[str], target_codec: Optional[str] = None, codec_mode: Optional[str] = None, target_bitrate: Optional[float] = None): """Create a new task and store in Redis""" now = datetime.utcnow().isoformat() task_data = { @@ -35,6 +35,8 @@ def create_task(self, task_id: str, chunk_id: str, chunk_url: str, resolution_ty "resolution_type": resolution_type or "", "compression_type": compression_type or "", "target_codec": target_codec or "", + "codec_mode": codec_mode or "", + "target_bitrate": str(target_bitrate) if target_bitrate is not None else "", "status": TaskStatus.QUEUED, "created_at": now, "updated_at": now @@ -149,7 +151,7 @@ async def insert_organic_upscaling_chunk(self, url: str, chunk_id: str, task_id: return await self._make_request("POST", api_url, payload.dict()) - async def insert_organic_compression_chunk(self, url: str, chunk_id: str, task_id: str, compression_type: str, target_codec: str = "av1_nvenc"): + async def insert_organic_compression_chunk(self, url: str, chunk_id: str, task_id: str, compression_type: str, target_codec: str = "av1", codec_mode: str = "CRF", target_bitrate: float = 10.0): """Insert chunk into organic compression queue via Redis service""" api_url = f"{self.endpoint}/api/insert_organic_compression_chunk" payload = InsertOrganicCompressionRequest( @@ -157,7 +159,9 @@ async def insert_organic_compression_chunk(self, url: str, chunk_id: str, task_i chunk_id=chunk_id, task_id=task_id, compression_type=compression_type, - target_codec=target_codec + target_codec=target_codec, + codec_mode=codec_mode, + target_bitrate=target_bitrate ) return await self._make_request("POST", api_url, payload.dict()) diff --git a/services/scoring/server.py b/services/scoring/server.py index 4c9f4aa8..478099f4 100644 --- a/services/scoring/server.py +++ b/services/scoring/server.py @@ -66,6 +66,9 @@ class CompressionScoringRequest(BaseModel): video_ids: List[str] uploaded_object_names: List[str] vmaf_threshold: float + target_codec: Optional[str] = "av1" # Target codec family (av1, h264, hevc, vp9, etc.) + codec_mode: Optional[str] = "CRF" # Codec mode: CBR, VBR, or CRF + target_bitrate: Optional[float] = 10.0 # Target bitrate in Mbps fps: Optional[float] = None subsample: Optional[int] = 1 verbose: Optional[bool] = False @@ -91,6 +94,9 @@ class OrganicsCompressionScoringRequest(BaseModel): distorted_urls: List[str] reference_urls: List[str] vmaf_thresholds: List[float] + target_codecs: Optional[List[str]] = None # List of target codecs (one per video) + codec_modes: Optional[List[str]] = None # List of codec modes (one per video) + target_bitrates: Optional[List[float]] = None # List of target bitrates (one per video) uids: List[int] fps: Optional[float] = None subsample: Optional[int] = 1 @@ -304,33 +310,42 @@ def validate_color_channels_on_frames(frames): """ Validate that frames have color information (not grayscale). Reuses frames already extracted for VMAF calculation. - + Args: frames: List of frames (BGR numpy arrays from cv2) - + Returns: tuple: (is_valid, reason) """ try: if not frames: return False, "No frames available for color validation" - + color_threshold = 5.0 # Same threshold as before - + brightness_threshold = 20.0 # Threshold for detecting black/near-black frames + for i, frame in enumerate(frames): # Convert BGR to RGB img_array = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) - + # Extract RGB channels r_channel = img_array[:,:,0].astype(float) g_channel = img_array[:,:,1].astype(float) b_channel = img_array[:,:,2].astype(float) - + + # Check if frame is black/near-black (low brightness) + avg_brightness = np.mean([r_channel.mean(), g_channel.mean(), b_channel.mean()]) + + # Skip validation for black/near-black frames (valid content) + if avg_brightness < brightness_threshold: + logger.debug(f"Frame {i} is black/near-black (brightness={avg_brightness:.2f}), skipping color validation") + continue + # Calculate average difference between channels rg_diff = np.mean(np.abs(r_channel - g_channel)) rb_diff = np.mean(np.abs(r_channel - b_channel)) gb_diff = np.mean(np.abs(g_channel - b_channel)) - + # If all channels are nearly identical, it's grayscale if rg_diff < color_threshold and rb_diff < color_threshold and gb_diff < color_threshold: logger.warning( @@ -338,9 +353,9 @@ def validate_color_channels_on_frames(frames): f"RG diff={rg_diff:.2f}, RB diff={rb_diff:.2f}, GB diff={gb_diff:.2f}" ) return False, f"Video has no color information (grayscale). UV channels required. Frame {i} detected as grayscale." - + return True, "Color channels validated" - + except Exception as e: logger.error(f"Error validating color channels: {e}") return False, f"Error validating color: {str(e)}" @@ -349,62 +364,69 @@ def validate_chroma_quality_on_frames(ref_frames, dist_frames, threshold=0.7): """ Validate chroma (UV) quality by comparing reference and distorted frames. Reuses frames already extracted for VMAF calculation. - + Args: ref_frames: List of reference frames (BGR numpy arrays) dist_frames: List of distorted frames (BGR numpy arrays) threshold: Minimum acceptable chroma similarity (0.0-1.0) - + Returns: tuple: (is_valid, reason) """ try: if len(ref_frames) != len(dist_frames): return False, "Frame count mismatch between reference and distorted" - + if not ref_frames or not dist_frames: return False, "No frames available for chroma validation" - + chroma_ratios = [] - + low_energy_threshold = 1.0 # Minimum chroma energy to consider (filters black/flat frames) + for i, (ref_frame, dist_frame) in enumerate(zip(ref_frames, dist_frames)): # Convert BGR to YUV ref_yuv = cv2.cvtColor(ref_frame, cv2.COLOR_BGR2YUV) dist_yuv = cv2.cvtColor(dist_frame, cv2.COLOR_BGR2YUV) - + # Extract U and V channels ref_u = ref_yuv[:, :, 1].astype(float) ref_v = ref_yuv[:, :, 2].astype(float) dist_u = dist_yuv[:, :, 1].astype(float) dist_v = dist_yuv[:, :, 2].astype(float) - + # Calculate chroma variance/energy ref_u_variance = np.var(ref_u) ref_v_variance = np.var(ref_v) dist_u_variance = np.var(dist_u) dist_v_variance = np.var(dist_v) - + ref_chroma_energy = ref_u_variance + ref_v_variance dist_chroma_energy = dist_u_variance + dist_v_variance - - if ref_chroma_energy > 0: - chroma_ratio = dist_chroma_energy / ref_chroma_energy - chroma_ratios.append(chroma_ratio) - logger.debug(f"Frame {i} chroma ratio: {chroma_ratio:.3f}") - + + # Skip frames with very low chroma energy (black/flat frames - valid content) + if ref_chroma_energy < low_energy_threshold: + logger.debug(f"Frame {i} has low chroma energy ({ref_chroma_energy:.3f}), skipping") + continue + + chroma_ratio = dist_chroma_energy / ref_chroma_energy + chroma_ratios.append(chroma_ratio) + logger.debug(f"Frame {i} chroma ratio: {chroma_ratio:.3f}") + + # If all frames are low-energy (e.g., all black), consider it valid if not chroma_ratios: - return False, "Could not calculate chroma ratios" - + logger.info("All frames have low chroma energy (e.g., black frames), skipping chroma validation") + return True, "Chroma validation skipped (low-energy content)" + # Calculate average chroma ratio avg_chroma_ratio = np.mean(chroma_ratios) - + logger.info(f"Average chroma quality ratio: {avg_chroma_ratio:.3f}, Threshold: {threshold}") - + if avg_chroma_ratio < threshold: return False, f"Chroma quality too low: {avg_chroma_ratio:.3f} < {threshold} (UV channels reduced/degraded)" - + return True, f"Chroma quality validated: {avg_chroma_ratio:.3f}" - + except Exception as e: logger.error(f"Error validating chroma quality: {e}") return False, f"Error validating chroma: {str(e)}" @@ -742,18 +764,54 @@ def is_valid_video(video_path): logger.error(f"Unexpected error validating {video_path}: {e}") return False -def validate_dist_encoding_settings(dist_path: str, ref_path: str, task: str): +def normalize_codec_family(codec_name: str) -> str: + """ + Normalize codec name to its family for comparison. + Returns standard ffprobe codec names to match video metadata. + + Args: + codec_name: The codec name (e.g., "av1", "hevc", "h265", "h264") + + Returns: + Normalized codec family name matching ffprobe output (e.g., "av1", "hevc", "h264", "vp9") + """ + codec_lower = codec_name.lower().strip() + + # AV1 variants + if any(variant in codec_lower for variant in ["av1", "libaom", "libsvtav1", "svt-av1"]): + return "av1" + + # H.265/HEVC variants (ffprobe returns "hevc", not "h265") + if any(variant in codec_lower for variant in ["hevc", "h265", "x265", "libx265"]): + return "hevc" + + # H.264/AVC variants + if any(variant in codec_lower for variant in ["h264", "avc", "x264", "libx264"]): + return "h264" + + # VP9 variants + if "vp9" in codec_lower or "libvpx-vp9" in codec_lower: + return "vp9" + + # VP8 variants + if "vp8" in codec_lower or "libvpx" in codec_lower: + return "vp8" + + # Return as-is if no match (for future codecs) + return codec_lower + +def validate_dist_encoding_settings(dist_path: str, ref_path: str, task: str, target_codec: str = "av1"): """ Validate that distorted video uses specific encoding settings. """ try: - # Enhanced ffprobe to capture both stream and format encoder tags + # Enhanced ffprobe to capture both stream and format encoder tags, including bitrate cmd = [ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name,profile,level,sample_aspect_ratio,pix_fmt," - "width,height,r_frame_rate,color_space,color_primaries,color_transfer", + "width,height,r_frame_rate,color_space,color_primaries,color_transfer,bit_rate", "-show_entries", "format=tags=encoder", "-show_format", "-of", "json", @@ -777,7 +835,13 @@ def validate_dist_encoding_settings(dist_path: str, ref_path: str, task: str): container = format_info.get("format_name", "") fps_str = video_stream.get("r_frame_rate", "0/1") encoder_tag = "" - + + # Get bitrate (prefer stream-level, fallback to format-level) + bit_rate = video_stream.get("bit_rate") or format_info.get("bit_rate") + if bit_rate: + bit_rate = int(bit_rate) # Convert to integer (bits per second) + bit_rate_mbps = bit_rate / 1_000_000 # Megabits per second + logger.info(f"Distorted video bitrate: {bit_rate_mbps:.2f} Mbps") # Colorspace properties dist_color_space = video_stream.get("color_space", None) dist_color_primaries = video_stream.get("color_primaries", None) @@ -799,14 +863,27 @@ def validate_dist_encoding_settings(dist_path: str, ref_path: str, task: str): errors.append(f"Resolution must be {ref_width}x{ref_height}, got {width}x{height}") # REQUIRED encoding checks - if task == "compression" and codec != "av1" or task == "upscaling" and codec != "hevc": - errors.append(f"Codec must be AV1 for compression & hevc for upscaling, got {codec}") + # Normalize both the detected codec and target codec to family names + detected_codec_family = normalize_codec_family(codec) + expected_codec_family = normalize_codec_family(target_codec) if task == "compression" else "hevc" + + if task == "compression": + if detected_codec_family != expected_codec_family: + errors.append(f"Codec must be {expected_codec_family} (got {codec} which normalizes to {detected_codec_family})") + elif task == "upscaling" and detected_codec_family != "hevc": + errors.append(f"Codec must be hevc for upscaling, got {codec}") if "ivf" in container.lower(): errors.append("Container must be MP4, got IVF (incompatible for concatenation)") if container not in ["mov,mp4,m4a,3gp,3g2,mj2", "mp4", "isom"]: errors.append(f"Container must be proper MP4, got {container}") - if profile != "Main": + + # Profile validation based on codec family + if detected_codec_family == "av1" and profile != "Main": errors.append(f"AV1 profile must be 'Main', got {profile}") + elif detected_codec_family == "hevc" and profile not in ["Main", "Main 10"]: + errors.append(f"HEVC profile must be 'Main' or 'Main 10', got {profile}") + elif detected_codec_family == "h264" and profile not in ["Main", "High", "Baseline"]: + errors.append(f"H.264 profile must be 'Main', 'High', or 'Baseline', got {profile}") if sar != "1:1": errors.append(f"Sample aspect ratio must be 1:1, got {sar}") if pix_fmt != "yuv420p": @@ -1537,11 +1614,12 @@ async def score_compression_synthetics(request: CompressionScoringRequest) -> Co step_time = time.time() - uid_start_time logger.info(f"ā™Žļø 4. Validated distorted video in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.") - # Validate encoding settings (MUST be AV1 in proper MP4 with specific params) - is_valid_encoding, encoding_msg = validate_dist_encoding_settings(dist_path, ref_path, task="compression") + # Validate encoding settings (codec must match target_codec) + target_codec = request.target_codec or "av1" + is_valid_encoding, encoding_msg = validate_dist_encoding_settings(dist_path, ref_path, task="compression", target_codec=target_codec) if not is_valid_encoding: logger.error(f"Invalid encoding settings for distorted video {dist_path}: {encoding_msg}") - logger.info(f" Required: AV1 codec, Main profile, yuv420p, MP4 container, 1:1 SAR") + logger.info(f" Required: {target_codec} codec family, proper profile, yuv420p, MP4 container, 1:1 SAR") vmaf_scores.append(0.0) compression_rates.append(1.0) final_scores.append(0.0) @@ -2104,8 +2182,11 @@ async def score_organics_compression(request: OrganicsCompressionScoringRequest) logger.error(f"failed to download distorted video: {dist_url}, error: {e}") distorted_video_paths.append(None) - for idx, (ref_url, dist_path, uid, vmaf_threshold) in enumerate( - zip(request.reference_urls, distorted_video_paths, request.uids, request.vmaf_thresholds) + # Get target_codecs list, defaulting to "av1" if not provided + target_codecs = request.target_codecs if request.target_codecs else ["av1"] * len(request.uids) + + for idx, (ref_url, dist_path, uid, vmaf_threshold, target_codec) in enumerate( + zip(request.reference_urls, distorted_video_paths, request.uids, request.vmaf_thresholds, target_codecs) ): logger.info(f"šŸ§© processing {uid}.... downloading reference video.... šŸ§©") ref_path = None @@ -2188,10 +2269,10 @@ async def score_organics_compression(request: OrganicsCompressionScoringRequest) final_scores.append(0.0) # 0 = miner penalty continue - is_valid_encoding, encoding_msg = validate_dist_encoding_settings(dist_path, ref_path, task="compression") + is_valid_encoding, encoding_msg = validate_dist_encoding_settings(dist_path, ref_path, task="compression", target_codec=target_codec) if not is_valid_encoding: logger.error(f"Invalid encoding settings for distorted video {dist_path}: {encoding_msg}") - logger.info(f" Required: AV1 codec, Main profile, yuv420p, MP4 container, 1:1 SAR") + logger.info(f" Required: {target_codec} codec family, proper profile, yuv420p, MP4 container, 1:1 SAR") vmaf_scores.append(0.0) compression_rates.append(1.0) final_scores.append(0.0) diff --git a/vidaio_subnet_core/protocol.py b/vidaio_subnet_core/protocol.py index 4ad40a26..1d30910b 100644 --- a/vidaio_subnet_core/protocol.py +++ b/vidaio_subnet_core/protocol.py @@ -58,8 +58,17 @@ class CompressionMinerPayload(BaseModel): le=100.0, ) target_codec: str = Field( - description="The target codec for compression (e.g., H264, H265, VP9, AV1)", - default="av1_nvenc", + description="The target codec for compression (e.g., av1, hevc, h264, vp9)", + default="av1", + ) + codec_mode: str = Field( + description="Codec mode: CBR (Constant Bitrate), VBR (Variable Bitrate), or CRF (Constant Rate Factor)", + default="CRF", + ) + target_bitrate: float = Field( + description="Target bitrate in Mbps (megabits per second)", + default=10.0, + gt=0.0, ) diff --git a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py index 29d44841..c9ffc1e7 100644 --- a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py +++ b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py @@ -173,23 +173,25 @@ async def build_synthetic_protocol(self, content_lengths: list[int], version, ro raise RuntimeError(f"Failed to get synthetic chunks after {self.max_retries} attempts") async def build_compression_protocol(self, vmaf_threshold: float, num_miners: int, version, round_id, - target_codec: str = "av1_nvenc") -> Tuple[list[str], list[str], list[str], list[VideoCompressionProtocol]]: + target_codec: str = "av1", codec_mode: str = "CRF", target_bitrate: float = 10.0) -> 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_nvenc") - + 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 @@ -262,12 +264,14 @@ async def build_compression_protocol(self, vmaf_threshold: float, num_miners: in video_ids.append(chunk["video_id"]) uploaded_object_names.append(chunk["uploaded_object_name"]) - # Use the provided target_codec parameter + # Use the provided compression parameters synapse = VideoCompressionProtocol( miner_payload=CompressionMinerPayload( reference_video_url=chunk["sharing_link"], vmaf_threshold=vmaf_threshold, - target_codec=target_codec + target_codec=target_codec, + codec_mode=codec_mode, + target_bitrate=target_bitrate ), version=version, round_id=round_id @@ -396,14 +400,18 @@ async def build_organic_compression_protocol(self, needed: int): logger.info(f"Invalid compression type: {chunk['compression_type']}") continue - # Get target_codec from chunk, default to av1_nvenc if not present - target_codec = chunk.get("target_codec", "av1_nvenc") + # 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, - target_codec=target_codec + target_codec=target_codec, + codec_mode=codec_mode, + target_bitrate=target_bitrate ), ) task_ids.append(chunk["task_id"]) From d5c15baff7ae28d7cd0300024862512df5bdf667 Mon Sep 17 00:00:00 2001 From: Arpan Tripathi Date: Mon, 17 Nov 2025 11:51:34 +0000 Subject: [PATCH 04/10] add missing payload args --- services/video_scheduler/server.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/services/video_scheduler/server.py b/services/video_scheduler/server.py index 67d13cba..632a56b6 100644 --- a/services/video_scheduler/server.py +++ b/services/video_scheduler/server.py @@ -36,6 +36,9 @@ class InsertOrganicCompressionRequest(BaseModel): chunk_id: str task_id: str compression_type: str + target_codec: Optional[str] = "av1" + codec_mode: Optional[str] = "CRF" + target_bitrate: Optional[float] = 10.0 class InsertResultRequest(BaseModel): processed_video_url: str From b65a105ca364d1e7ebb4e9bdbdc714ed8c5334e6 Mon Sep 17 00:00:00 2001 From: Arpan Tripathi Date: Mon, 17 Nov 2025 12:03:56 +0000 Subject: [PATCH 05/10] debug --- services/video_scheduler/server.py | 5 ++++- .../validating/synthesizing/challenge_synthesizer.py | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/services/video_scheduler/server.py b/services/video_scheduler/server.py index 632a56b6..01ee0522 100644 --- a/services/video_scheduler/server.py +++ b/services/video_scheduler/server.py @@ -80,7 +80,10 @@ def api_insert_organic_compression_chunk(payload: InsertOrganicCompressionReques "url": payload.url, "chunk_id": payload.chunk_id, "task_id": payload.task_id, - "compression_type": payload.compression_type + "compression_type": payload.compression_type, + "target_codec": payload.target_codec, + "codec_mode": payload.codec_mode, + "target_bitrate": payload.target_bitrate } push_organic_compression_chunk(r, data) return {"message": "Organic compression chunk inserted"} diff --git a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py index c9ffc1e7..8ff0aaa2 100644 --- a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py +++ b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py @@ -376,7 +376,7 @@ async def build_organic_compression_protocol(self, needed: int): chunks: List[Dict] = data["chunks"] logger.info("Received organic compression chunks from video-scheduler API") - required_fields = ["url", "chunk_id", "task_id", "compression_type"] + required_fields = ["url", "chunk_id", "task_id", "compression_type", "target_codec", "codec_mode", "target_bitrate"] if any(not all(field in chunk for field in required_fields) for chunk in chunks): logger.info("Missing required fields in some chunk data, retrying...") await asyncio.sleep(self.retry_delay) From 568acb2914a7a05c9db3eaae46d9b3093cf8bba4 Mon Sep 17 00:00:00 2001 From: Arpan Tripathi Date: Mon, 17 Nov 2025 12:06:59 +0000 Subject: [PATCH 06/10] Update challenge_synthesizer.py --- .../validating/synthesizing/challenge_synthesizer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py index 8ff0aaa2..c9ffc1e7 100644 --- a/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py +++ b/vidaio_subnet_core/validating/synthesizing/challenge_synthesizer.py @@ -376,7 +376,7 @@ async def build_organic_compression_protocol(self, needed: int): chunks: List[Dict] = data["chunks"] logger.info("Received organic compression chunks from video-scheduler API") - required_fields = ["url", "chunk_id", "task_id", "compression_type", "target_codec", "codec_mode", "target_bitrate"] + required_fields = ["url", "chunk_id", "task_id", "compression_type"] if any(not all(field in chunk for field in required_fields) for chunk in chunks): logger.info("Missing required fields in some chunk data, retrying...") await asyncio.sleep(self.retry_delay) From 06ed9a6b5e05909825f4a188603f460f7f85fdd3 Mon Sep 17 00:00:00 2001 From: Ahmad Ayad Date: Thu, 20 Nov 2025 20:27:55 +0100 Subject: [PATCH 07/10] added codec_mode and target_bitrate to the miner flow. Also, vase miner now use target_codec, codec_mode, and target_bitrate in their encoding --- .gitignore | 3 +- requirements.txt | 8 +- services/compress/encoder.py | 40 +++- services/compress/server.py | 249 +++++++++++++++++--- services/compress/utils/encode_video.py | 73 +++++- services/compress/utils/processing_utils.py | 4 +- 6 files changed, 326 insertions(+), 51 deletions(-) diff --git a/.gitignore b/.gitignore index 19a6a45f..faa53dbd 100644 --- a/.gitignore +++ b/.gitignore @@ -30,4 +30,5 @@ vmaf/ services/scoring/vmaf services/upscaling/models/ test.py -ignore/ \ No newline at end of file +ignore/ +test_outputs/ \ No newline at end of file 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 c69cffad..16012edc 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 3fa0e7c3..1bffb1f9 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 == 90: + elif vmaf_threshold == VMAF_THRESHOLD_MEDIUM: target_quality = 'Medium' - elif vmaf_threshold == 95: + 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..239837d6 100644 --- a/services/compress/utils/encode_video.py +++ b/services/compress/utils/encode_video.py @@ -132,13 +132,13 @@ 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, max_bit_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 @@ -148,8 +148,10 @@ def encode_video(input_path, output_path, codec, rate=None, max_bit_rate=None, p 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 """ @@ -289,6 +291,69 @@ def encode_video(input_path, output_path, codec, rate=None, max_bit_rate=None, p if logging_enabled: print(f"Applying preset override: {preset}") + # 5.5. Apply codec_mode and target_bitrate if provided + if codec_mode and target_bitrate: + if logging_enabled: + print(f"Applying codec_mode='{codec_mode}' with target_bitrate={target_bitrate} Mbps") + + if codec_mode.upper() == 'CBR': + # Constant Bitrate Mode - fixed bitrate throughout + bitrate_kbps = int(target_bitrate * 1000) # Convert Mbps to kbps + bitrate_value = f"{bitrate_kbps}k" + + current_settings['bitrate'] = bitrate_value + current_settings['maxrate'] = bitrate_value + current_settings['bufsize'] = f"{bitrate_kbps * 2}k" # Buffer = 2x bitrate + + # Remove CRF/CQ parameters for CBR mode (incompatible) + if 'crf' in current_settings: + del current_settings['crf'] + if logging_enabled: + print(f"Removed 'crf' for CBR mode") + if 'cq' in current_settings: + del current_settings['cq'] + if logging_enabled: + print(f"Removed 'cq' for CBR mode") + + # Set rate control mode for NVENC + 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 + 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" + + # Keep CRF/CQ for quality target, but limit with maxrate + # This gives best quality within the bitrate constraint + 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, keeping CQ/CRF for quality") + + elif codec_mode.upper() == 'CRF': + # CRF mode is the default - no changes needed + # This mode prioritizes quality (uses the rate/CQ parameter) + if logging_enabled: + print(f"CRF mode: Using quality-based encoding (rate={rate})") + + else: + if logging_enabled: + print(f"Warning: Unknown codec_mode '{codec_mode}', defaulting to CRF behavior") + # 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 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 ) From 931618600ecc519f1d8540cb457b3b11d191dd8c Mon Sep 17 00:00:00 2001 From: Ahmad Ayad Date: Tue, 25 Nov 2025 22:18:40 +0100 Subject: [PATCH 08/10] cleaned the encode_video file and made it much more readable --- services/compress/utils/encode_video.py | 318 ++++++++++++------------ 1 file changed, 153 insertions(+), 165 deletions(-) diff --git a/services/compress/utils/encode_video.py b/services/compress/utils/encode_video.py index 239837d6..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,7 +249,7 @@ 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, codec_mode=None, target_bitrate=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. @@ -144,7 +261,6 @@ def encode_video(input_path, output_path, codec, rate=None, max_bit_rate=None, p 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) @@ -188,116 +304,16 @@ 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 - - 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 - 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 - - 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}.") - - # 5. Apply preset override if provided - if preset is not None: - current_settings['preset'] = preset - if logging_enabled: - print(f"Applying preset override: {preset}") - - # 5.5. Apply codec_mode and target_bitrate if provided + # 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") 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" @@ -305,17 +321,10 @@ def encode_video(input_path, output_path, codec, rate=None, max_bit_rate=None, p current_settings['maxrate'] = bitrate_value current_settings['bufsize'] = f"{bitrate_kbps * 2}k" # Buffer = 2x bitrate - # Remove CRF/CQ parameters for CBR mode (incompatible) - if 'crf' in current_settings: - del current_settings['crf'] - if logging_enabled: - print(f"Removed 'crf' for CBR mode") - if 'cq' in current_settings: - del current_settings['cq'] - if logging_enabled: - print(f"Removed 'cq' for CBR mode") + # Remove quality parameters (incompatible with CBR) + cleanup_quality_params(current_settings) - # Set rate control mode for NVENC + # Set rate control mode for NVENC/QSV if codec.endswith('_nvenc'): current_settings['rc'] = 'cbr' if logging_enabled: @@ -328,95 +337,74 @@ def encode_video(input_path, output_path, codec, rate=None, max_bit_rate=None, p 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" - # Keep CRF/CQ for quality target, but limit with maxrate - # This gives best quality within the bitrate constraint + # 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, keeping CQ/CRF for quality") + 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 - no changes needed - # This mode prioritizes quality (uses the rate/CQ parameter) + # CRF mode is the default - will apply rate parameter below if logging_enabled: - print(f"CRF mode: Using quality-based encoding (rate={rate})") + print(f"CRF mode: Will apply quality-based encoding (rate={rate})") else: if logging_enabled: print(f"Warning: Unknown codec_mode '{codec_mode}', defaulting to CRF behavior") - # 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.") + # 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}") # 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 From 5dfcdc5b95d4535ffa99ec1623bda7d9e36dbddd Mon Sep 17 00:00:00 2001 From: Ahmad Ayad Date: Mon, 1 Dec 2025 22:38:48 +0100 Subject: [PATCH 09/10] deleted color validation from this branch --- services/scoring/server.py | 224 ------------------------------------- 1 file changed, 224 deletions(-) diff --git a/services/scoring/server.py b/services/scoring/server.py index 4c724280..755dee2b 100644 --- a/services/scoring/server.py +++ b/services/scoring/server.py @@ -306,130 +306,7 @@ def extract_frames_from_y4m(y4m_path): return frames -def validate_color_channels_on_frames(frames): - """ - Validate that frames have color information (not grayscale). - Reuses frames already extracted for VMAF calculation. - - Args: - frames: List of frames (BGR numpy arrays from cv2) - - Returns: - tuple: (is_valid, reason) - """ - try: - if not frames: - return False, "No frames available for color validation" - - color_threshold = 5.0 # Same threshold as before - brightness_threshold = 20.0 # Threshold for detecting black/near-black frames - - for i, frame in enumerate(frames): - # Convert BGR to RGB - img_array = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) - - # Extract RGB channels - r_channel = img_array[:,:,0].astype(float) - g_channel = img_array[:,:,1].astype(float) - b_channel = img_array[:,:,2].astype(float) - - # Check if frame is black/near-black (low brightness) - avg_brightness = np.mean([r_channel.mean(), g_channel.mean(), b_channel.mean()]) - - # Skip validation for black/near-black frames (valid content) - if avg_brightness < brightness_threshold: - logger.debug(f"Frame {i} is black/near-black (brightness={avg_brightness:.2f}), skipping color validation") - continue - - # Calculate average difference between channels - rg_diff = np.mean(np.abs(r_channel - g_channel)) - rb_diff = np.mean(np.abs(r_channel - b_channel)) - gb_diff = np.mean(np.abs(g_channel - b_channel)) - - # If all channels are nearly identical, it's grayscale - if rg_diff < color_threshold and rb_diff < color_threshold and gb_diff < color_threshold: - logger.warning( - f"Frame {i} appears to be grayscale: " - f"RG diff={rg_diff:.2f}, RB diff={rb_diff:.2f}, GB diff={gb_diff:.2f}" - ) - return False, f"Video has no color information (grayscale). UV channels required. Frame {i} detected as grayscale." - - return True, "Color channels validated" - - except Exception as e: - logger.error(f"Error validating color channels: {e}") - return False, f"Error validating color: {str(e)}" - -def validate_chroma_quality_on_frames(ref_frames, dist_frames, threshold=0.7): - """ - Validate chroma (UV) quality by comparing reference and distorted frames. - Reuses frames already extracted for VMAF calculation. - Args: - ref_frames: List of reference frames (BGR numpy arrays) - dist_frames: List of distorted frames (BGR numpy arrays) - threshold: Minimum acceptable chroma similarity (0.0-1.0) - - Returns: - tuple: (is_valid, reason) - """ - try: - if len(ref_frames) != len(dist_frames): - return False, "Frame count mismatch between reference and distorted" - - if not ref_frames or not dist_frames: - return False, "No frames available for chroma validation" - - chroma_ratios = [] - low_energy_threshold = 1.0 # Minimum chroma energy to consider (filters black/flat frames) - - for i, (ref_frame, dist_frame) in enumerate(zip(ref_frames, dist_frames)): - # Convert BGR to YUV - ref_yuv = cv2.cvtColor(ref_frame, cv2.COLOR_BGR2YUV) - dist_yuv = cv2.cvtColor(dist_frame, cv2.COLOR_BGR2YUV) - - # Extract U and V channels - ref_u = ref_yuv[:, :, 1].astype(float) - ref_v = ref_yuv[:, :, 2].astype(float) - dist_u = dist_yuv[:, :, 1].astype(float) - dist_v = dist_yuv[:, :, 2].astype(float) - - # Calculate chroma variance/energy - ref_u_variance = np.var(ref_u) - ref_v_variance = np.var(ref_v) - dist_u_variance = np.var(dist_u) - dist_v_variance = np.var(dist_v) - - ref_chroma_energy = ref_u_variance + ref_v_variance - dist_chroma_energy = dist_u_variance + dist_v_variance - - # Skip frames with very low chroma energy (black/flat frames - valid content) - if ref_chroma_energy < low_energy_threshold: - logger.debug(f"Frame {i} has low chroma energy ({ref_chroma_energy:.3f}), skipping") - continue - - chroma_ratio = dist_chroma_energy / ref_chroma_energy - chroma_ratios.append(chroma_ratio) - logger.debug(f"Frame {i} chroma ratio: {chroma_ratio:.3f}") - - # If all frames are low-energy (e.g., all black), consider it valid - if not chroma_ratios: - logger.info("All frames have low chroma energy (e.g., black frames), skipping chroma validation") - return True, "Chroma validation skipped (low-energy content)" - - # Calculate average chroma ratio - avg_chroma_ratio = np.mean(chroma_ratios) - - logger.info(f"Average chroma quality ratio: {avg_chroma_ratio:.3f}, Threshold: {threshold}") - - if avg_chroma_ratio < threshold: - return False, f"Chroma quality too low: {avg_chroma_ratio:.3f} < {threshold} (UV channels reduced/degraded)" - - return True, f"Chroma quality validated: {avg_chroma_ratio:.3f}" - - except Exception as e: - logger.error(f"Error validating chroma quality: {e}") - return False, f"Error validating chroma: {str(e)}" # Calculate ClipIQA+ inspired score def calculate_clipiqa_plus_score(frames): @@ -1699,53 +1576,7 @@ async def score_compression_synthetics(request: CompressionScoringRequest) -> Co os.unlink(dist_y4m_path) continue - # Now reuse the Y4M files for color validation (no redundant conversion!) - # Extract frames for color validation from Y4M files - logger.info(f"Extracting frames from Y4M for color validation (reusing VMAF Y4M files)") - dist_frames = extract_frames_from_y4m(dist_y4m_path) - step_time = time.time() - uid_start_time - logger.info(f"ā™Žļø 8.5. Extracted {len(dist_frames)} frames from Y4M for color validation in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.") - - # Validate color channels (grayscale check) - color_valid, color_reason = validate_color_channels_on_frames(dist_frames) - if not color_valid: - logger.error(f"UID {uid}: {color_reason}") - vmaf_scores.append(0.0) - compression_rates.append(0.9999) - final_scores.append(0.0) - reasons.append(f"Color validation failed: {color_reason}") - if dist_path and os.path.exists(dist_path): - os.unlink(dist_path) - if dist_y4m_path and os.path.exists(dist_y4m_path): - os.unlink(dist_y4m_path) - continue - - logger.info(f"āœ… UID {uid}: Color channels validated - {color_reason}") - step_time = time.time() - uid_start_time - logger.info(f"ā™Žļø 8.6. Validated color channels in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.") - - # Extract reference frames from Y4M for chroma quality comparison - ref_frames = extract_frames_from_y4m(ref_y4m_path) - # Validate chroma quality (prevents partial UV reduction) - chroma_valid, chroma_reason = validate_chroma_quality_on_frames( - ref_frames, dist_frames, threshold=0.7 - ) - if not chroma_valid: - logger.error(f"UID {uid}: {chroma_reason}") - vmaf_scores.append(0.0) - compression_rates.append(0.9999) - final_scores.append(0.0) - reasons.append(f"Chroma validation failed: {chroma_reason}") - if dist_path and os.path.exists(dist_path): - os.unlink(dist_path) - if dist_y4m_path and os.path.exists(dist_y4m_path): - os.unlink(dist_y4m_path) - continue - - logger.info(f"āœ… UID {uid}: Chroma quality validated - {chroma_reason}") - step_time = time.time() - uid_start_time - logger.info(f"ā™Žļø 8.7. Validated chroma quality in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.") # Calculate compression score using the proper formula # Check scoring function for details @@ -2379,61 +2210,6 @@ async def score_organics_compression(request: OrganicsCompressionScoringRequest) os.unlink(dist_y4m_path) continue - # Now reuse the Y4M files for color validation (no redundant conversion!) - # Extract frames for color validation from Y4M files - logger.info(f"Extracting frames from Y4M for color validation from clip (reusing VMAF Y4M files)") - dist_clip_frames = extract_frames_from_y4m(dist_y4m_path) - step_time = time.time() - uid_start_time - logger.info(f"ā™Žļø 10.5. Extracted {len(dist_clip_frames)} frames from Y4M for color validation in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.") - - # Validate color channels (grayscale check) - color_valid, color_reason = validate_color_channels_on_frames(dist_clip_frames) - if not color_valid: - logger.error(f"UID {uid}: {color_reason}") - vmaf_scores.append(0.0) - compression_rates.append(0.9999) - final_scores.append(0.0) - reasons.append(f"Color validation failed: {color_reason}") - if dist_path and os.path.exists(dist_path): - os.unlink(dist_path) - if ref_clip_path and os.path.exists(ref_clip_path): - os.unlink(ref_clip_path) - if dist_clip_path and os.path.exists(dist_clip_path): - os.unlink(dist_clip_path) - if dist_y4m_path and os.path.exists(dist_y4m_path): - os.unlink(dist_y4m_path) - continue - - logger.info(f"āœ… UID {uid}: Color channels validated - {color_reason}") - step_time = time.time() - uid_start_time - logger.info(f"ā™Žļø 10.6. Validated color channels in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.") - - # Extract reference frames from Y4M for chroma quality comparison - ref_clip_frames_data = extract_frames_from_y4m(ref_y4m_path) - - # Validate chroma quality (prevents partial UV reduction) - chroma_valid, chroma_reason = validate_chroma_quality_on_frames( - ref_clip_frames_data, dist_clip_frames, threshold=0.7 - ) - if not chroma_valid: - logger.error(f"UID {uid}: {chroma_reason}") - vmaf_scores.append(0.0) - compression_rates.append(0.9999) - final_scores.append(0.0) - reasons.append(f"Chroma validation failed: {chroma_reason}") - if dist_path and os.path.exists(dist_path): - os.unlink(dist_path) - if ref_clip_path and os.path.exists(ref_clip_path): - os.unlink(ref_clip_path) - if dist_clip_path and os.path.exists(dist_clip_path): - os.unlink(dist_clip_path) - if dist_y4m_path and os.path.exists(dist_y4m_path): - os.unlink(dist_y4m_path) - continue - - logger.info(f"āœ… UID {uid}: Chroma quality validated - {chroma_reason}") - step_time = time.time() - uid_start_time - logger.info(f"ā™Žļø 10.7. Validated chroma quality in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.") # Calculate compression score using the proper formula # Check scoring function for details From bbe0551ade2b7e837632c0c9ea662efa7d256691 Mon Sep 17 00:00:00 2001 From: Ahmad Ayad Date: Wed, 3 Dec 2025 20:25:34 +0100 Subject: [PATCH 10/10] fixed backward compatibility issue with validator protocol not providing target_codec --- neurons/miner.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/neurons/miner.py b/neurons/miner.py index 633a37e1..707993a0 100644 --- a/neurons/miner.py +++ b/neurons/miner.py @@ -67,9 +67,9 @@ async def forward_compression_requests(self, synapse: VideoCompressionProtocol) payload_url: str = synapse.miner_payload.reference_video_url vmaf_threshold: float = synapse.miner_payload.vmaf_threshold - target_codec: str = synapse.miner_payload.target_codec - codec_mode: str = synapse.miner_payload.codec_mode - target_bitrate: float = synapse.miner_payload.target_bitrate + 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} | VMAF: {vmaf_threshold} | Codec: {target_codec} | Mode: {codec_mode} | Bitrate: {target_bitrate} Mbps šŸ›œšŸ›œšŸ›œ")