Arpan Tripathi
compression competition mode (#179)
1f4a4dd unverified 3 months ago
177 kB
import os
import cv2
import glob
import hashlib
import json
import time
import torch
import random
import asyncio
import aiohttp
from loguru import logger
import tempfile
import subprocess
import numpy as np
from concurrent.futures import ThreadPoolExecutor
from pydantic import BaseModel
from typing import Optional, List
from collections import defaultdict
from firerequests import FireRequests
from vidaio_subnet_core import CONFIG
from torchvision.models import resnet50
from fastapi import FastAPI, HTTPException
import torchvision.transforms as transforms
from pieapp_metric import calculate_pieapp_score
from vidaio_subnet_core.utilities.storage_client import storage_client
from vmaf_metric import calculate_vmaf, convert_mp4_to_y4m, trim_video, trim_video_select, vmaf_metric_ffmpeg, is_vmaf_ffmpeg_available
from services.video_scheduler.video_utils import get_trim_video_path, delete_videos_with_fileid
from services.scoring.scoring_function import calculate_compression_score
from upscaling_scoring import (
    calculate_final_score,
    calculate_length_score,
    calculate_preliminary_score,
    calculate_quality_score,
)
from vidaio_subnet_core.competition.dataset import DatasetError, ModalVolumeStore
from vidaio_subnet_core.competition.media_contracts import CompetitionScoringMedia
from vidaio_subnet_core.competition.scoring import (
    CompetitionItemScorer,
    ItemScoringError,
    compute_aggregates,
)
from vidaio_subnet_core.competition.scoring_api import (
    CompetitionAggregatePayload,
    CompetitionAggregateRequest,
    CompetitionAggregateResponse,
    CompetitionHistoryComponent,
    CompetitionScoredItemPayload,
    CompetitionScoringBatchRequest,
    CompetitionScoringBatchResponse,
    CompetitionScoringBatchResult,
)

# Compression scoring constants
COMPRESSION_RATE_WEIGHT = 0.7  # w_c
COMPRESSION_VMAF_WEIGHT = 0.3  # w_vmaf
SOFT_THRESHOLD_MARGIN = 5.0  # Margin below VMAF threshold for soft scoring zone
MAX_VMAF_MODEL_DELTA = 3.0
TONE_CHECK_SAMPLE_COUNT = 5
TONE_CHECK_EXTRACTION_TIMEOUT = 6.5
_TONE_SAMPLE_SALT = os.urandom(16)

FRAME_TOLERANCE = 5  # Tolerance in frames for fast ffprobe frame count read
UPSCALING_FILE_SIZE_MULTIPLIERS = {
    2: 8,
    4: 20,
}


app = FastAPI()
fire_requests = FireRequests()

VMAF_THRESHOLD = CONFIG.score.vmaf_threshold
PIEAPP_SAMPLE_COUNT = CONFIG.score.pieapp_sample_count
PIEAPP_THRESHOLD = CONFIG.score.pieapp_threshold
VMAF_SAMPLE_COUNT = CONFIG.score.vmaf_sample_count
ORGANIC_PROXY_VERIFY_DOWNLOAD_ENDPOINT = os.getenv(
    "ORGANIC_PROXY_VERIFY_DOWNLOAD_ENDPOINT",
    "http://135.181.63.197:20000/api/v1/videos/verify_download",
)
ORGANIC_PROXY_API_KEY = os.getenv("ORGANIC_PROXY_API_KEY", "")

# Check if vmaf_ffmpeg Docker image with libvmaf_cuda is available
VMAF_FFMPEG_AVAILABLE = is_vmaf_ffmpeg_available()

import numpy as np
from PIL import Image


def detect_tone_manipulation(ref_png, dist_png):
    """Detect affine, nonlinear, local, and chroma tone manipulation on one frame."""
    size = (320, 180)
    with Image.open(ref_png) as ref_img, Image.open(dist_png) as dist_img:
        ref_rgb = np.asarray(ref_img.convert("RGB").resize(size, Image.BILINEAR), float)
        dist_rgb = np.asarray(dist_img.convert("RGB").resize(size, Image.BILINEAR), float)

    # Match PIL's grayscale conversion closely while retaining color for the
    # chroma checks below. Inputs are already small, so this analysis is cheap.
    ref_y = 0.299 * ref_rgb[..., 0] + 0.587 * ref_rgb[..., 1] + 0.114 * ref_rgb[..., 2]
    dist_y = 0.299 * dist_rgb[..., 0] + 0.587 * dist_rgb[..., 1] + 0.114 * dist_rgb[..., 2]
    r = ref_y.ravel()
    d = dist_y.ravel()

    # Drop near-clipped codes where the relationship saturates and biases slope.
    m = (r > 6) & (r < 249)
    if m.sum() < 500 or r[m].std() < 1.0:
        return {
            "slope": 1.0,
            "offset": 0.0,
            "detected": False,
            "severity": "unknown",
            "reason": "too little usable signal",
            "reliable": False,
        }

    rm, dm = r[m], d[m]
    corr = float(np.corrcoef(rm, dm)[0, 1])
    slope, offset = (float(v) for v in np.polyfit(rm, dm, 1))

    # Low correlation or implausible slope means mismatched frames, not a tone transform.
    if corr < 0.9 or not (0.5 < slope < 1.6):
        return {
            "slope": round(slope, 4),
            "offset": round(offset, 2),
            "detected": False,
            "severity": "unreliable",
            "reason": f"frames don't correspond (corr={corr:.2f}, slope={slope:.3f})",
            "reliable": False,
        }

    dev = abs(slope - 1.0)

    # A global fit can hide an S-curve whose positive and negative changes
    # cancel. Measure the median residual in luminance bins, including the
    # shadows/highlights excluded from the affine fit.
    predicted_y = slope * r + offset
    bin_residuals = []
    raw_bin_deltas = []
    for low in range(0, 256, 16):
        bin_mask = (r >= low) & (r < low + 16)
        if bin_mask.sum() >= 100:
            bin_residuals.append(float(np.median(d[bin_mask] - predicted_y[bin_mask])))
            raw_bin_deltas.append(float(np.median(d[bin_mask] - r[bin_mask])))
    max_nonlinear_delta = max((abs(value) for value in bin_residuals), default=0.0)
    max_tone_curve_delta = max((abs(value) for value in raw_bin_deltas), default=0.0)

    # Detect spatially selective adjustments that disappear in the whole-frame
    # regression. Requiring two affected tiles filters isolated codec errors.
    local_alerts = 0
    global_residual = dist_y - (slope * ref_y + offset)
    for tile_y in np.array_split(global_residual, 4, axis=0):
        for tile in np.array_split(tile_y, 4, axis=1):
            if abs(float(np.median(tile))) >= 2.0:
                local_alerts += 1

    # Compare chroma vectors directly. This catches saturation and hue changes
    # that preserve grayscale luminance and therefore evade the slope check.
    def chroma_vectors(rgb):
        y = 0.299 * rgb[..., 0] + 0.587 * rgb[..., 1] + 0.114 * rgb[..., 2]
        cb = (rgb[..., 2] - y) * 0.564
        cr = (rgb[..., 0] - y) * 0.713
        return cb.ravel(), cr.ravel()

    ref_cb, ref_cr = chroma_vectors(ref_rgb)
    dist_cb, dist_cr = chroma_vectors(dist_rgb)
    ref_energy = float(np.mean(ref_cb**2 + ref_cr**2))
    dist_energy = float(np.mean(dist_cb**2 + dist_cr**2))
    saturation_ratio = (dist_energy / ref_energy) ** 0.5 if ref_energy >= 4.0 else 1.0
    dot = float(np.sum(ref_cb * dist_cb + ref_cr * dist_cr))
    cross = float(np.sum(ref_cb * dist_cr - ref_cr * dist_cb))
    hue_shift_degrees = float(np.degrees(np.arctan2(cross, dot))) if ref_energy >= 4.0 else 0.0
    chroma_mean_shift = float(np.hypot(np.mean(dist_cb - ref_cb), np.mean(dist_cr - ref_cr)))

    affine_alert = dev >= 0.012 or abs(offset) >= 2.0
    # These are per-frame candidate thresholds. The batch aggregator requires
    # the signal on at least two frames unless it is twice as strong, which
    # keeps ordinary codec noise from becoming a rejection.
    nonlinear_alert = max_nonlinear_delta >= 1.0 or max_tone_curve_delta >= 2.0
    local_alert = local_alerts >= 2
    chroma_alert = (
        abs(saturation_ratio - 1.0) >= 0.08
        or abs(hue_shift_degrees) >= 2.0
        or chroma_mean_shift >= 2.0
    )
    detected = affine_alert or nonlinear_alert or local_alert or chroma_alert

    signals = []
    if affine_alert:
        signals.append(f"affine={slope:.3f}*ref{offset:+.1f}")
    if nonlinear_alert:
        signals.append(f"nonlinear_delta={max_nonlinear_delta:.2f}")
    if local_alert:
        signals.append(f"local_tiles={local_alerts}/16")
    if chroma_alert:
        signals.append(
            f"chroma(sat={saturation_ratio:.3f}, hue={hue_shift_degrees:+.2f}deg, "
            f"mean_shift={chroma_mean_shift:.2f})"
        )

    if detected:
        sev = "alert"
        reason = "tone manipulation signals: " + ", ".join(signals)
    elif dev >= 0.005 or abs(offset) >= 1.0:
        sev = "info"
        reason = f"mild levels shift: {slope:.3f}*ref {offset:+.1f}"
    else:
        sev = "good"
        reason = f"levels preserved: {slope:.3f}*ref {offset:+.1f}"

    return {
        "slope": round(slope, 4),
        "offset": round(offset, 2),
        "detected": detected,
        "severity": sev,
        "reason": reason,
        "reliable": True,
        "correlation": round(corr, 4),
        "nonlinear_delta": round(max_nonlinear_delta, 2),
        "tone_curve_delta": round(max_tone_curve_delta, 2),
        "local_alert_tiles": local_alerts,
        "saturation_ratio": round(saturation_ratio, 4),
        "hue_shift_degrees": round(hue_shift_degrees, 2),
        "chroma_mean_shift": round(chroma_mean_shift, 2),
    }


def validate_upscaling_file_size(reference_path, distorted_path, scale_factor):
    """Validate that an upscaled video stays within the allowed file size."""
    max_multiplier = UPSCALING_FILE_SIZE_MULTIPLIERS[scale_factor]
    reference_file_size = os.path.getsize(reference_path)
    distorted_file_size = os.path.getsize(distorted_path)
    max_distorted_file_size = reference_file_size * max_multiplier

    logger.info(
        f"Upscaling file size check: distorted={distorted_file_size / (1024 * 1024):.2f} MB, "
        f"reference={reference_file_size / (1024 * 1024):.2f} MB, "
        f"limit={max_distorted_file_size / (1024 * 1024):.2f} MB ({max_multiplier}x)"
    )

    return distorted_file_size <= max_distorted_file_size, max_multiplier

class UpscalingScoringRequest(BaseModel):
    """
    Request model for upscaling scoring. Contains URLs for distorted videos and the reference video path.
    """
    distorted_urls: List[str]
    reference_paths: List[str]
    uids: List[int]
    payload_urls: List[str]
    video_ids: List[str]
    uploaded_object_names: List[str]
    content_lengths: List[int]
    task_types: List[str]
    fps: Optional[float] = None
    subsample: Optional[int] = 1
    verbose: Optional[bool] = True
    progress: Optional[bool] = False

class CompressionScoringRequest(BaseModel):
    """
    Request model for compression scoring. Contains paths for distorted videos and the reference video path.
    """
    distorted_file_paths: List[Optional[str]]
    distorted_urls: Optional[List[str]] = None
    reference_paths: List[str]
    uids: List[int]
    video_ids: List[str]
    uploaded_object_names: List[str]
    vmaf_thresholds: List[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
    progress: Optional[bool] = False

class OrganicsUpscalingScoringRequest(BaseModel):
    """
    Request model for scoring. Contains URLs for distorted videos and the reference video path.
    """
    distorted_urls: List[str]
    reference_urls: List[str]
    task_types: List[str]
    uids: List[int]
    fps: Optional[float] = None
    subsample: Optional[int] = 1
    verbose: Optional[bool] = False
    progress: Optional[bool] = False

class OrganicsCompressionScoringRequest(BaseModel):
    """
    Request model for scoring. Contains URLs for distorted videos and the reference video path.
    """
    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
    verbose: Optional[bool] = False
    progress: Optional[bool] = False


class UpscalingScoringResponse(BaseModel):
    """
    Response model for upscaling scoring. Contains the list of calculated scores for each distorted video.
    """
    vmaf_scores: List[float]
    pieapp_scores: List[float]
    quality_scores: List[float]
    length_scores: List[float]
    final_scores: List[float]
    reasons: List[str]

class CompressionScoringResponse(BaseModel):
    """
    Response model for compression scoring. Contains the list of calculated scores for each distorted video.
    """
    vmaf_scores: List[float]
    base_vmaf_scores: Optional[List[Optional[float]]] = None
    compression_rates: List[float]
    final_scores: List[float]
    reasons: List[str]

class OrganicsUpscalingScoringResponse(BaseModel):
    """
    Response model for organics scoring. Contains the list of calculated scores for each distorted video.
    """
    vmaf_scores: List[float]
    pieapp_scores: List[float]
    quality_scores: List[float]
    length_scores: List[float]
    final_scores: List[float]
    reasons: List[str]

class OrganicsCompressionScoringResponse(BaseModel):
    """
    Response model for organics scoring. Contains the list of calculated scores for each distorted video.
    """
    vmaf_scores: List[float]
    base_vmaf_scores: Optional[List[Optional[float]]] = None
    compression_rates: List[float]
    final_scores: List[float]
    reasons: List[str]


@app.post(
    "/score_compression_competition",
    response_model=CompetitionScoringBatchResponse,
)
async def score_compression_competition(
    request: CompetitionScoringBatchRequest,
) -> CompetitionScoringBatchResponse:
    """Read and score one trusted competition batch outside the validator."""

    logger.info(
        "Starting competition compression scoring batch: competition_id={} "
        "items={} output_volume={}",
        request.manifest.competition_id,
        [value.item.evaluation_id for value in request.items],
        request.output_volume_name,
    )
    store = ModalVolumeStore(environment_name=request.modal_environment)
    scorer = CompetitionItemScorer()
    results = []
    for batch_item in request.items:
        evaluation_id = batch_item.item.evaluation_id
        try:
            source_bytes, output_bytes = await asyncio.gather(
                asyncio.to_thread(
                    store.read_bytes,
                    request.input_volume_name,
                    batch_item.item.source_path,
                ),
                asyncio.to_thread(
                    store.read_bytes,
                    request.output_volume_name,
                    batch_item.output_path,
                ),
            )
            metrics = await asyncio.to_thread(
                scorer.score_media,
                request.manifest,
                batch_item.item,
                CompetitionScoringMedia.compression(source_bytes, output_bytes),
                runtime_seconds=batch_item.runtime_seconds,
                allocated_gpu_type=batch_item.allocated_gpu_type,
                allocated_gpu_count=batch_item.allocated_gpu_count,
                allocated_cpu_cores=batch_item.allocated_cpu_cores,
            )
            results.append(
                CompetitionScoringBatchResult(
                    evaluation_id=evaluation_id,
                    status="SCORED",
                    metrics=CompetitionScoredItemPayload.from_scored_item(metrics),
                )
            )
        except DatasetError as exc:
            logger.error(
                "Competition scoring Volume read failed: competition_id={} "
                "evaluation_id={} reason={}",
                request.manifest.competition_id,
                evaluation_id,
                str(exc),
            )
            results.append(
                CompetitionScoringBatchResult(
                    evaluation_id=evaluation_id,
                    status="FAILED",
                    reason_code="VOLUME_READ_FAILED",
                )
            )
        except ItemScoringError as exc:
            results.append(
                CompetitionScoringBatchResult(
                    evaluation_id=evaluation_id,
                    status="FAILED",
                    reason_code=exc.reason_code,
                    metrics=(
                        CompetitionScoredItemPayload.from_scored_item(exc.metrics)
                        if exc.metrics is not None
                        else None
                    ),
                )
            )
        except Exception as exc:
            logger.exception(
                "Competition scoring infrastructure failure: competition_id={} "
                "evaluation_id={} reason={}",
                request.manifest.competition_id,
                evaluation_id,
                str(exc),
            )
            results.append(
                CompetitionScoringBatchResult(
                    evaluation_id=evaluation_id,
                    status="FAILED",
                    reason_code="EVALUATION_INFRASTRUCTURE_ERROR",
                )
            )

    logger.info(
        "Completed competition compression scoring batch: competition_id={} "
        "statuses={}",
        request.manifest.competition_id,
        [value.status for value in results],
    )
    return CompetitionScoringBatchResponse(results=tuple(results))


@app.post(
    "/score_compression_competition_aggregates",
    response_model=CompetitionAggregateResponse,
)
async def score_compression_competition_aggregates(
    request: CompetitionAggregateRequest,
) -> CompetitionAggregateResponse:
    """Compute final contender and per-history components as one batch."""

    aggregates, components = await asyncio.to_thread(
        compute_aggregates,
        request.manifest,
        request.histories,
    )
    return CompetitionAggregateResponse(
        aggregates=tuple(
            CompetitionAggregatePayload.from_aggregate(value)
            for value in aggregates
        ),
        components=tuple(
            CompetitionHistoryComponent(
                history_id=history_id,
                media_score=values[0],
                compression=values[1],
                vmaf_quality=values[2],
                cost_efficiency=values[3],
                completion=values[4],
            )
            for history_id, values in sorted(components.items())
        ),
    )

# Load pre-trained model for feature extraction
def load_quality_model():
    """Load ResNet50 model for quality assessment"""
    from torchvision.models import ResNet50_Weights
    model = resnet50(weights=ResNet50_Weights.IMAGENET1K_V1)
    model.eval()
    return model

_shared_session: Optional[aiohttp.ClientSession] = None

async def get_shared_session() -> aiohttp.ClientSession:
    """Get or create a shared aiohttp ClientSession for connection pooling."""
    global _shared_session
    if _shared_session is None or _shared_session.closed:
        # sock_read timeout prevents hanging on stalled downloads
        timeout = aiohttp.ClientTimeout(sock_connect=30, sock_read=30, total=300)
        connector = aiohttp.TCPConnector(limit=50)  # Adjust concurrency limit as needed
        _shared_session = aiohttp.ClientSession(timeout=timeout, connector=connector)
    return _shared_session

def calculate_base_vmaf(
    ref_y4m_path: str,
    dist_y4m_path: str,
    neg_model: bool = False,
) -> Optional[float]:
    """Calculate VMAF with GPU-accelerated libvmaf for Y4M inputs."""
    model_label = "VMAF NEG v0.6.1" if neg_model else "VMAF v0.6.1"
    try:
        vmaf_score = vmaf_metric_ffmpeg(
            dist_path=dist_y4m_path,
            ref_path=ref_y4m_path,
            neg_model=neg_model,
        )
        logger.info(
            f"{model_label} score (Y4M, ffmpeg/docker): {vmaf_score}"
        )
        return vmaf_score
    except Exception as vmaf_err:
        logger.warning(
            f"{model_label} calculation failed in Y4M ffmpeg/docker path; "
            f"VMAF model delta check is unavailable: {vmaf_err}"
        )
        return None


def _tone_sample_timestamps(
    duration_seconds: Optional[float],
    sample_key: Optional[str],
    fallback_timestamp: float,
    sample_count: int = TONE_CHECK_SAMPLE_COUNT,
) -> List[float]:
    """Choose unpredictable, repeatable, stratified timestamps for one source video."""
    duration = float(duration_seconds or 0.0)
    if duration <= 0.0 or sample_count <= 1:
        return [max(0.0, float(fallback_timestamp or 0.0))]

    # The process-local salt prevents miners from predicting the exact samples.
    # sample_key keeps timestamps identical when different miners encode the
    # same source on this scoring worker.
    key = str(sample_key or f"duration:{duration:.6f}").encode("utf-8")
    seed = int.from_bytes(hashlib.sha256(_TONE_SAMPLE_SALT + key).digest()[:8], "big")
    rng = random.Random(seed)
    latest_safe_timestamp = max(0.0, duration - (1.0 / 30.0))
    timestamps = []
    for index in range(sample_count):
        # Keep away from exact stratum boundaries and randomly jitter within
        # each fifth of the video.
        fraction = (index + rng.uniform(0.15, 0.85)) / sample_count
        timestamps.append(min(latest_safe_timestamp, fraction * duration))
    return timestamps


def _extract_tone_check_frames(
    video_path: str,
    timestamps_seconds: List[float],
    output_paths: List[str],
):
    """Extract several random-access frames with one FFmpeg process."""
    if len(timestamps_seconds) != len(output_paths) or not output_paths:
        raise ValueError("tone-check timestamps and outputs must have the same non-zero length")

    cmd = [
        "ffmpeg",
        "-hide_banner",
        "-loglevel",
        "error",
        "-y",
    ]
    for timestamp_seconds in timestamps_seconds:
        cmd.extend(["-ss", f"{timestamp_seconds:.6f}", "-i", video_path])
    for input_index, output_path in enumerate(output_paths):
        cmd.extend(
            [
                "-map",
                f"{input_index}:v:0",
                "-frames:v",
                "1",
                "-vf",
                "scale=320:180:flags=bilinear",
                "-compression_level",
                "1",
                output_path,
            ]
        )

    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=TONE_CHECK_EXTRACTION_TIMEOUT,
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or "ffmpeg frame extraction failed")
    missing_outputs = [
        output_path
        for output_path in output_paths
        if not os.path.exists(output_path) or os.path.getsize(output_path) == 0
    ]
    if missing_outputs:
        raise RuntimeError(f"ffmpeg did not produce {len(missing_outputs)} tone-check frame(s)")


def _extract_tone_check_frame(video_path: str, timestamp_seconds: float, output_path: str):
    """Backward-compatible single-frame wrapper."""
    _extract_tone_check_frames(video_path, [timestamp_seconds], [output_path])


def _aggregate_tone_results(frame_results: List[dict], timestamps: List[float]) -> dict:
    """Combine frame evidence while resisting isolated codec artifacts."""
    reliable_results = [result for result in frame_results if result.get("reliable") is True]
    unreliable_count = len(frame_results) - len(reliable_results)
    alert_results = [result for result in reliable_results if result.get("detected") is True]

    def is_strong_alert(result):
        return (
            abs(float(result.get("slope", 1.0)) - 1.0) >= 0.024
            or abs(float(result.get("offset", 0.0))) >= 4.0
            or float(result.get("nonlinear_delta", 0.0)) >= 3.0
            or float(result.get("tone_curve_delta", 0.0)) >= 6.0
            or int(result.get("local_alert_tiles", 0)) >= 4
            or abs(float(result.get("saturation_ratio", 1.0)) - 1.0) >= 0.16
            or abs(float(result.get("hue_shift_degrees", 0.0))) >= 4.0
            or float(result.get("chroma_mean_shift", 0.0)) >= 4.0
        )

    strong_alerts = sum(is_strong_alert(result) for result in alert_results)
    verification_failed = unreliable_count > len(frame_results) // 2
    detected = len(alert_results) >= 2 or strong_alerts >= 1 or verification_failed

    slopes = [float(result.get("slope", 1.0)) for result in reliable_results] or [1.0]
    offsets = [float(result.get("offset", 0.0)) for result in reliable_results] or [0.0]
    slope = float(np.median(slopes))
    offset = float(np.median(offsets))
    max_nonlinear_delta = max(
        (float(result.get("nonlinear_delta", 0.0)) for result in reliable_results),
        default=0.0,
    )
    max_tone_curve_delta = max(
        (float(result.get("tone_curve_delta", 0.0)) for result in reliable_results),
        default=0.0,
    )
    max_hue_shift = max(
        (abs(float(result.get("hue_shift_degrees", 0.0))) for result in reliable_results),
        default=0.0,
    )
    max_saturation_deviation = max(
        (abs(float(result.get("saturation_ratio", 1.0)) - 1.0) for result in reliable_results),
        default=0.0,
    )

    if verification_failed:
        severity = "alert"
        reason = (
            f"tone verification unreliable on {unreliable_count}/{len(frame_results)} frames"
        )
    elif detected:
        severity = "alert"
        reason = (
            f"tone manipulation detected on {len(alert_results)}/{len(frame_results)} frames; "
            f"strong_alerts={strong_alerts}; max_nonlinear_delta={max_nonlinear_delta:.2f}; "
            f"max_hue_shift={max_hue_shift:.2f}deg"
        )
    elif alert_results:
        severity = "info"
        reason = (
            f"isolated tone signal on 1/{len(frame_results)} frames; "
            f"median levels preserved: {slope:.3f}*ref {offset:+.1f}"
        )
    else:
        severity = "good" if reliable_results else "unknown"
        reason = (
            f"levels preserved across {len(reliable_results)}/{len(frame_results)} frames: "
            f"median {slope:.3f}*ref {offset:+.1f}"
        )

    return {
        "slope": round(slope, 4),
        "offset": round(offset, 2),
        "detected": detected,
        "severity": severity,
        "reason": reason,
        "sample_count": len(frame_results),
        "reliable_samples": len(reliable_results),
        "alert_samples": len(alert_results),
        "strong_alerts": strong_alerts,
        "max_nonlinear_delta": round(max_nonlinear_delta, 2),
        "max_tone_curve_delta": round(max_tone_curve_delta, 2),
        "max_hue_shift_degrees": round(max_hue_shift, 2),
        "max_saturation_deviation": round(max_saturation_deviation, 4),
        "timestamps": [round(timestamp, 3) for timestamp in timestamps],
    }


def check_tone_manipulation_for_compression(
    ref_path: str,
    dist_path: str,
    timestamp_seconds: float,
    uid: Optional[int] = None,
    context: str = "compression",
    duration_seconds: Optional[float] = None,
    sample_key: Optional[str] = None,
) -> Optional[dict]:
    """Detect and log multi-frame tone/levels manipulation for compression scoring."""
    uid_label = f" UID {uid}" if uid is not None else ""
    check_started = time.monotonic()
    try:
        timestamp_seconds = max(0.0, float(timestamp_seconds or 0.0))
        timestamps = _tone_sample_timestamps(
            duration_seconds,
            sample_key,
            timestamp_seconds,
        )
        with tempfile.TemporaryDirectory(prefix="tone_check_") as temp_dir:
            ref_pngs = [
                os.path.join(temp_dir, f"ref_{index}.png")
                for index in range(len(timestamps))
            ]
            dist_pngs = [
                os.path.join(temp_dir, f"dist_{index}.png")
                for index in range(len(timestamps))
            ]

            # Reference and distorted seeks are independent. Running the two
            # batched commands concurrently keeps five-frame sampling near the
            # latency of the previous two sequential single-frame commands.
            with ThreadPoolExecutor(max_workers=2) as executor:
                ref_future = executor.submit(
                    _extract_tone_check_frames,
                    ref_path,
                    timestamps,
                    ref_pngs,
                )
                dist_future = executor.submit(
                    _extract_tone_check_frames,
                    dist_path,
                    timestamps,
                    dist_pngs,
                )
                ref_future.result()
                dist_future.result()

            frame_results = [
                detect_tone_manipulation(ref_png, dist_png)
                for ref_png, dist_png in zip(ref_pngs, dist_pngs)
            ]
            tone_result = _aggregate_tone_results(frame_results, timestamps)

        elapsed = time.monotonic() - check_started
        tone_result["elapsed_seconds"] = round(elapsed, 3)
        logger.info(
            f"Tone manipulation check{uid_label} [{context}] "
            f"across {len(timestamps)} frames in {elapsed:.3f}s: {tone_result}"
        )
        return tone_result
    except Exception as e:
        elapsed = time.monotonic() - check_started
        logger.warning(
            f"Tone manipulation check failed{uid_label} "
            f"[{context}] after {elapsed:.3f}s: {e}"
        )
        return None


async def verify_organic_proxy_download(video_url: str, session: Optional[aiohttp.ClientSession] = None) -> bool:
    """Check whether the organic proxy can download a miner's output URL."""
    verify_session = session or await get_shared_session()
    headers = {"X-API-Key": ORGANIC_PROXY_API_KEY}

    async with verify_session.post(
        ORGANIC_PROXY_VERIFY_DOWNLOAD_ENDPOINT,
        json={"video_url": video_url},
        headers=headers,
        timeout=aiohttp.ClientTimeout(total=30),
    ) as response:
        if response.status != 200:
            response_text = await response.text()
            raise Exception(
                f"Organic proxy download verification failed. HTTP status: {response.status}, "
                f"response: {response_text[:500]}"
            )
        else:
            logger.info(f"Organic proxy download verification succeeded for {video_url}")

        response_data = await response.json()
        downloadable = response_data.get("downloadable")
        if not isinstance(downloadable, bool):
            raise Exception("Organic proxy download verification response is missing a boolean downloadable field")

        if not downloadable:
            logger.warning(
                f"Organic proxy could not download {video_url}: "
                f"{response_data.get('reason', 'no reason provided')}"
            )

        return downloadable

async def download_video(
    video_url: str,
    verbose: bool,
    session: Optional[aiohttp.ClientSession] = None,
    return_final_url: bool = False,
) -> tuple[str, float] | tuple[str, float, str]:
    """
    Download a video from the given URL and save it to a temporary file.
    Supports retries, connection pooling, partial download resumes, and tuned timeouts.
    Args:
        video_url (str): The URL of the video to download.
        verbose (bool): Whether to show download progress.
        session (Optional[aiohttp.ClientSession]): Shared session for connection pooling.
    Returns:
        tuple[str, float]: A tuple containing the path to the downloaded video file
                           and the time taken to download it.
    Raises:
        Exception: If the download fails or takes longer than the timeout.
    """
    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as vid_temp:
        file_path = vid_temp.name  # Path to the temporary file

    max_retries = 2
    start_time = time.time()  # Record start time

    dl_session = session or await get_shared_session()
    final_url = video_url

    for attempt in range(max_retries + 1):
        try:
            current_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
            headers = {}
            
            # If we have partially downloaded data, try to resume
            if current_size > 0:
                headers["Range"] = f"bytes={current_size}-"
                if verbose:
                    logger.info(f"Resuming download from byte {current_size} for {video_url} to {file_path} (Attempt {attempt + 1}/{max_retries + 1})")
            else:
                if verbose:
                    logger.info(f"Downloading video from {video_url} to {file_path} (Attempt {attempt + 1}/{max_retries + 1})")

            async with dl_session.get(video_url, headers=headers) as response:
                final_url = str(response.url)
                if response.status == 416:  # Range Not Satisfiable
                    # The file might be fully downloaded already, or the server rejected the range.
                    # We truncate the file and retry from scratch to be safe.
                    with open(file_path, "wb") as f:
                        pass
                    raise Exception("Server returned 416 Range Not Satisfiable. Starting over.")
                
                if response.status not in (200, 206):
                    raise Exception(f"Failed to download video. HTTP status: {response.status}")

                # If status is 206 Partial Content, append. If 200 OK, the server ignored Range, so overwrite.
                mode = "ab" if response.status == 206 else "wb"

                with open(file_path, mode) as f:
                    async for chunk in response.content.iter_chunked(2 * 1024 * 1024):
                        f.write(chunk)

            end_time = time.time()  # Record end time
            download_time = end_time - start_time  # Calculate download duration

            if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
                raise Exception(f"Download failed or file is empty: {file_path}")

            if verbose:
                logger.info(f"File successfully downloaded to: {file_path}")
                logger.info(f"Download time: {download_time:.2f} seconds")

            if return_final_url:
                return file_path, download_time, final_url
            return file_path, download_time
        except aiohttp.ServerTimeoutError:
            if attempt == max_retries:
                raise Exception("Download failed: Server connection timed out")
            if verbose:
                logger.warning(f"Download failed: Server connection timed out. Retrying {attempt + 1}/{max_retries}...")
            await asyncio.sleep(2 ** attempt)
        except asyncio.TimeoutError:
            if attempt == max_retries:
                raise Exception("Download timed out")
            if verbose:
                logger.warning(f"Download timed out. Retrying {attempt + 1}/{max_retries}...")
            await asyncio.sleep(2 ** attempt)
        except aiohttp.ClientError as e:
            if attempt == max_retries:
                raise Exception(f"Download failed due to a network error: {type(e).__name__}: {repr(e)}")
            if verbose:
                logger.warning(f"Download failed due to a network error: {type(e).__name__}: {repr(e)}. Retrying {attempt + 1}/{max_retries}...")
            await asyncio.sleep(2 ** attempt)
        except Exception as e:
            if attempt == max_retries:
                raise
            if verbose:
                logger.warning(f"Download failed: {str(e)}. Retrying {attempt + 1}/{max_retries}...")
            await asyncio.sleep(2 ** attempt)

def _normalize_duplicate_url(url: Optional[str]) -> str:
    return (url or "").strip()

async def _resolve_final_url(video_url: str) -> str:
    if not video_url or len(video_url) < 10:
        return ""

    timeout = aiohttp.ClientTimeout(sock_connect=10, total=30)
    try:
        async with aiohttp.ClientSession(timeout=timeout) as session:
            try:
                async with session.head(video_url, allow_redirects=True) as response:
                    if response.status < 400:
                        return str(response.url)
            except Exception:
                pass

            async with session.get(
                video_url,
                allow_redirects=True,
                headers={"Range": "bytes=0-0"},
            ) as response:
                if response.status < 400:
                    return str(response.url)
    except Exception as e:
        logger.warning(f"Failed to resolve final URL for duplicate detection: {video_url} | {e}")

    return video_url

async def _resolve_final_urls(video_urls: List[str]) -> List[str]:
    return await asyncio.gather(
        *[_resolve_final_url(url) for url in video_urls],
        return_exceptions=False,
    )

def _mark_duplicate_groups(
    duplicate_reasons: dict[int, list[str]],
    label: str,
    values: List[object],
    uids: List[int],
) -> None:
    groups = defaultdict(list)
    for idx, value in enumerate(values):
        if value in (None, ""):
            continue
        groups[value].append(idx)

    for value, indices in groups.items():
        if len(indices) <= 1:
            continue
        uid_list = [uids[idx] for idx in indices]
        reason = (
            f"duplicate miner response by {label}: {value} shared by "
            f"indices={indices}, uids={uid_list}"
        )
        logger.warning(reason)
        for idx in indices:
            duplicate_reasons[idx].append(reason)

def _duplicate_response_reasons(
    uids: List[int],
    submitted_urls: List[str],
    downloaded_paths: List[Optional[str]],
    final_urls: Optional[List[str]] = None,
) -> dict[int, str]:
    duplicate_reasons: dict[int, list[str]] = defaultdict(list)

    _mark_duplicate_groups(
        duplicate_reasons,
        "submitted URL",
        [_normalize_duplicate_url(url) for url in submitted_urls],
        uids,
    )

    if final_urls is not None:
        _mark_duplicate_groups(
            duplicate_reasons,
            "redirected URL",
            [_normalize_duplicate_url(url) for url in final_urls],
            uids,
        )

    file_sizes = []
    for path in downloaded_paths:
        if path and os.path.exists(path):
            file_sizes.append(os.path.getsize(path))
        else:
            file_sizes.append(None)
    _mark_duplicate_groups(duplicate_reasons, "downloaded file size", file_sizes, uids)

    return {
        idx: "MINER FAILURE: " + " | ".join(reasons)
        for idx, reasons in duplicate_reasons.items()
    }

# Function to get ClipIQA+ score programmatically
def get_clipiqa_score(video_path, num_frames=3):
    """Get ClipIQA+ score for a video file"""
    frames = extract_frames(video_path, num_frames)
    score = calculate_clipiqa_plus_score(frames)
    return score

# Extract frames from video
def extract_frames(video_path, num_frames=3, frame_indices=None):
    """Extract frames from video using specified indices or random ones"""
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    
    if total_frames < num_frames:
        raise ValueError(f"Video has only {total_frames} frames, but {num_frames} frames requested")
    
    # Use provided frame indices or select random ones
    if frame_indices is None:
        frame_indices = random.sample(range(total_frames), num_frames)
    else:
        # Validate that all requested indices are within bounds
        if max(frame_indices) >= total_frames:
            raise ValueError(f"Frame index {max(frame_indices)} exceeds video length {total_frames}")
    
    frames = []
    for idx in frame_indices:
        cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
        ret, frame = cap.read()
        if ret:
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)  # Convert BGR to RGB
            frames.append(frame)
        else:
            raise ValueError(f"Failed to read frame at index {idx}")
    
    cap.release()
    return frames

def extract_frames_from_y4m(y4m_path):
    """
    Extract frames from Y4M file for color validation.
    This reuses the Y4M file already created for VMAF calculation,
    eliminating redundant video processing.
    
    Args:
        y4m_path: Path to Y4M file (output of convert_mp4_to_y4m)
        
    Returns:
        list: List of frames as numpy arrays (BGR format for cv2)
    """
    frames = []
    
    # Create temporary directory for extracted frames
    temp_dir = tempfile.mkdtemp()
    output_pattern = os.path.join(temp_dir, "frame_%04d.png")
    
    try:
        # Extract all frames from Y4M as PNG using FFmpeg
        # Y4M files already contain only the selected frames, so no filtering needed
        extract_cmd = [
            "ffmpeg", "-i", y4m_path,
            output_pattern,
            "-y",
            "-hide_banner", "-loglevel", "error"
        ]
        
        result = subprocess.run(
            extract_cmd, 
            capture_output=True, 
            text=True, 
            timeout=30
        )
        
        if result.returncode != 0:
            logger.error(f"FFmpeg Y4M frame extraction failed: {result.stderr}")
            return frames
        
        # Read extracted PNG files in order
        png_files = sorted(glob.glob(os.path.join(temp_dir, "frame_*.png")))
        
        for png_file in png_files:
            frame = cv2.imread(png_file)
            if frame is not None:
                frames.append(frame)
            else:
                logger.warning(f"Failed to read extracted frame: {png_file}")
        
        logger.debug(f"Successfully extracted {len(frames)} frames from Y4M file")
        
    except subprocess.TimeoutExpired:
        logger.error("FFmpeg Y4M frame extraction timed out")
    except Exception as e:
        logger.error(f"Error extracting frames from Y4M: {e}")
    finally:
        # Cleanup temporary files
        try:
            for png_file in glob.glob(os.path.join(temp_dir, "*.png")):
                os.remove(png_file)
            os.rmdir(temp_dir)
        except Exception as e:
            logger.warning(f"Failed to cleanup temp frames: {e}")
    
    return frames

import numpy as np
import cv2

import numpy as np
import cv2

def validate_color_channels_on_frames(
    ref_frames,
    dist_frames,
    max_gray_ratio=0.50,
    brightness_threshold=20.0,
    ref_color_threshold=6.0,
    dist_gray_threshold=4.0,
    min_checked_frames=30,
    debug_collect_indices=False,
):
    """
    Validate color preservation between a reference and a distorted video sequence.
    The function compares per-frame color information using a simple RGB channel
    difference heuristic and evaluates color consistency over time rather than
    on a single-frame basis.
    Design principles:
      - Frames that are very dark or nearly grayscale in the reference are ignored,
        as they do not provide reliable evidence of color content.
      - Only frames with clear color presence in the reference are evaluated.
      - A limited fraction of frames with reduced chroma in the distorted sequence
        is tolerated to account for natural content variation and encoding effects.
    Args:
        ref_frames (List[np.ndarray]):
            Reference video frames (BGR format, as returned by OpenCV).
        dist_frames (List[np.ndarray]):
            Distorted/processed video frames (BGR format, as returned by OpenCV).
        max_gray_ratio (float):
            Maximum allowed fraction (0.0–1.0) of evaluated frames that may exhibit
            reduced color variation in the distorted video.
        brightness_threshold (float):
            Minimum average brightness of a reference frame required for evaluation.
        ref_color_threshold (float):
            Minimum RGB channel-difference value for a reference frame to be
            considered as having meaningful color content.
        dist_gray_threshold (float):
            Maximum RGB channel-difference value below which a distorted frame is
            considered to have low color variation.
        min_checked_frames (int):
            Minimum number of evaluated frames required before applying strict
            ratio-based validation.
        debug_collect_indices (bool):
            If True, frame indices contributing to validation failures are included
            in the returned message for diagnostic purposes.
    Returns:
        Tuple[bool, str]:
            A tuple containing:
              - is_valid (bool): True if color preservation is acceptable.
              - reason (str): Human-readable explanation of the validation result.
    """
    try:
        if not ref_frames or not dist_frames:
            return False, "No frames available for color validation"

        checked_frames = 0
        low_color_frames = 0
        low_color_indices = []

        for i, (ref_frame, dist_frame) in enumerate(zip(ref_frames, dist_frames)):
            # Convert from BGR (OpenCV default) to RGB for channel comparison
            ref_rgb = cv2.cvtColor(ref_frame, cv2.COLOR_BGR2RGB).astype(np.float32)
            dist_rgb = cv2.cvtColor(dist_frame, cv2.COLOR_BGR2RGB).astype(np.float32)

            # Skip very dark reference frames
            avg_ref_brightness = float(np.mean(ref_rgb))
            if avg_ref_brightness < brightness_threshold:
                continue

            # Estimate color variation in the reference frame
            ref_color_diff = max(
                float(np.mean(np.abs(ref_rgb[:, :, 0] - ref_rgb[:, :, 1]))),
                float(np.mean(np.abs(ref_rgb[:, :, 0] - ref_rgb[:, :, 2]))),
                float(np.mean(np.abs(ref_rgb[:, :, 1] - ref_rgb[:, :, 2]))),
            )

            # Only evaluate frames with sufficient color information in the reference
            if ref_color_diff < ref_color_threshold:
                continue

            checked_frames += 1

            # Estimate color variation in the distorted frame
            dist_color_diff = max(
                float(np.mean(np.abs(dist_rgb[:, :, 0] - dist_rgb[:, :, 1]))),
                float(np.mean(np.abs(dist_rgb[:, :, 0] - dist_rgb[:, :, 2]))),
                float(np.mean(np.abs(dist_rgb[:, :, 1] - dist_rgb[:, :, 2]))),
            )

            # Count frames with noticeably reduced color variation
            if dist_color_diff < dist_gray_threshold:
                low_color_frames += 1
                if debug_collect_indices:
                    low_color_indices.append(i)

                logger.debug(
                    f"Frame {i}: reduced color variation detected "
                    f"(ref_diff={ref_color_diff:.2f}, dist_diff={dist_color_diff:.2f})"
                )

        if checked_frames == 0:
            return True, (
                "No reference frames with sufficient color information were found; "
                "color validation was skipped"
            )

        low_color_ratio = low_color_frames / checked_frames

        logger.info(
            "Color validation summary: "
            f"{low_color_frames}/{checked_frames} frames with reduced color variation "
            f"({low_color_ratio:.2%}); allowed maximum {max_gray_ratio:.2%}"
        )

        # Handle small sample sizes conservatively
        if checked_frames < min_checked_frames:
            if low_color_ratio > max_gray_ratio:
                return False, (
                    f"Color preservation check failed on a small sample: "
                    f"{low_color_ratio:.1%} of evaluated frames show reduced color variation "
                    f"(>{max_gray_ratio:.0%} allowed)"
                )
            return True, (
                f"Color preservation validated on a small sample: "
                f"{low_color_ratio:.1%} of frames show reduced color variation"
            )

        # Standard ratio-based decision
        if low_color_ratio > max_gray_ratio:
            if debug_collect_indices and low_color_indices:
                preview = ", ".join(map(str, low_color_indices[:20]))
                extra = "" if len(low_color_indices) <= 20 else f" ...(+{len(low_color_indices) - 20} more)"
                return False, (
                    f"Color preservation failed: {low_color_ratio:.1%} of evaluated frames "
                    f"exhibit reduced color variation (>{max_gray_ratio:.0%} allowed). "
                    f"Example frame indices: [{preview}{extra}]"
                )

            return False, (
                f"Color preservation failed: {low_color_ratio:.1%} of evaluated frames "
                f"exhibit reduced color variation (>{max_gray_ratio:.0%} allowed)"
            )

        return True, (
            f"Color preservation validated: {low_color_ratio:.1%} of evaluated frames "
            f"exhibit reduced color variation (≤{max_gray_ratio:.0%} allowed)"
        )

    except Exception as e:
        logger.error(f"Error during color channel validation: {e}")
        return False, f"Error during color validation: {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 abs(len(ref_frames) - len(dist_frames)) > FRAME_TOLERANCE:
            return False, "Frame count mismatch between reference and distorted: {} vs {}".format(len(ref_frames), len(dist_frames))

        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)}"


def _get_signalstats(video_path, n_subsample=1, start_time=None, duration=None, max_frames=30):
    """
    Run ffmpeg signalstats filter on a video and return per-frame statistics.
    Extracts YAVG (luma average) and SATAVG (saturation average) per frame
    directly from the MP4 — no Y4M conversion needed.
    Uses CUDA hardware-accelerated decoding when available, falling back to
    CPU if GPU decode fails. Uses metadata=mode=print to ensure stats are
    printed to stderr.
    FFmpeg processing is bounded by max_frames — FFmpeg stops decoding after
    max_frames frames, so even hour-long videos are fast.
    Args:
        video_path: Path to input video
        n_subsample: Analyze every Nth frame (1 = all frames, 5 = every 5th frame)
        start_time: Optional start time in seconds (analyze from this point)
        duration: Optional duration in seconds (analyze this many seconds)
        max_frames: Maximum number of frames FFmpeg will decode (default 30).
                    FFmpeg stops after this many frames via -frames:v.
    Returns:
        list of dicts, each with at least 'YAVG' and 'SATAVG' keys.
    """
    # Build filter chain
    if n_subsample > 1:
        vf = f"select='not(mod(n\\,{n_subsample}))',signalstats,metadata=mode=print"
    else:
        vf = "signalstats,metadata=mode=print"

    # Build seek/duration args (placed before -i for fast input seeking)
    seek_args = []
    if start_time is not None:
        seek_args += ["-ss", str(start_time)]
    if duration is not None:
        seek_args += ["-t", str(duration)]

    # -frames:v tells FFmpeg to stop after processing this many frames
    frame_limit_args = ["-frames:v", str(max_frames)] if max_frames else []

    # Try GPU-accelerated decoding first (NVDEC)
    cmd_gpu = [
        "ffmpeg",
        "-hwaccel", "cuda",
    ] + seek_args + [
        "-i", video_path,
        "-vf", vf,
    ] + frame_limit_args + [
        "-f", "null", "-",
        "-hide_banner", "-loglevel", "info",
    ]

    result = subprocess.run(cmd_gpu, capture_output=True, text=True, timeout=120)

    # If GPU decode failed, fall back to CPU
    if result.returncode != 0:
        logger.info(f"GPU decode failed for signalstats, falling back to CPU decode")
        cmd_cpu = [
            "ffmpeg",
        ] + seek_args + [
            "-i", video_path,
            "-vf", vf,
        ] + frame_limit_args + [
            "-f", "null", "-",
            "-hide_banner", "-loglevel", "info",
        ]
        result = subprocess.run(cmd_cpu, capture_output=True, text=True, timeout=120)

    frames = []
    current_frame = {}

    for line in result.stderr.split('\n'):
        line = line.strip()

        # New frame boundary: "[Parsed_metadata_1 @ 0x...] frame:N pts:M ..."
        # Lines are prefixed by FFmpeg, so we can't use startswith('frame:')
        if 'frame:' in line and 'pts:' in line:
            # Save the previous frame if it has the keys we need
            if current_frame and 'SATAVG' in current_frame and 'YAVG' in current_frame:
                frames.append(current_frame)
            current_frame = {}
        elif 'lavfi.signalstats.' in line:
            # Parse: "lavfi.signalstats.SATAVG=29.800000"
            parts = line.split('lavfi.signalstats.', 1)
            if len(parts) == 2:
                key_val = parts[1]
                if '=' in key_val:
                    key, val = key_val.split('=', 1)
                    try:
                        current_frame[key] = float(val)
                    except ValueError:
                        pass

    # Don't forget the last frame
    if current_frame and 'SATAVG' in current_frame and 'YAVG' in current_frame:
        frames.append(current_frame)

    if not frames:
        logger.warning(
            f"signalstats returned 0 frames for {video_path}. "
            f"FFmpeg returncode={result.returncode}, stderr length={len(result.stderr)}"
        )
    else:
        logger.info(f"signalstats collected {len(frames)} frames for {os.path.basename(video_path)}")

    return frames


def validate_color_channels_ffmpeg(
    ref_stats,
    dist_stats,
    max_gray_ratio=0.50,
    brightness_threshold=20.0,
    ref_sat_threshold=6.0,
    dist_sat_threshold=4.0,
    min_checked_frames=3,
):
    """
    FFmpeg-based color validation using pre-computed signalstats.
    Detects grayscale conversion by comparing per-frame saturation.
    Uses the same logic as validate_color_channels_on_frames:
      - Skip dark reference frames (YAVG < brightness_threshold)
      - Skip naturally grayscale reference frames (SATAVG < ref_sat_threshold)
      - Count distorted frames with low saturation (SATAVG < dist_sat_threshold)
      - Fail if too many distorted frames lost color
    Args:
        ref_stats: Pre-computed signalstats for reference video (from _get_signalstats)
        dist_stats: Pre-computed signalstats for distorted video (from _get_signalstats)
        max_gray_ratio: Max fraction of frames allowed with reduced color
        brightness_threshold: Min YAVG to consider a ref frame
        ref_sat_threshold: Min ref SATAVG to consider frame as colorful
        dist_sat_threshold: Max dist SATAVG to classify as gray
        min_checked_frames: Min evaluated frames for strict ratio validation
    Returns:
        tuple: (is_valid, reason)
    """
    try:

        if not ref_stats or not dist_stats:
            return False, "No frames available for color validation (signalstats)"

        checked_frames = 0
        low_color_frames = 0

        for i, (ref_s, dist_s) in enumerate(zip(ref_stats, dist_stats)):
            ref_yavg = ref_s.get('YAVG', 0)
            ref_satavg = ref_s.get('SATAVG', 0)
            dist_satavg = dist_s.get('SATAVG', 0)

            # Skip very dark reference frames
            if ref_yavg < brightness_threshold:
                continue

            # Skip naturally grayscale reference frames
            if ref_satavg < ref_sat_threshold:
                continue

            checked_frames += 1

            # Count distorted frames with very low saturation
            if dist_satavg < dist_sat_threshold:
                low_color_frames += 1
                logger.debug(
                    f"Frame {i}: reduced saturation detected "
                    f"(ref_sat={ref_satavg:.2f}, dist_sat={dist_satavg:.2f})"
                )

        if checked_frames == 0:
            return True, (
                "No reference frames with sufficient color information were found; "
                "color validation was skipped"
            )

        low_color_ratio = low_color_frames / checked_frames

        logger.info(
            "Color validation (ffmpeg) summary: "
            f"{low_color_frames}/{checked_frames} frames with reduced saturation "
            f"({low_color_ratio:.2%}); allowed maximum {max_gray_ratio:.2%}"
        )

        if low_color_ratio > max_gray_ratio:
            return False, (
                f"Color preservation failed: {low_color_ratio:.1%} of evaluated frames "
                f"exhibit reduced saturation (>{max_gray_ratio:.0%} allowed)"
            )

        return True, (
            f"Color preservation validated: {low_color_ratio:.1%} of evaluated frames "
            f"exhibit reduced saturation (≤{max_gray_ratio:.0%} allowed)"
        )

    except Exception as e:
        logger.error(f"Error during ffmpeg color validation: {e}")
        return False, f"Error during color validation: {str(e)}"


def validate_chroma_quality_ffmpeg(ref_stats, dist_stats, threshold=0.7):
    """
    FFmpeg-based chroma validation using pre-computed signalstats.
    Detects partial UV channel reduction by comparing saturation energy.
    Uses SATAVG² as a proxy for chroma energy (var(U) + var(V)),
    then computes the ratio: dist_energy / ref_energy per frame.
    Corner cases handled:
      - Ref frames with near-zero saturation (naturally B&W) are skipped
      - If all frames are low-energy, validation passes (content is valid B&W)
    Args:
        ref_stats: Pre-computed signalstats for reference video (from _get_signalstats)
        dist_stats: Pre-computed signalstats for distorted video (from _get_signalstats)
        threshold: Minimum acceptable chroma energy ratio (0.0-1.0)
    Returns:
        tuple: (is_valid, reason)
    """
    try:
        if not ref_stats or not dist_stats:
            return False, "No frames available for chroma validation (signalstats)"

        if abs(len(ref_stats) - len(dist_stats)) > FRAME_TOLERANCE:
            return False, (
                f"Frame count mismatch between reference and distorted: "
                f"{len(ref_stats)} vs {len(dist_stats)}"
            )

        chroma_ratios = []
        low_energy_threshold = 1.0  # Min SATAVG to consider (filters flat/black frames)

        for i, (ref_s, dist_s) in enumerate(zip(ref_stats, dist_stats)):
            ref_satavg = ref_s.get('SATAVG', 0)
            dist_satavg = dist_s.get('SATAVG', 0)

            # Skip frames with very low saturation in reference (B&W or flat content)
            if ref_satavg < low_energy_threshold:
                logger.debug(f"Frame {i} has low ref saturation ({ref_satavg:.3f}), skipping")
                continue

            # Use SATAVG² as proxy for chroma energy (analogous to var(U) + var(V))
            ref_energy = ref_satavg ** 2
            dist_energy = dist_satavg ** 2
            chroma_ratio = dist_energy / ref_energy
            chroma_ratios.append(chroma_ratio)
            logger.debug(f"Frame {i} chroma ratio: {chroma_ratio:.3f}")

        if not chroma_ratios:
            logger.info("All frames have low saturation (e.g., B&W content), skipping chroma validation")
            return True, "Chroma validation skipped (low-saturation content)"

        avg_chroma_ratio = float(np.mean(chroma_ratios))
        logger.info(f"Average chroma quality ratio (ffmpeg): {avg_chroma_ratio:.3f}, Threshold: {threshold}")

        if avg_chroma_ratio < threshold:
            return False, (
                f"Chroma quality too low: {avg_chroma_ratio:.3f} < {threshold} "
                f"(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 (ffmpeg): {e}")
        return False, f"Error validating chroma: {str(e)}"

# Calculate ClipIQA+ inspired score
def calculate_clipiqa_plus_score(frames):
    """Calculate ClipIQA+ inspired score for given frames"""
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = load_quality_model().to(device)
    
    # Preprocessing pipeline
    preprocess = transforms.Compose([
        transforms.ToPILImage(),
        transforms.Resize((224, 224)),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
    ])
    
    scores = []
    for frame in frames:
        # Preprocess frame
        frame_tensor = preprocess(frame).unsqueeze(0).to(device)
        
        with torch.no_grad():
            # Extract features using ResNet50
            features = model(frame_tensor)
            
            # Calculate quality metrics
            # 1. Feature magnitude (higher = more complex/rich content)
            feature_magnitude = torch.norm(features, p=2).item()
            
            # 2. Feature variance (higher = more diverse content)
            feature_variance = torch.var(features).item()
            
            # 3. Sharpness estimation using Laplacian variance
            gray_frame = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
            laplacian_var = cv2.Laplacian(gray_frame, cv2.CV_64F).var()
            
            # 4. Contrast estimation
            contrast = np.std(gray_frame)
            
            # Combine metrics with weights (inspired by ClipIQA+ approach)
            # Normalize each metric to [0, 1] range
            norm_magnitude = min(1.0, feature_magnitude / 100.0)
            norm_variance = min(1.0, feature_variance / 1000.0)
            norm_sharpness = min(1.0, laplacian_var / 1000.0)
            norm_contrast = min(1.0, contrast / 100.0)
            
            # Weighted combination (ClipIQA+ inspired weights)
            quality_score = (
                0.3 * norm_magnitude +
                0.25 * norm_variance +
                0.25 * norm_sharpness +
                0.2 * norm_contrast
            )
            
            scores.append(quality_score)
    
    # Return the average score across all frames
    return np.mean(scores)

def calculate_psnr(ref_frame: np.ndarray, dist_frame: np.ndarray) -> float:
    """
    Calculate Peak Signal-to-Noise Ratio (PSNR) between reference and distorted frames.
    Args:
        ref_frame (np.ndarray): The reference video frame.
        dist_frame (np.ndarray): The distorted video frame.
    Returns:
        float: The PSNR value between the reference and distorted frames.
    """
    mse = np.mean((ref_frame - dist_frame) ** 2)
    if mse == 0:
        return 1000  # Maximum PSNR value (perfect similarity)
    return 10 * np.log10((255.0**2) / mse)

def get_sample_frames(ref_cap, dist_cap, total_frames):
    """
    Get sample frames from both reference and distorted videos.
    
    Args:
        ref_cap: Reference video capture
        dist_cap: Distorted video capture
        total_frames: Total number of frames in the videos
        
    Returns:
        tuple: (ref_frames, dist_frames) - Lists of sampled frames
    """
    # Determine how many frames to sample
    frames_to_sample = min(PIEAPP_SAMPLE_COUNT, total_frames)
    
    # Generate a random starting point that ensures we can get consecutive frames
    # without exceeding the total number of frames
    max_start_frame = total_frames - frames_to_sample
    if max_start_frame <= 0:
        start_frame = 0
    else:
        start_frame = random.randint(0, max_start_frame)
    
    logger.info(f"Sampling {frames_to_sample} consecutive frames starting from frame {start_frame}")
    
    # Extract frames from reference video
    ref_frames = []
    ref_cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
    for _ in range(frames_to_sample):
        ret, frame = ref_cap.read()
        if not ret:
            break
        ref_frames.append(frame)
    
    # Extract frames from distorted video
    dist_frames = []
    dist_cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
    for _ in range(frames_to_sample):
        ret, frame = dist_cap.read()
        if not ret:
            break
        dist_frames.append(frame)
    
    return ref_frames, dist_frames

def get_frame_count(video_path):
    """
    Get frame count using ffprobe with optimized approach.
    First tries fast metadata read, falls back to frame counting if needed.
    """
    try:
        # Method 1: Fast metadata read (similar to OpenCV speed)
        cmd = [
            "ffprobe",
            "-v", "error",
            "-select_streams", "v:0",
            "-show_entries", "stream=nb_frames",  # Read from metadata (FAST)
            "-of", "default=nokey=1:noprint_wrappers=1",
            video_path
        ]
        result = subprocess.run(
            cmd, 
            capture_output=True, 
            text=True, 
            check=True, 
            timeout=5  # Fast timeout for metadata read
        )
        
        frame_count_str = result.stdout.strip()
        if frame_count_str and frame_count_str != "N/A":
            frame_count = int(frame_count_str)
            logger.debug(f"ffprobe (fast): {video_path} has {frame_count} frames")
            return frame_count
        
        # Method 2: Frame counting fallback (slower but accurate)
        logger.debug(f"Metadata unavailable, counting frames for {video_path}")
        cmd = [
            "ffprobe",
            "-v", "error",
            "-count_frames",  # Actually count frames (SLOWER)
            "-select_streams", "v:0",
            "-show_entries", "stream=nb_read_frames",
            "-of", "default=nokey=1:noprint_wrappers=1",
            video_path
        ]
        result = subprocess.run(
            cmd, 
            capture_output=True, 
            text=True, 
            check=True, 
            timeout=30  # Longer timeout for counting
        )
        frame_count = int(result.stdout.strip())
        logger.debug(f"ffprobe (counted): {video_path} has {frame_count} frames")
        return frame_count
        
    except subprocess.TimeoutExpired:
        logger.error(f"ffprobe timeout for {video_path}")
        raise Exception(f"Frame count timeout: ffprobe took too long")
    except subprocess.CalledProcessError as e:
        logger.error(f"ffprobe failed for {video_path}: {e.stderr}")
        raise Exception(f"Frame count failed: ffprobe error - {e.stderr}")
    except ValueError as e:
        logger.error(f"ffprobe returned invalid frame count for {video_path}: {e}")
        raise Exception(f"Frame count failed: invalid ffprobe output")
    except Exception as e:
        logger.error(f"Unexpected error with ffprobe for {video_path}: {e}")
        raise Exception(f"Frame count failed: {e}")

def is_valid_video(video_path):
    """
    Check if a video file is valid and can be opened using ffprobe.
    Optimized for speed - only checks basic properties without counting frames.
    
    Returns:
        bool: True if video is valid, False otherwise
    """
    try:
        # First check if file exists and has size
        if not os.path.exists(video_path):
            logger.error(f"Video file does not exist: {video_path}")
            return False
        
        file_size = os.path.getsize(video_path)
        if file_size == 0:
            logger.error(f"Video file is empty: {video_path}")
            return False
        
        if file_size < 1024:  # Less than 1KB is suspicious
            logger.warning(f"Video file is very small ({file_size} bytes): {video_path}")
        
        # Use ffprobe to validate video stream (FAST - no frame counting)
        cmd = [
            "ffprobe",
            "-v", "error",
            "-select_streams", "v:0",
            "-show_entries", "stream=codec_type,width,height,duration",  # Basic properties only
            "-of", "csv=p=0",
            video_path
        ]
        result = subprocess.run(
            cmd, 
            capture_output=True, 
            text=True, 
            check=True, 
            timeout=5  # Reduced timeout since we're not counting frames
        )
        
        # Parse output: format is "video,width,height,duration"
        output = result.stdout.strip()
        if not output:
            logger.error(f"ffprobe returned empty output for {video_path}")
            return False
        
        parts = output.split(',')
        if len(parts) < 3:
            logger.error(f"ffprobe validation failed: incomplete output for {video_path}")
            return False
        
        # Check if it's a video stream
        if parts[0] != "video":
            logger.error(f"ffprobe validation failed: not a video file: {video_path}")
            return False
        
        # Check if width and height are valid
        try:
            width = int(parts[1])
            height = int(parts[2])
            if width <= 0 or height <= 0:
                logger.error(f"Invalid video dimensions: {width}x{height} for {video_path}")
                return False
        except (ValueError, IndexError):
            logger.error(f"Could not parse video dimensions for {video_path}")
            return False
        
        # Optionally check duration if available
        if len(parts) > 3 and parts[3]:
            try:
                duration = float(parts[3])
                if duration <= 0:
                    logger.warning(f"Video has invalid duration: {duration}s for {video_path}")
            except ValueError:
                pass  # Duration not critical for validation
        
        return True
        
    except subprocess.TimeoutExpired:
        logger.error(f"ffprobe validation timeout for {video_path}")
        return False
    except subprocess.CalledProcessError as e:
        logger.error(f"ffprobe validation failed for {video_path}: {e.stderr}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error validating {video_path}: {e}")
        return False
        
    except subprocess.TimeoutExpired:
        logger.error(f"ffprobe validation timeout for {video_path}")
        return False
    except subprocess.CalledProcessError as e:
        logger.error(f"ffprobe validation failed for {video_path}: {e.stderr}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error validating {video_path}: {e}")
        return False

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",
                                    codec_mode: str = "CRF", target_bitrate: float = 10.0):
    """
    Validate that distorted video uses specific encoding settings.
    Args:
        dist_path: Path to the distorted video
        ref_path: Path to the reference video
        task: Task type ("compression" or "upscaling")
        target_codec: Target codec family (av1, hevc, h264, etc.)
        codec_mode: Codec mode (CRF, CBR, or VBR)
        target_bitrate: Target bitrate in Mbps (only validated for CBR and VBR modes)
    """
    try:
        # 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,avg_frame_rate,color_space,color_primaries,color_transfer,bit_rate",
            "-show_entries", "format=tags=encoder",
            "-show_format",
            "-of", "json",
            dist_path
        ]
        
        result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=10)
        info = json.loads(result.stdout)
        
        video_stream = info.get("streams", [{}])[0]
        format_info = info.get("format", {})
        format_tags = format_info.get("tags", {})
        
        codec = video_stream.get("codec_name", "")
        profile = video_stream.get("profile", "")
        level = video_stream.get("level")
        sar = video_stream.get("sample_aspect_ratio", "1:1")
        pix_fmt = video_stream.get("pix_fmt", "")
        width = video_stream.get("width", 0)
        height = video_stream.get("height", 0)
        container = format_info.get("format_name", "")
        encoder_tag = ""

        def parse_fps(fps_str):
            try:
                num, denom = map(float, fps_str.split('/'))
                return num / denom if denom != 0 else 0.0
            except (ValueError, ZeroDivisionError):
                return 0.0

        # Extract both
        r_fps = parse_fps(video_stream.get("r_frame_rate", "0/1"))
        avg_fps = parse_fps(video_stream.get("avg_frame_rate", "0/1"))

        # Get bitrate (prefer stream-level, fallback to format-level)
        bit_rate = video_stream.get("bit_rate") or format_info.get("bit_rate")
        bit_rate_mbps = None
        dist_bpp = None
        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")

        # Bits-per-pixel for distorted video (requires bitrate, fps, width, height)
        try:
            if bit_rate and avg_fps > 0 and width and height:
                dist_bpp = bit_rate / (avg_fps * width * height)
        except Exception:
            pass

        # Colorspace properties
        dist_color_space = video_stream.get("color_space", None)
        dist_color_primaries = video_stream.get("color_primaries", None)
        dist_color_transfer = video_stream.get("color_transfer", None)

        # Check encoder tags
        stream_tags = video_stream.get("tags", {})
        encoder_tag = stream_tags.get("encoder", "").lower() or format_tags.get("encoder", "").lower()

        errors = []

        # Validate bitrate for CBR and VBR modes (skip for CRF mode)
        if codec_mode and codec_mode.upper() in ["CBR", "VBR"]:
            if bit_rate_mbps is None:
                errors.append(f"Bitrate information missing for {codec_mode} mode")
            else:
                # Allow up to 10% difference between target and actual bitrate
                bitrate_tolerance = 0.10  # 10%
                lower_bound = target_bitrate * (1 - bitrate_tolerance)
                upper_bound = target_bitrate * (1 + bitrate_tolerance)

                if bit_rate_mbps > upper_bound:
                    errors.append(
                        f"Bitrate mismatch for {codec_mode} mode: expected {target_bitrate:.2f} Mbps "
                        f"(±10%), got {bit_rate_mbps:.2f} Mbps (allowed values must be less than {upper_bound:.2f} Mbps)"
                    )
                else:
                    logger.info(f"✅ Bitrate validation passed for {codec_mode} mode: {bit_rate_mbps:.2f} Mbps (target: {target_bitrate:.2f} Mbps)")
        elif codec_mode and codec_mode.upper() == "CRF":
            logger.info(f"ℹ️ Bitrate check skipped for CRF mode (quality-based encoding)")
        else:
            logger.warning(f"⚠️ Unknown codec mode: {codec_mode}, skipping bitrate validation")

        ref_fps = get_video_fps(ref_path)
        if abs(avg_fps - ref_fps) > 0.3:
            errors.append(f"FPS must be {ref_fps}, got {avg_fps}")

        if task == "compression":
            ref_width, ref_height = get_video_dimensions(ref_path)
            if width != ref_width or height != ref_height:
                errors.append(f"Resolution must be {ref_width}x{ref_height}, got {width}x{height}")

        # REQUIRED encoding checks
        # 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 not in ["hevc", "av1"]:
            errors.append(f"Codec must be hevc or av1 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}")

        # 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":
            errors.append(f"Pixel format must be yuv420p, got {pix_fmt}")

        ref_bpp = None

        # Colorspace validation
        if ref_path and os.path.exists(ref_path):
            ref_bit_rate = None
            try:
                # Get reference colorspace
                ref_cmd = [
                    "ffprobe", "-v", "error", "-select_streams", "v:0",
                    "-show_entries", "stream=color_space,color_primaries,color_transfer,width,height,r_frame_rate,avg_frame_rate,bit_rate",
                    "-of", "json", ref_path
                ]
                ref_result = subprocess.run(ref_cmd, capture_output=True, text=True, 
                                          timeout=5, check=True)
                ref_info = json.loads(ref_result.stdout)
                ref_stream = ref_info.get("streams", [{}])[0]
                
                ref_color_space = ref_stream.get("color_space", None)
                ref_color_primaries = ref_stream.get("color_primaries", None)
                ref_color_transfer = ref_stream.get("color_transfer", None)
                ref_width = ref_stream.get("width")
                ref_height = ref_stream.get("height")
                ref_fps_raw = ref_stream.get("r_frame_rate")
                ref_avg_fps_raw = ref_stream.get("avg_frame_rate")
                ref_bit_rate = ref_stream.get("bit_rate") or ref_info.get("format", {}).get("bit_rate")
                
                # Colorspace - mismatch affects color accuracy
                if (ref_color_space and dist_color_space and 
                    ref_color_space != dist_color_space):
                    errors.append(f"color_space mismatch, reference: {ref_color_space}, distorted: {dist_color_space}")
                
                # Transfer characteristics - mismatch affects VMAF significantly
                if (ref_color_transfer and dist_color_transfer and 
                    ref_color_transfer != dist_color_transfer):
                    errors.append(f"color_transfer mismatch, reference: {ref_color_transfer}, distorted: {dist_color_transfer}")
                
                # Primaries - less critical but still affects color accuracy
                if (ref_color_primaries and dist_color_primaries and 
                    ref_color_primaries != dist_color_primaries):
                    errors.append(f"color_primaries mismatch, reference: {ref_color_primaries}, distorted: {dist_color_primaries}")

                logger.warning(f"  Ref: space={ref_color_space}, primaries={ref_color_primaries}, transfer={ref_color_transfer}")
                logger.warning(f"  Dist: space={dist_color_space}, primaries={dist_color_primaries}, transfer={dist_color_transfer}")
                    # NOT a hard fail - allows innovation in HDR/Dolby Vision/etc.

                # Bits-per-pixel for reference video
                try:
                    if ref_bit_rate and ref_avg_fps_raw and ref_width and ref_height:
                        num, denom = ref_avg_fps_raw.split('/')
                        ref_fps_val = float(num) / float(denom) if float(denom) != 0 else 0.0
                        if ref_fps_val > 0:
                            ref_bpp = int(ref_bit_rate) / (ref_fps_val * ref_width * ref_height)
                except Exception:
                    pass
                
            except subprocess.CalledProcessError:
                logger.warning("Could not read reference colorspace info")
        
        # SVT-AV1 detection, only logging for now
        is_svtav1 = any(keyword in encoder_tag for keyword in [
            "libsvtav1", "svt-av1", "svtav1"
        ])
        # # AV1 Level validation (FFmpeg uses internal integer levels)
        # av1_level_valid = False
        # if level is not None:
        #     # FFmpeg level 12 ≈ AV1 Level 5.1 (4K), map roughly
        #     # This is internal FFmpeg representation, not spec levels
        #     if isinstance(level, int) and level >= 8:  # Reasonable minimum for HD+
        #         av1_level_valid = True
        #     else:
        #         errors.append(f"AV1 level {level} too low for practical use")
        # else:
        #     # Level missing is acceptable if other params valid
        #     logger.warning("AV1 level not specified in metadata")
        
        # Encoder preference validation (not blocking)
        
        if is_svtav1:
            logger.info(f"✅ SVT-AV1 encoder detected: {encoder_tag}")
        elif encoder_tag:
            other_av1_encoders = ["libaom-av1", "rav1e", "libdav1d"]
            is_known_av1 = any(other in encoder_tag for other in other_av1_encoders)
            if is_known_av1:
                logger.info(f"ℹ️ Known AV1 encoder: {encoder_tag}")
            else:
                logger.warning(f"⚠️ Unknown encoder: {encoder_tag}")
        else:
            logger.warning("⚠️ No encoder tag found")
        
        logger.debug(f"Video analysis: {width}x{height}@{avg_fps}, {codec}/{profile}, "
                    f"container={container}, color_space={dist_color_space}, "
                    f"SVT-AV1={is_svtav1}")
        
        if errors:
            return False, "; ".join(errors)
        
        color_info = f"space={dist_color_space or 'default'}" if dist_color_space else ""
        encoder_status = "SVT-AV1" if is_svtav1 else "Other AV1"

        bit_rate_display = f"{bit_rate_mbps:.2f}" if bit_rate_mbps is not None else "unknown"
        status_parts = [f"Valid encoding ({codec}, {width}x{height}, {color_info}, level {level}, fps {avg_fps}, bitrate {bit_rate_display} Mbps)"]
        if codec_mode:
            status_parts.append(f"mode={codec_mode}")
        if target_bitrate is not None:
            status_parts.append(f"target_bitrate={target_bitrate} Mbps")
        if dist_bpp is not None:
            status_parts.append(f"dist_bpp={dist_bpp:.4f}")
        if ref_bpp is not None:
            status_parts.append(f"ref_bpp={ref_bpp:.4f}")
        return True, "; ".join(status_parts)
        
    except subprocess.CalledProcessError as e:
        return False, f"ffprobe failed: {e.stderr}"
    except Exception as e:
        return False, f"Validation error: {str(e)}"

def get_video_dimensions(video_path):
    """
    Get video dimensions (width, height) using ffprobe.
    This avoids OpenCV's AV1 decoder warnings.
    Returns:
        tuple: (width, height) or (None, None) if failed
    """
    try:
        cmd = [
            "ffprobe",
            "-v", "error",
            "-select_streams", "v:0",
            "-show_entries", "stream=width,height",
            "-of", "csv=p=0",
            video_path
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=5)

        # Filter out empty strings from split (handles trailing commas like "1920,1080,")
        dimensions = [x for x in result.stdout.strip().split(',') if x]

        if len(dimensions) != 2:
            logger.warning(f"ffprobe returned unexpected format for {video_path}: '{result.stdout.strip()}'")
            return None, None

        width, height = map(int, dimensions)
        return width, height
    except Exception as e:
        logger.warning(f"ffprobe dimension check failed for {video_path}: {e}")
        return None, None

def get_video_fps(video_path, original_str=False):
    """
    Get video FPS using ffprobe.
    This avoids OpenCV's AV1 decoder warnings.
    
    Returns:
        float: FPS or None if failed
    """
    try:
        cmd = [
            "ffprobe",
            "-v", "error",
            "-select_streams", "v:0",
            "-show_entries", "stream=avg_frame_rate",
            "-of", "default=noprint_wrappers=1:nokey=1",
            video_path
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=5)
        # avg_frame_rate returns format like "30000/1001" or "30/1"
        fps_str = result.stdout.strip()
        if original_str:
            return fps_str
        if '/' in fps_str:
            num, den = map(int, fps_str.split('/'))
            return num / den
        return float(fps_str)
    except Exception as e:
        logger.warning(f"ffprobe FPS check failed for {video_path}: {e}")
        return None

def calculate_pieapp_score_on_samples(ref_frames, dist_frames):
    """
    Calculate PIE-APP score on sampled frames without creating temporary files.
    
    Args:
        ref_frames: List of reference frames
        dist_frames: List of distorted frames
        
    Returns:
        float: Average PIE-APP score
    """
    if not ref_frames:
        logger.info("No ref frames to process")
        return -100
    
    if not dist_frames:
        logger.info("No dist frames to process")
        return 2.0
    
    class FrameProvider:
        def __init__(self, frames):
            self.frames = frames
            self.current_frame = 0
            self.frame_count = len(frames)
        
        def read(self):
            if self.current_frame < self.frame_count:
                frame = self.frames[self.current_frame]
                self.current_frame += 1
                return True, frame
            return False, None
        
        def get(self, prop_id):
            if prop_id == cv2.CAP_PROP_FRAME_COUNT:
                return self.frame_count
            return 0
        
        def set(self, prop_id, value):
            if prop_id == cv2.CAP_PROP_POS_FRAMES:
                self.current_frame = int(value) if value < self.frame_count else self.frame_count
                return True
            return False
        
        def release(self):
            # Nothing to release
            pass
        
        def isOpened(self):
            return self.frame_count > 0
    
    ref_provider = FrameProvider(ref_frames)
    dist_provider = FrameProvider(dist_frames)
    
    try:
        score = calculate_pieapp_score(ref_provider, dist_provider, frame_interval=1)
        return score
    except Exception as e:
        logger.error(f"Error calculating PieAPP score on frames: {str(e)}")
        return -100

def upscale_video(input_path, scale_factor=2):
    """
    Upscales a video using FFmpeg by the specified scale factor.
    
    Args:
        input_path (str): Path to the input video file
        scale_factor (int): Factor by which to upscale the video (default: 2)
        
    Returns:
        str: Path to the upscaled video file
    """
    filename, extension = os.path.splitext(input_path)
    output_path = f"{filename}_upscaled{extension}"
    
    # Get video dimensions using ffprobe (avoids AV1 warnings)
    width, height = get_video_dimensions(input_path)
    if width is None or height is None:
        raise ValueError(f"Could not get video dimensions for {input_path}")
    
    new_width = width * scale_factor
    new_height = height * scale_factor
    
    cmd = [
        "ffmpeg",
        "-i", input_path,
        "-vf", f"scale={new_width}:{new_height}",
        "-c:v", "libx264",
        "-preset", "medium",  
        "-crf", "18",         
        "-an",
        output_path
    ]
    
    try:
        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        logger.info(f"Successfully upscaled video to {new_width}x{new_height}")
        return output_path
    except subprocess.CalledProcessError as e:
        logger.error(f"Error upscaling video: {e}")
        return input_path

@app.post("/score_upscaling_synthetics")
async def score_upscaling_synthetics(request: UpscalingScoringRequest) -> UpscalingScoringResponse:
    logger.info("#################### 🤖 start upscaling request scoring ####################")

    start_time = time.time()
    quality_scores = []
    length_scores = []
    final_scores = []
    vmaf_scores = []
    pieapp_scores = []
    reasons = []

    if len(request.reference_paths) != len(request.distorted_urls):
        raise HTTPException(
            status_code=400, 
            detail="Number of reference paths must match number of distorted URLs"
        )
    
    if len(request.uids) != len(request.distorted_urls):
        raise HTTPException(
            status_code=400, 
            detail="Number of UIDs must match number of distorted URLs"
        )
    
    if len(request.payload_urls) != len(request.distorted_urls):
        raise HTTPException(
            status_code=400, 
            detail="Number of payload URLs must match number of distorted URLs"
        )

    semaphore = asyncio.Semaphore(15)

    async def _download_distorted_for_duplicate_check(idx: int, uid: int, dist_url: str):
        if not dist_url or len(dist_url) < 10:
            return {"path": None, "final_url": dist_url, "error": "invalid or missing distorted video URL"}
        async with semaphore:
            try:
                path, download_time, final_url = await download_video(
                    dist_url,
                    request.verbose,
                    return_final_url=True,
                )
                logger.info(
                    f"UID {uid}: downloaded distorted video for duplicate check "
                    f"in {download_time:.2f}s | final_url={final_url}"
                )
                return {"path": path, "final_url": final_url, "error": None}
            except Exception as e:
                return {"path": None, "final_url": dist_url, "error": str(e)}

    download_tasks = [
        asyncio.create_task(_download_distorted_for_duplicate_check(idx, uid, dist_url))
        for idx, (uid, dist_url) in enumerate(zip(request.uids, request.distorted_urls))
    ]
    download_results = await asyncio.gather(*download_tasks, return_exceptions=True)
    distorted_download_paths = []
    final_distorted_urls = []
    distorted_download_errors = {}

    for idx, result in enumerate(download_results):
        if isinstance(result, Exception):
            distorted_download_paths.append(None)
            final_distorted_urls.append(request.distorted_urls[idx])
            distorted_download_errors[idx] = str(result)
            continue

        distorted_download_paths.append(result["path"])
        final_distorted_urls.append(result["final_url"])
        if result["error"]:
            distorted_download_errors[idx] = result["error"]

    duplicate_reasons = _duplicate_response_reasons(
        request.uids,
        request.distorted_urls,
        distorted_download_paths,
        final_distorted_urls,
    )

    for idx, (ref_path, dist_url, payload_url, uid, video_id, uploaded_object_name, content_length, task_type) in enumerate(zip(
        request.reference_paths, 
        request.distorted_urls, 
        request.payload_urls,
        request.uids,
        request.video_ids,
        request.uploaded_object_names,
        request.content_lengths,
        request.task_types
    )):
        try:
            logger.info(f"🧩 Processing pair {idx+1}/{len(request.distorted_urls)}: UID {uid} 🧩")
            
            uid_start_time = time.time()  # Start time for this UID

            ref_y4m_path = None
            dist_path = distorted_download_paths[idx] if idx < len(distorted_download_paths) else None
            payload_path = None

            if idx in duplicate_reasons:
                reason = duplicate_reasons[idx]
                logger.error(f"UID {uid}: {reason}. Assigning score of 0.")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append(reason)
                continue

            if idx in distorted_download_errors:
                error_msg = f"Failed to download distorted video from {dist_url}: {distorted_download_errors[idx]}"
                logger.error(f"{error_msg}. Assigning score of 0.")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append(error_msg)
                continue

            scale_factor = 2
            if task_type == "SD24K":
                scale_factor = 4

            logger.info(f"scale factor: {scale_factor}")

            # Validate reference video using ffprobe (avoids AV1 warnings)
            if not is_valid_video(ref_path):
                # Add diagnostic information
                file_exists = os.path.exists(ref_path)
                file_size = os.path.getsize(ref_path) if file_exists else 0
                
                logger.error(f"Error opening reference video file {ref_path}.")
                logger.info(f"  File exists: {file_exists}")
                logger.info(f"  File size: {file_size} bytes")
                logger.info(f"  Current working directory: {os.getcwd()}")
                logger.info(f"  Assigning score of 0.")
                
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(-100)
                reasons.append(f"error opening reference video file: {ref_path} (exists: {file_exists}, size: {file_size})")
                continue

            try:
                payload_path, download_time = await download_video(payload_url, request.verbose)
            except Exception as e:
                error_msg = f"Failed to download payload video from {payload_url}: {str(e)}"
                logger.error(f"{error_msg}. Assigning score of -100, to be skipped while scoring.")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(-100)
                reasons.append(error_msg)
                continue

            is_valid_file_size, max_file_size_multiplier = validate_upscaling_file_size(
                payload_path, dist_path, scale_factor
            )
            if not is_valid_file_size:
                logger.info(
                    f"Distorted video file size exceeds {max_file_size_multiplier}x reference "
                    "file size limit. Penalizing miner."
                )
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append(
                    f"MINER FAILURE: distorted video file size exceeds "
                    f"{max_file_size_multiplier}x reference file size limit"
                )
                continue

            step_time = time.time() - uid_start_time

            # Validate video using ffprobe (avoids AV1 warnings)
            if not is_valid_video(dist_path):
                logger.error(f"Error opening distorted video file from {dist_url}. Assigning score of 0.")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append("error opening distorted video file")
                if dist_path and os.path.exists(dist_path):
                    os.unlink(dist_path)
                continue

            # === RESOLUTION CHECK ===
            dist_width, dist_height = get_video_dimensions(dist_path)
            logger.info(f"Distorted video resolution: {dist_width}x{dist_height}")
            ref_width, ref_height = get_video_dimensions(payload_path)
            logger.info(f"Reference video resolution: {ref_width}x{ref_height}")
            expected_width = ref_width * scale_factor
            expected_height = ref_height * scale_factor
            
            if dist_width != expected_width or dist_height != expected_height:
                logger.info(f"resolution mismatch: expected {expected_width}x{expected_height}, got {dist_width}x{dist_height}. penalizing miner.")
                reasons.append("MINER FAILURE: incorrect upscaling resolution")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)  # 0 = miner penalty
                continue
            logger.info(f"Resolution check passed: expected {expected_width}x{expected_height}, got {dist_width}x{dist_height}.")

            # Only log success after confirming the file was validated
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 1. Validated reference video and miner output in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            ref_total_frames = get_frame_count(ref_path)
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 2. Retrieved reference video frame count ({ref_total_frames}) in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            if ref_total_frames < 10:
                logger.info(f"Video must contain at least 10 frames. Assigning score of 0.")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(-100)
                reasons.append("reference video has fewer than 10 frames")
                continue

            sample_size = min(PIEAPP_SAMPLE_COUNT, ref_total_frames)
            max_start_frame = ref_total_frames - sample_size
            start_frame = 0 if max_start_frame <= 0 else random.randint(0, max_start_frame)

            logger.info(f"Selected frame range for pieapp score {idx+1}: {start_frame} to {start_frame + sample_size - 1}")
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 3. Selected frame range in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            # Open video with OpenCV only when needed for frame extraction
            ref_cap = cv2.VideoCapture(ref_path)
            ref_frames = []
            ref_cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
            for _ in range(sample_size):
                ret, frame = ref_cap.read()
                if not ret:
                    break
                ref_frames.append(frame)
            ref_cap.release()
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 4. Extracted sampled frames from reference 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="upscaling")
            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")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append(f"invalid encoding settings: {encoding_msg}")
                continue
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 5. Validated encoding settings ({encoding_msg}) in {step_time:.2f} seconds.")

            if len(dist_url) < 10:
                logger.info(f"Wrong dist download URL: {dist_url}. Assigning score of 0.")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append("Invalid download URL: the distorted video download URL must be at least 10 characters long.")
                continue

            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 6. Validated distorted video in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            # Get frame count using ffprobe (no AV1 warnings)
            dist_total_frames = get_frame_count(dist_path)

            logger.info(f"Distorted video has {dist_total_frames} frames.")
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 7. Retrieved distorted video frame count in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            if abs(dist_total_frames - ref_total_frames) > FRAME_TOLERANCE:
                logger.info(
                    f"Video length mismatch for pair {idx+1}: ref({ref_total_frames}) != dist({dist_total_frames}) & frame_tolerance({FRAME_TOLERANCE}). Assigning score of 0."
                )
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append("video length mismatch")
                if dist_path and os.path.exists(dist_path):
                    os.unlink(dist_path)
                continue

            # Calculate VMAF — Docker-first with Y4M fallback
            # ref_clip_vmaf_path = None
            # dist_clip_vmaf_path = None
            try:
                vmaf_start = time.time()

                # Trim 1-second clips for VMAF calculation (frame-accurate)
                ref_fps = get_video_fps(ref_path) or 30.0
                # vmaf_num_frames = int(ref_fps)  # 1 second worth of frames
                # max_start_frame = max(0, ref_total_frames - vmaf_num_frames)
                # vmaf_start_frame = random.randint(0, max_start_frame)

                # ref_clip_vmaf_path = trim_video_select(ref_path, vmaf_start_frame, vmaf_num_frames)
                # dist_clip_vmaf_path = trim_video_select(dist_path, vmaf_start_frame, vmaf_num_frames)
                # logger.info(f"Trimmed {vmaf_num_frames} frames starting at frame {vmaf_start_frame} for VMAF calculation")

                if VMAF_FFMPEG_AVAILABLE:
                    try:
                        logger.info("Attempting Docker-based VMAF calculation (libvmaf_cuda)...")
                        vmaf_score = vmaf_metric_ffmpeg(
                            dist_path=dist_path,
                            ref_path=ref_path,
                            skip_frames=np.random.randint(0, ref_fps),
                            n_subsample=np.random.randint(20, 30),
                        )
                        logger.info(f"Docker VMAF calculation succeeded: {vmaf_score}")
                    except Exception as docker_err:
                        logger.warning(f"Docker VMAF failed, falling back to Y4M: {docker_err}")
                        vmaf_score = None  # trigger fallback below

                if not VMAF_FFMPEG_AVAILABLE or vmaf_score is None:
                    # Fallback: Y4M-based VMAF calculation
                    logger.info("Using Y4M-based VMAF calculation (fallback)...")
                    random_frames = sorted(random.sample(range(ref_total_frames), min(ref_total_frames, VMAF_SAMPLE_COUNT)))
                    ref_y4m_path = convert_mp4_to_y4m(ref_path, random_frames)
                    vmaf_score = calculate_vmaf(ref_y4m_path, dist_path, random_frames, neg_model=False)

                vmaf_calc_time = time.time() - vmaf_start
                logger.info(f"☣️☣️ VMAF calculation took {vmaf_calc_time:.2f} seconds.")

                step_time = time.time() - uid_start_time
                logger.info(f"♎️ 8. Completed VMAF calculation in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

                if vmaf_score is not None:
                    vmaf_scores.append(vmaf_score)
                else:
                    vmaf_score = 0.0
                    vmaf_scores.append(vmaf_score)
                logger.info(f"🎾 VMAF score is {vmaf_score}")
            except Exception as e:
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append("failed to calculate VMAF score due to video dimension mismatch")
                logger.error(f"Error calculating VMAF score: {e}")
                continue
            finally:
                pass
                # Clean up trimmed VMAF clips
                # if ref_clip_vmaf_path and os.path.exists(ref_clip_vmaf_path):
                #     os.unlink(ref_clip_vmaf_path)
                # if dist_clip_vmaf_path and os.path.exists(dist_clip_vmaf_path):
                #     os.unlink(dist_clip_vmaf_path)

            if vmaf_score / 100 < VMAF_THRESHOLD:
                logger.info(f"VMAF score is too low, giving zero score, current VMAF score: {vmaf_score}")
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append(f"VMAF score is too low, current VMAF score: {vmaf_score}")
                continue

            # Open video with OpenCV only when needed for frame extraction
            dist_cap = cv2.VideoCapture(dist_path)
            
            # Extract distorted frames
            dist_frames = []
            dist_cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
            for _ in range(sample_size):
                ret, frame = dist_cap.read()
                if not ret:
                    break
                dist_frames.append(frame)
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 11. Extracted sampled frames from distorted video in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            # Calculate PieAPP score
            pieapp_score = calculate_pieapp_score_on_samples(ref_frames, dist_frames)
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 12. Calculated PieAPP score in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")
            logger.info(f"🎾 PieAPP score is {pieapp_score}")

            if pieapp_score == -100:
                logger.error(f"Uncertain error in pieapp calculation")
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(-100)
                reasons.append("Uncertain error in pieapp calculation")
                dist_cap.release()
                continue

            pieapp_scores.append(pieapp_score)
            s_q = calculate_quality_score(pieapp_score)
            logger.info(f"🏀 Quality score is {s_q}")

            dist_cap.release()

            s_l = calculate_length_score(content_length)

            s_pre = calculate_preliminary_score(s_q, s_l)

            s_f = calculate_final_score(s_pre)

            quality_scores.append(s_q)
            length_scores.append(s_l)
            final_scores.append(s_f)
            reasons.append("success")

            step_time = time.time() - uid_start_time
            logger.info(f"🛑 Processed one UID in {step_time:.2f} seconds.")

        except Exception as e:
            error_msg = f"Failed to process video from {dist_url}: {str(e)}"
            logger.error(f"{error_msg}. Assigning score of 0.")
            vmaf_scores.append(0.0)
            pieapp_scores.append(0.0)
            quality_scores.append(0.0)
            length_scores.append(0.0)
            final_scores.append(0.0)
            reasons.append("failed to process video")

        finally:
            # Clean up resources for this pair
            # ref_cap is released immediately after use for frame extraction
            if ref_y4m_path and os.path.exists(ref_y4m_path):
                os.unlink(ref_y4m_path)
            if dist_path and os.path.exists(dist_path):
                os.unlink(dist_path)

    # tmp_directory = "/tmp"
    # try:
    #     logger.info("🧹 Cleaning up temporary files in /tmp...")
    #     for file_path in glob.glob(os.path.join(tmp_directory, "*.mp4")):
    #         os.remove(file_path)
    #         logger.info(f"Deleted: {file_path}")
    #     for file_path in glob.glob(os.path.join(tmp_directory, "*.y4m")):
    #         os.remove(file_path)
    #         logger.info(f"Deleted: {file_path}")
    # except Exception as e:
    #     logger.error(f"⚠️ Error during cleanup: {e}")

    processed_time = time.time() - start_time
    logger.info(f"Completed batch scoring of {len(request.distorted_urls)} pairs within {processed_time:.2f} seconds")
    logger.info(f"🍉🍉🍉 Calculated final scores: {final_scores} 🍉🍉🍉")
    
    return UpscalingScoringResponse(
        vmaf_scores=vmaf_scores,
        pieapp_scores=pieapp_scores,
        quality_scores=quality_scores,
        length_scores=length_scores,
        final_scores=final_scores,
        reasons=reasons
    )

@app.post("/score_compression_synthetics")
async def score_compression_synthetics(request: CompressionScoringRequest) -> CompressionScoringResponse:
    logger.info("#################### 🤖 start compression request scoring ####################")
    start_time = time.time()
    compression_rates = []
    final_scores = []
    vmaf_scores = []
    base_vmaf_scores = []
    reasons = []

    def append_vmaf_scores(vmaf_score, base_vmaf_score=None):
        vmaf_scores.append(vmaf_score)
        base_vmaf_scores.append(base_vmaf_score)

    if len(request.reference_paths) != len(request.distorted_file_paths):
        raise HTTPException(
            status_code=400, 
            detail="Number of reference paths must match number of distorted file paths"
        )
    
    if len(request.uids) != len(request.distorted_file_paths):
        raise HTTPException(
            status_code=400, 
            detail="Number of UIDs must match number of distorted file paths"
        )

    distorted_urls = request.distorted_urls or [""] * len(request.distorted_file_paths)
    if len(distorted_urls) != len(request.distorted_file_paths):
        raise HTTPException(
            status_code=400,
            detail="Number of distorted URLs must match number of distorted file paths",
        )

    final_distorted_urls = await _resolve_final_urls(distorted_urls)
    duplicate_reasons = _duplicate_response_reasons(
        request.uids,
        distorted_urls,
        request.distorted_file_paths,
        final_distorted_urls,
    )

    for idx, (ref_path, dist_path, uid, video_id, uploaded_object_name) in enumerate(zip(
        request.reference_paths, 
        request.distorted_file_paths, 
        request.uids,
        request.video_ids,
        request.uploaded_object_names,
    )):
        try:
            logger.info(f"🧩 Processing pair {idx+1}/{len(request.distorted_file_paths)}: UID {uid} 🧩")
            
            uid_start_time = time.time()  # Start time for this UID
            vmaf_threshold = request.vmaf_thresholds[idx]
            
            ref_y4m_path = None
            dist_y4m_path = None

            if idx in duplicate_reasons:
                reason = duplicate_reasons[idx]
                logger.error(f"UID {uid}: {reason}. Assigning score of 0.")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)
                final_scores.append(0.0)
                reasons.append(reason)
                continue

            # Validate reference video using ffprobe (avoids AV1 warnings)
            if not is_valid_video(ref_path):
                # Add diagnostic information
                file_exists = os.path.exists(ref_path)
                file_size = os.path.getsize(ref_path) if file_exists else 0
                
                logger.error(f"Error opening reference video file {ref_path}.")
                logger.info(f"  File exists: {file_exists}")
                logger.info(f"  File size: {file_size} bytes")
                logger.info(f"  Current working directory: {os.getcwd()}")
                logger.info(f"  Assigning score of 0.")
                
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)   # No compression achieved
                final_scores.append(-100)
                reasons.append(f"error opening reference video file: {ref_path} (exists: {file_exists}, size: {file_size})")
                continue
            
            # Only log success after confirming the file was validated
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 1. Validated reference video in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            ref_total_frames = get_frame_count(ref_path)
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 2. Retrieved reference video frame count ({ref_total_frames}) in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            if ref_total_frames < 10:
                logger.error(f"Video must contain at least 10 frames. Assigning score of 0.")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)   # No compression achieved
                final_scores.append(-100)
                reasons.append("reference video has fewer than 10 frames")
                continue

            # Get reference video file size for compression rate calculation
            ref_file_size = os.path.getsize(ref_path)
            logger.info(f"Reference video file size: {ref_file_size} bytes")

            if not dist_path or not os.path.exists(dist_path):
                logger.error(f"Distorted file path is missing or invalid: {dist_path}. Assigning score of 0.")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)   # No compression achieved
                final_scores.append(0.0)
                reasons.append(f"Invalid or missing distorted video file")
                continue

            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 3. Verified distorted video path in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            # Validate video using ffprobe (avoids AV1 warnings)
            if not is_valid_video(dist_path):
                logger.error(f"Error opening distorted video file from {dist_path}. Assigning score of 0.")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)   # No compression achieved
                final_scores.append(0.0)
                reasons.append("error opening distorted video file")
                continue

            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 (codec must match target_codec)
            target_codec = request.target_codec or "av1"
            codec_mode = request.codec_mode or "CRF"
            target_bitrate = request.target_bitrate or 10.0

            is_valid_encoding, encoding_msg = validate_dist_encoding_settings(
                dist_path, ref_path, task="compression",
                target_codec=target_codec, codec_mode=codec_mode, target_bitrate=target_bitrate
            )
            if not is_valid_encoding:
                logger.error(f"Invalid encoding settings for distorted video {dist_path}: {encoding_msg}")
                logger.info(f"  Required: {target_codec} codec family, proper profile, yuv420p, MP4 container, 1:1 SAR")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999) 
                final_scores.append(0.0)
                reasons.append(f"invalid encoding settings: {encoding_msg}")
                if dist_path and os.path.exists(dist_path):
                    os.unlink(dist_path)
                continue

            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 4.5. Validated encoding settings ({encoding_msg}) in {step_time:.2f} seconds.")

            dist_total_frames = get_frame_count(dist_path)

            logger.info(f"Distorted video has {dist_total_frames} frames.")
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 5. Retrieved distorted video frame count in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            if abs(dist_total_frames - ref_total_frames) > FRAME_TOLERANCE:
                logger.error(
                    f"Video length mismatch for pair {idx+1}: ref({ref_total_frames}) != dist({dist_total_frames}) & frame_tolerance({FRAME_TOLERANCE}). Assigning score of 0."
                )
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)   # No compression achieved
                final_scores.append(0.0)
                reasons.append("video length mismatch")
                if dist_path and os.path.exists(dist_path):
                    os.unlink(dist_path)
                continue

            # Get distorted video file size for compression rate calculation
            dist_file_size = os.path.getsize(dist_path)
            logger.info(f"Distorted video file size: {dist_file_size} bytes")

            # Calculate compression rate BEFORE VMAF calculation
            if ref_file_size > 0 and dist_file_size < ref_file_size:
                compression_rate = dist_file_size / ref_file_size
                logger.info(f"Compression rate: {compression_rate:.4f} ({dist_file_size}/{ref_file_size})")
            else:
                compression_rate = 1.0
                logger.info("Reference file size is 0 or distorted file size >= reference, setting compression rate to 1.0")

            # Calculate VMAF — Docker-first with Y4M fallback
            # Docker path: FFmpeg signalstats for color/chroma validation (no Y4M needed)
            # Fallback path: Y4M-based VMAF + Y4M-based color/chroma validation
            # ref_clip_vmaf_path = None
            # dist_clip_vmaf_path = None
            dist_y4m_path = None
            used_docker_vmaf = False
            try:
                vmaf_start = time.time()
                vmaf_score = None
                base_vmaf_score = None

                # Trim 1-second clips for VMAF calculation (frame-accurate)
                ref_fps_val = get_video_fps(ref_path) or 30.0
                vmaf_num_frames = int(ref_fps_val)  # 1 second worth of frames
                max_start_frame = max(0, ref_total_frames - vmaf_num_frames)
                vmaf_start_frame = random.randint(0, max_start_frame)

                # ref_clip_vmaf_path = trim_video_select(ref_path, vmaf_start_frame, vmaf_num_frames)
                # dist_clip_vmaf_path = trim_video_select(dist_path, vmaf_start_frame, vmaf_num_frames)
                # logger.info(f"Trimmed {vmaf_num_frames} frames starting at frame {vmaf_start_frame} for VMAF calculation")

                if VMAF_FFMPEG_AVAILABLE:
                    try:
                        logger.info("Attempting Docker-based VMAF calculation (libvmaf_cuda)...")
                        vmaf_score, base_vmaf_score = vmaf_metric_ffmpeg(
                            dist_path=dist_path,
                            ref_path=ref_path,
                            neg_model=True,
                            return_base_model_score=True,
                            # skip_frames=np.random.randint(0, ref_fps_val),
                            # n_subsample=np.random.randint(20, 30),
                        )
                        logger.info(
                            f"Docker VMAF NEG calculation succeeded: {vmaf_score}; "
                            f"base VMAF: {base_vmaf_score}"
                        )
                        used_docker_vmaf = True
                    except Exception as docker_err:
                        logger.warning(f"Docker VMAF failed, falling back to Y4M: {docker_err}")
                        vmaf_score = None
                        base_vmaf_score = None

                if vmaf_score is None:
                    # Fallback: Y4M-based VMAF calculation
                    logger.info("Using Y4M-based VMAF calculation (fallback)...")
                    random_frames = sorted(random.sample(range(ref_total_frames), min(ref_total_frames, VMAF_SAMPLE_COUNT)))
                    ref_y4m_path = convert_mp4_to_y4m(ref_path, random_frames)
                    logger.info("The reference video has been successfully converted to Y4M format.")
                    vmaf_score, dist_y4m_path = calculate_vmaf(ref_y4m_path, dist_path, random_frames, neg_model=True, return_y4m_path=True)
                    if vmaf_score is not None and dist_y4m_path:
                        base_vmaf_score = calculate_base_vmaf(ref_y4m_path, dist_y4m_path)

                vmaf_calc_time = time.time() - vmaf_start
                logger.info(f"☣️☣️ VMAF calculation took {vmaf_calc_time:.2f} seconds.")

                step_time = time.time() - uid_start_time
                logger.info(f"♎️ 8. Completed VMAF calculation in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

                if vmaf_score is not None:
                    append_vmaf_scores(vmaf_score, base_vmaf_score)
                else:
                    vmaf_score = 0.0
                    append_vmaf_scores(vmaf_score, base_vmaf_score)
                logger.info(
                    f"🎾 VMAF NEG score is {vmaf_score} , Base VMAF: {base_vmaf_score}, "
                    f"Threshold: {vmaf_threshold}, Diff: {vmaf_score - vmaf_threshold}"
                )
            except Exception as e:
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)   # No compression achieved
                final_scores.append(0.0)
                reasons.append("failed to calculate VMAF score due to video dimension mismatch")
                logger.error(f"Error calculating VMAF score: {e}")
                if dist_y4m_path and os.path.exists(dist_y4m_path):
                    os.unlink(dist_y4m_path)
                continue
            finally:
                pass
                # Clean up trimmed VMAF clips — signalstats runs on originals, not clips
                # if ref_clip_vmaf_path and os.path.exists(ref_clip_vmaf_path):
                #     os.unlink(ref_clip_vmaf_path)
                # if dist_clip_vmaf_path and os.path.exists(dist_clip_vmaf_path):
                #     os.unlink(dist_clip_vmaf_path)

            if base_vmaf_score is not None:
                vmaf_model_delta = abs(base_vmaf_score - vmaf_score)
                logger.info(
                    f"VMAF model delta: {vmaf_model_delta:.4f} "
                    f"(maximum allowed: {MAX_VMAF_MODEL_DELTA:.4f})"
                )
                if vmaf_model_delta > MAX_VMAF_MODEL_DELTA:
                    delta_reason = (
                        f"VMAF model delta {vmaf_model_delta:.4f} exceeds maximum "
                        f"{MAX_VMAF_MODEL_DELTA:.4f}"
                    )
                    logger.error(f"UID {uid}: {delta_reason}. Assigning score of 0.")
                    compression_rates.append(compression_rate)
                    final_scores.append(0.0)
                    reasons.append(f"{delta_reason}; {encoding_msg}")
                    continue

            tone_result = check_tone_manipulation_for_compression(
                ref_path=ref_path,
                dist_path=dist_path,
                timestamp_seconds=(vmaf_start_frame / ref_fps_val) if ref_fps_val else 0.0,
                uid=uid,
                context="synthetics compression",
                duration_seconds=((ref_total_frames - 1) / ref_fps_val) if ref_fps_val else None,
                sample_key=video_id,
            )
            if tone_result and tone_result.get("detected") is True:
                tone_reason = f"Tone manipulation detected: {tone_result['reason']}"
                logger.error(f"UID {uid}: {tone_reason}. Assigning score of 0.")
                compression_rates.append(compression_rate)
                final_scores.append(0.0)
                reasons.append(f"{tone_reason}; {encoding_msg}")
                continue

            # === COLOR / CHROMA VALIDATION ===
            if used_docker_vmaf:
                # Docker path: use FFmpeg signalstats on original videos with time bounds (no Y4M needed)
                logger.info("Using FFmpeg signalstats for color/chroma validation (no Y4M conversion)")

                # Compute signalstats once on original videos (not trimmed clips — avoids keyframe issues)
                # Convert frame-based position back to time for signalstats
                signalstats_start_time = vmaf_start_frame / ref_fps_val
                signalstats_duration = vmaf_num_frames / ref_fps_val
                ref_signal_stats = _get_signalstats(ref_path, start_time=signalstats_start_time, duration=signalstats_duration)
                dist_signal_stats = _get_signalstats(dist_path, start_time=signalstats_start_time, duration=signalstats_duration)

                # Color validation (grayscale detection)
                color_valid, color_reason = validate_color_channels_ffmpeg(ref_signal_stats, dist_signal_stats)
                if not color_valid:
                    logger.error(f"UID {uid}: {color_reason}")
                    compression_rates.append(0.9999)
                    final_scores.append(0.0)
                    reasons.append(f"Color validation failed: {color_reason}")
                    continue

                logger.info(f"✅ UID {uid}: Color channels validated (ffmpeg) - {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.")

                # Chroma validation (UV reduction detection)
                chroma_valid, chroma_reason = validate_chroma_quality_ffmpeg(ref_signal_stats, dist_signal_stats, threshold=0.7)
                if not chroma_valid:
                    logger.error(f"UID {uid}: {chroma_reason}")
                    compression_rates.append(0.9999)
                    final_scores.append(0.0)
                    reasons.append(f"Chroma validation failed: {chroma_reason}")
                    continue

                logger.info(f"✅ UID {uid}: Chroma quality validated (ffmpeg) - {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.")
            else:
                # Fallback path: use Y4M-based validation (existing approach)
                logger.info("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.")

                ref_frames = extract_frames_from_y4m(ref_y4m_path)

                # Validate color channels (grayscale check)
                color_valid, color_reason = validate_color_channels_on_frames(ref_frames, dist_frames)
                if not color_valid:
                    logger.error(f"UID {uid}: {color_reason}")
                    compression_rates.append(0.9999)
                    final_scores.append(0.0)
                    reasons.append(f"Color validation failed: {color_reason}")
                    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.")

                # 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}")
                    compression_rates.append(0.9999)
                    final_scores.append(0.0)
                    reasons.append(f"Chroma validation failed: {chroma_reason}")
                    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

            final_score, compression_component, quality_component, reason = calculate_compression_score(
                vmaf_score=vmaf_score,
                compression_rate=compression_rate,
                vmaf_threshold=vmaf_threshold,
                compression_weight=COMPRESSION_RATE_WEIGHT,
                quality_weight=COMPRESSION_VMAF_WEIGHT,
                soft_threshold_margin=SOFT_THRESHOLD_MARGIN
            )
           

            #  quality component: 
            logger.info(f"VMAF quality component: {quality_component:.4f}")

            # compression component
            logger.info(f"🎯 Compression score is {compression_component:.4f}")

            # final score
            logger.info(f"🎯 Final score is {final_score:.4f}")
            compression_rates.append(compression_rate)
            final_scores.append(final_score)
            reasons.append(f"{reason}; {encoding_msg}")

            step_time = time.time() - uid_start_time
            logger.info(f"🛑 Processed one UID in {step_time:.2f} seconds.")

        except Exception as e:
            error_msg = f"Failed to process video from {dist_path}: {str(e)}"
            logger.error(f"{error_msg}. Assigning score of 0.")
            append_vmaf_scores(0.0)
            compression_rates.append(0.9999)   # No compression achieved
            final_scores.append(0.0)
            reasons.append("failed to process video")

        finally:
            # Clean up resources for this pair
            # ref_cap is not used in this endpoint (using ffprobe instead)
            if ref_y4m_path and os.path.exists(ref_y4m_path):
                os.unlink(ref_y4m_path)
            if dist_y4m_path and os.path.exists(dist_y4m_path):
                os.unlink(dist_y4m_path)

    # tmp_directory = "/tmp"
    # try:
    #     logger.info("🧹 Cleaning up temporary files in /tmp...")
    #     for file_path in glob.glob(os.path.join(tmp_directory, "*.mp4")):
    #         os.remove(file_path)
    #         logger.info(f"Deleted: {file_path}")
    #     for file_path in glob.glob(os.path.join(tmp_directory, "*.y4m")):
    #         os.remove(file_path)
    #         logger.info(f"Deleted: {file_path}")
    # except Exception as e:
    #     logger.error(f"⚠️ Error during cleanup: {e}")

    processed_time = time.time() - start_time
    logger.info(f"Completed batch scoring of {len(request.distorted_file_paths)} pairs within {processed_time:.2f} seconds")
    logger.info(f"🍉🍉🍉 Calculated final scores: {final_scores} 🍉🍉🍉")
    
    return CompressionScoringResponse(
        vmaf_scores=vmaf_scores,
        base_vmaf_scores=base_vmaf_scores,
        compression_rates=compression_rates,
        final_scores=final_scores,
        reasons=reasons
    )

@app.post("/score_organics_upscaling")
async def score_organics_upscaling(request: OrganicsUpscalingScoringRequest) -> OrganicsUpscalingScoringResponse:
    logger.info("#################### 🤖 start scoring ####################")
    batch_start_time = time.time()
    vmaf_scores = []
    pieapp_scores = []
    quality_scores = []
    length_scores = []
    final_scores = []
    reasons = []

    proxy_downloadable_results = []
    distorted_video_paths = []
    for dist_url in request.distorted_urls:
        if len(dist_url) < 10:
            proxy_downloadable_results.append(True)
            distorted_video_paths.append(None)
            continue
        try:
            proxy_downloadable = await verify_organic_proxy_download(dist_url)
        except Exception as e:
            logger.warning(
                f"organic proxy download verification unavailable for {dist_url}; "
                f"bypassing check and continuing scoring. error: {e}"
            )
            proxy_downloadable = True

        proxy_downloadable_results.append(proxy_downloadable)
        if not proxy_downloadable:
            distorted_video_paths.append(None)
            continue

        try:
            path, download_time = await download_video(dist_url, request.verbose)
            distorted_video_paths.append(path)
        except Exception as e:
            logger.error(f"failed to download distorted video: {dist_url}, error: {e}")
            distorted_video_paths.append(None)

    for idx, (ref_url, dist_path, uid, task_type, proxy_downloadable) in enumerate(
        zip(request.reference_urls, distorted_video_paths, request.uids, request.task_types, proxy_downloadable_results)
    ):
        logger.info(f"🧩 processing {uid}.... downloading reference video.... 🧩")
        ref_path = None
        
        ref_upscaled_y4m_path = None
        ref_clip_path = None
        dist_clip_path = None
        ref_upscaled_clip_path = None

        scale_factor = 2
        if task_type == "SD24K":
            scale_factor = 4

        logger.info(f"scale factor: {scale_factor}")

        # === REFERENCE VIDEO VALIDATION (NOT MINER'S FAULT) ===
        try:
            # download reference video
            if len(ref_url) < 10:
                logger.info(f"invalid reference download url: {ref_url}. skipping scoring")
                reasons.append("invalid reference url - skipped")
                vmaf_scores.append(-1)
                pieapp_scores.append(-1)
                quality_scores.append(-1)
                length_scores.append(-1)
                final_scores.append(-1)  # -1 means skip, don't penalize
                continue

            ref_path, download_time = await download_video(ref_url, request.verbose)
            
            # Validate reference video using ffprobe (avoids AV1 warnings)
            if not is_valid_video(ref_path):
                file_exists = os.path.exists(ref_path)
                file_size = os.path.getsize(ref_path) if file_exists else 0
                
                logger.error(f"error opening reference video from {ref_url}. skipping scoring")
                logger.info(f"  File path: {ref_path}")
                logger.info(f"  File exists: {file_exists}")
                logger.info(f"  File size: {file_size} bytes")
                
                reasons.append("corrupted reference video - skipped")
                vmaf_scores.append(-1)
                pieapp_scores.append(-1)
                quality_scores.append(-1)
                length_scores.append(-1)
                final_scores.append(-1)  # -1 means skip, don't penalize
                continue
            
            # Get video metadata using ffprobe (no AV1 warnings)
            ref_total_frames = get_frame_count(ref_path)
            ref_fps = get_video_fps(ref_path)
            ref_width, ref_height = get_video_dimensions(ref_path)
            
            if ref_fps is None or ref_width is None or ref_height is None:
                logger.error(f"failed to get reference video metadata. skipping scoring")
                reasons.append("failed to read reference video metadata - skipped")
                vmaf_scores.append(-1)
                pieapp_scores.append(-1)
                quality_scores.append(-1)
                length_scores.append(-1)
                final_scores.append(-1)
                continue

            if ref_total_frames < 10:
                logger.info(f"reference video too short (<10 frames). skipping scoring")
                reasons.append("reference video too short - skipped")
                vmaf_scores.append(-1)
                pieapp_scores.append(-1)
                quality_scores.append(-1)
                length_scores.append(-1)
                final_scores.append(-1)  # -1 means skip, don't penalize
                continue

        except Exception as e:
            logger.error(f"system error processing reference video: {e}. skipping scoring")
            reasons.append("system error with reference - skipped")
            vmaf_scores.append(-1)
            pieapp_scores.append(-1)
            quality_scores.append(-1)
            length_scores.append(-1)
            final_scores.append(-1)  # -1 means skip, don't penalize
            continue

        # === MINER OUTPUT VALIDATION (MINER'S FAULT) ===
        try:
            if proxy_downloadable is None:
                logger.warning(
                    f"organic proxy download verification unavailable for uid {uid}; "
                    "bypassing check and continuing scoring."
                )
                proxy_downloadable = True

            if not proxy_downloadable:
                logger.error(f"organic proxy failed to download distorted video for uid {uid}. penalizing miner.")
                reasons.append(f"MINER FAILURE: distorted video is not downloadable through organic proxy via endpoint: {ORGANIC_PROXY_VERIFY_DOWNLOAD_ENDPOINT}, miner is recommended to whitelist organic proxy IP in S3 compatible storage")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                continue

            # check if distorted video failed to download
            if dist_path is None:
                logger.error(f"failed to download distorted video for uid {uid}. penalizing miner.")
                reasons.append("MINER FAILURE: failed to download distorted video, invalid url")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)  # 0 = miner penalty
                continue

            is_valid_file_size, max_file_size_multiplier = validate_upscaling_file_size(
                ref_path, dist_path, scale_factor
            )
            if not is_valid_file_size:
                logger.info(
                    f"distorted video file size exceeds {max_file_size_multiplier}x reference "
                    "file size limit. penalizing miner."
                )
                reasons.append(
                    f"MINER FAILURE: distorted video file size exceeds "
                    f"{max_file_size_multiplier}x reference file size limit"
                )
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)  # 0 = miner penalty
                if ref_path and os.path.exists(ref_path):
                    os.unlink(ref_path)
                if dist_path and os.path.exists(dist_path):
                    os.unlink(dist_path)
                continue

            # Check if miner's output can be opened (using ffprobe to avoid AV1 warnings)
            if not is_valid_video(dist_path):
                logger.error(f"error opening distorted video from {dist_path}. penalizing miner.")
                reasons.append("MINER FAILURE: corrupted output video")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)  # 0 = miner penalty
                continue

            is_valid_encoding, encoding_msg = validate_dist_encoding_settings(dist_path, ref_path, task="upscaling")
            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")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append(f"invalid encoding settings: {encoding_msg}")
                continue
            
            # Get frame count and dimensions using ffprobe (no AV1 warnings)
            dist_total_frames = get_frame_count(dist_path)
            dist_width, dist_height = get_video_dimensions(dist_path)
            
            if dist_width is None or dist_height is None:
                logger.error(f"failed to get video dimensions for {dist_path}. penalizing miner.")
                reasons.append("MINER FAILURE: cannot read video dimensions")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                continue
                
            logger.info(f"distorted video has {dist_total_frames} frames.")

            # Check frame count match
            if dist_total_frames != ref_total_frames:
                logger.info(f"video length mismatch: ref({ref_total_frames}) != dist({dist_total_frames}). penalizing miner.")
                reasons.append("MINER FAILURE: video length mismatch")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)  # 0 = miner penalty
                continue

            # Check minimum frame count for miner output
            if dist_total_frames < 10:
                logger.info(f"miner output too short (<10 frames). penalizing miner.")
                reasons.append("MINER FAILURE: output video too short")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)  # 0 = miner penalty
                continue

            # === RESOLUTION CHECK ===
            expected_width = ref_width * scale_factor
            expected_height = ref_height * scale_factor
            
            if dist_width != expected_width or dist_height != expected_height:
                logger.info(f"resolution mismatch: expected {expected_width}x{expected_height}, got {dist_width}x{dist_height}. penalizing miner.")
                reasons.append("MINER FAILURE: incorrect upscaling resolution")
                vmaf_scores.append(0.0)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)  # 0 = miner penalty
                continue

        except Exception as e:
            logger.error(f"error validating miner output: {e}. penalizing miner.")
            reasons.append(f"MINER FAILURE: {e}")
            vmaf_scores.append(0.0)
            pieapp_scores.append(0.0)
            quality_scores.append(0.0)
            length_scores.append(0.0)
            final_scores.append(0.0)  # 0 = miner penalty
            continue

        # === MAIN SCORING LOGIC ===
        try:
            # Calculate 0.25 second clip using frame-accurate extraction
            clip_num_frames = max(1, int(ref_fps * 0.25))  # 0.25 seconds worth of frames
            max_start_frame = max(0, ref_total_frames - clip_num_frames)
            start_frame = random.randint(0, max_start_frame)

            logger.info(f"Creating {clip_num_frames}-frame clips starting at frame {start_frame}")

            # Frame-accurate extraction for both ref and dist
            ref_clip_path = trim_video_select(ref_path, start_frame, clip_num_frames)
            logger.info(f"Created reference clip: {ref_clip_path}")

            dist_clip_path = trim_video_select(dist_path, start_frame, clip_num_frames)
            logger.info(f"Created distorted clip: {dist_clip_path}")
            
            # Upscale the reference clip
            ref_upscaled_clip_path = upscale_video(ref_clip_path, scale_factor)
            logger.info(f"Created upscaled reference clip: {ref_upscaled_clip_path}")
            
            # Calculate ClipIQ scores
            logger.info("Calculating ClipIQ scores...")
            ref_clipiqa_score = get_clipiqa_score(ref_clip_path, num_frames=3)
            dist_clipiqa_score = get_clipiqa_score(dist_clip_path, num_frames=3)
            ref_upscaled_clipiqa_score = get_clipiqa_score(ref_upscaled_clip_path, num_frames=3)
            
            logger.info(f"Reference ClipIQ score: {ref_clipiqa_score:.4f}")
            logger.info(f"Distorted ClipIQ score: {dist_clipiqa_score:.4f}")
            logger.info(f"Upscaled reference ClipIQ score: {ref_upscaled_clipiqa_score:.4f}")
            
            # Calculate VMAF — Docker-first with Y4M fallback
            logger.info("Calculating VMAF score...")
            vmaf_score = None

            if VMAF_FFMPEG_AVAILABLE and ref_upscaled_clip_path and dist_clip_path:
                try:
                    logger.info("Attempting Docker-based VMAF calculation (libvmaf_cuda)...")
                    vmaf_score = vmaf_metric_ffmpeg(
                        dist_path=dist_clip_path,
                        ref_path=ref_upscaled_clip_path,
                        skip_frames=0,
                        n_subsample=1,
                    )
                    logger.info(f"Docker VMAF calculation succeeded: {vmaf_score}")
                except Exception as docker_err:
                    logger.warning(f"Docker VMAF failed, falling back to Y4M: {docker_err}")
                    vmaf_score = None

            if vmaf_score is None:
                # Fallback: Y4M-based VMAF calculation
                logger.info("Using Y4M-based VMAF calculation (fallback)...")
                ref_clip_frames = get_frame_count(ref_upscaled_clip_path)
                random_frames = sorted(random.sample(range(ref_clip_frames), min(ref_clip_frames, VMAF_SAMPLE_COUNT)))
                ref_y4m_path = convert_mp4_to_y4m(ref_upscaled_clip_path, random_frames)
                vmaf_score = calculate_vmaf(ref_y4m_path, dist_clip_path, random_frames, neg_model=False)
            
            if vmaf_score is None:
                vmaf_score = 0.0
            
            logger.info(f"VMAF score: {vmaf_score:.4f}")
            
            # Check VMAF threshold
            if vmaf_score < 50:
                logger.info(f"VMAF score below threshold (50): {vmaf_score}. Assigning score 0.")
                vmaf_scores.append(vmaf_score)
                pieapp_scores.append(0.0)
                quality_scores.append(0.0)
                length_scores.append(0.0)
                final_scores.append(0.0)
                reasons.append(f"VMAF score too low: {vmaf_score}")
                continue
            
            # Final scoring based on ClipIQ scores
            if dist_clipiqa_score > ref_clipiqa_score:
                final_score = 3
                reason = "upscaled better than reference"
            elif dist_clipiqa_score > (ref_upscaled_clipiqa_score * 1.05):
                final_score = 2
                reason = "upscaled at least 5% better than ffmpeg upscaled"
            elif dist_clipiqa_score > (ref_upscaled_clipiqa_score):
                final_score = 1
                reason = "Better than ffmpeg upscaled"
            else:
                final_score = 0
                reason = "Worse than ffmpeg upscaled"
            
            logger.info(f"Final score: {final_score} ({reason})")
            
            vmaf_scores.append(vmaf_score)
            pieapp_scores.append(0.0)  
            quality_scores.append(0.0)  
            length_scores.append(0.0)
            final_scores.append(final_score)
            reasons.append(f"{reason}")

        except Exception as e:
            logger.error(f"error in main scoring logic: {e}. penalizing miner.")
            reasons.append(f"MINER FAILURE: scoring error - {e}")
            vmaf_scores.append(0.0)
            pieapp_scores.append(0.0)
            quality_scores.append(0.0)
            length_scores.append(0.0)
            final_scores.append(0.0)  # 0 = miner penalty

        finally:
            # Clean up resources
            
            if ref_path and os.path.exists(ref_path):
                os.unlink(ref_path)
            if dist_path and os.path.exists(dist_path):
                os.unlink(dist_path)
            if ref_upscaled_y4m_path and os.path.exists(ref_upscaled_y4m_path):
                os.unlink(ref_upscaled_y4m_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 ref_upscaled_clip_path and os.path.exists(ref_upscaled_clip_path):
                os.unlink(ref_upscaled_clip_path)

    processed_time = time.time() - batch_start_time
    logger.info(f"🍉🍉🍉 calculated scores: {final_scores} 🍉🍉🍉")
    logger.info(f"completed one batch scoring within {processed_time:.2f} seconds")
    
    # Summary statistics
    successful_miners = sum(1 for score in final_scores if score >= 1)
    failed_miners = sum(1 for score in final_scores if score == 0.0)
    skipped_miners = sum(1 for score in final_scores if score == -1)
    
    logger.info(f"📊 SCORING SUMMARY:")
    logger.info(f"  ✅ Successful miners: {successful_miners}")
    logger.info(f"  ❌ Failed miners: {failed_miners}")
    logger.info(f"  ⏭️ Skipped miners: {skipped_miners}")
    
    return OrganicsUpscalingScoringResponse(
        vmaf_scores=vmaf_scores,
        pieapp_scores=pieapp_scores,
        quality_scores=quality_scores,
        length_scores=length_scores,
        final_scores=final_scores,
        reasons=reasons
    )

@app.post("/score_organics_compression")
async def score_organics_compression(request: OrganicsCompressionScoringRequest) -> OrganicsCompressionScoringResponse:
    logger.info("#################### 🤖 start scoring ####################")
    start_time = time.time()
    vmaf_scores = []
    base_vmaf_scores = []
    compression_rates = []
    final_scores = []
    reasons = []

    def append_vmaf_scores(vmaf_score, base_vmaf_score=None):
        vmaf_scores.append(vmaf_score)
        base_vmaf_scores.append(base_vmaf_score)

    proxy_downloadable_results = []
    distorted_video_paths = []
    for dist_url in request.distorted_urls:
        if len(dist_url) < 10:
            proxy_downloadable_results.append(True)
            distorted_video_paths.append(None)
            continue
        try:
            proxy_downloadable = await verify_organic_proxy_download(dist_url)
        except Exception as e:
            logger.warning(
                f"organic proxy download verification unavailable for {dist_url}; "
                f"bypassing check and continuing scoring. error: {e}"
            )
            proxy_downloadable = True

        proxy_downloadable_results.append(proxy_downloadable)
        if not proxy_downloadable:
            distorted_video_paths.append(None)
            continue

        try:
            path, download_time = await download_video(dist_url, request.verbose)
            distorted_video_paths.append(path)
        except Exception as e:
            logger.error(f"failed to download distorted video: {dist_url}, error: {e}")
            distorted_video_paths.append(None)

    # Get target_codecs, codec_modes, and target_bitrates lists, defaulting if not provided
    target_codecs = request.target_codecs if request.target_codecs else ["av1"] * len(request.uids)
    codec_modes = request.codec_modes if request.codec_modes else ["CRF"] * len(request.uids)
    target_bitrates = request.target_bitrates if request.target_bitrates else [10.0] * len(request.uids)

    for idx, (ref_url, dist_path, uid, vmaf_threshold, target_codec, codec_mode, target_bitrate, proxy_downloadable) in enumerate(
        zip(request.reference_urls, distorted_video_paths, request.uids, request.vmaf_thresholds,
            target_codecs, codec_modes, target_bitrates, proxy_downloadable_results)
    ):
        logger.info(f"🧩 processing {uid}.... downloading reference video.... 🧩")
        ref_path = None
        
        ref_y4m_path = None
        dist_y4m_path = None

        uid_start_time = time.time() # start time for each uid

        # === REFERENCE VIDEO VALIDATION (NOT MINER'S FAULT) ===
        try:
            # download reference video
            if len(ref_url) < 10:
                logger.info(f"invalid reference download url: {ref_url}. skipping scoring")
                append_vmaf_scores(-1)
                compression_rates.append(-1)
                reasons.append("invalid reference url - skipped")
                final_scores.append(-1)  # -1 means skip, don't penalize
                continue

            ref_path, download_time = await download_video(ref_url, request.verbose)
            
            # Validate reference video using ffprobe (avoids AV1 warnings)
            if not is_valid_video(ref_path):
                file_exists = os.path.exists(ref_path)
                file_size = os.path.getsize(ref_path) if file_exists else 0
                
                logger.error(f"error opening reference video from {ref_url}. skipping scoring")
                logger.info(f"  File path: {ref_path}")
                logger.info(f"  File exists: {file_exists}")
                logger.info(f"  File size: {file_size} bytes")
                
                append_vmaf_scores(-1)
                compression_rates.append(-1)
                reasons.append("corrupted reference video - skipped")
                final_scores.append(-1)  # -1 means skip, don't penalize
                continue

            ref_total_frames = get_frame_count(ref_path)

            # Get reference video file size for compression rate calculation
            ref_file_size = os.path.getsize(ref_path)
            logger.info(f"Reference video file size: {ref_file_size} bytes")

            if ref_total_frames < 10:
                logger.info(f"reference video too short (<10 frames). skipping scoring")
                append_vmaf_scores(-1)
                compression_rates.append(-1)
                reasons.append("reference video too short - skipped")
                final_scores.append(-1)  # -1 means skip, don't penalize
                continue

        except Exception as e:
            logger.error(f"system error processing reference video: {e}. skipping scoring")
            append_vmaf_scores(-1)
            compression_rates.append(-1)
            reasons.append("system error with reference - skipped")
            final_scores.append(-1)  # -1 means skip, don't penalize
            continue

        # === MINER OUTPUT VALIDATION (MINER'S FAULT) ===
        try:
            if proxy_downloadable is None:
                logger.warning(
                    f"organic proxy download verification unavailable for uid {uid}; "
                    "bypassing check and continuing scoring."
                )
                proxy_downloadable = True

            if not proxy_downloadable:
                logger.error(f"organic proxy failed to download distorted video for uid {uid}. penalizing miner.")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)
                reasons.append(f"MINER FAILURE: distorted video is not downloadable through organic proxy via endpoint: {ORGANIC_PROXY_VERIFY_DOWNLOAD_ENDPOINT}, miner is recommended to whitelist organic proxy IP in S3 compatible storage")
                final_scores.append(0.0)
                continue

            # check if distorted video failed to download
            if dist_path is None:
                logger.error(f"failed to download distorted video for uid {uid}. penalizing miner.")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)
                reasons.append("MINER FAILURE: failed to download distorted video, invalid url")
                final_scores.append(0.0)  # 0 = miner penalty
                continue

            # Check if miner's output can be opened (using ffprobe to avoid AV1 warnings)
            if not is_valid_video(dist_path):
                logger.error(f"error opening distorted video from {dist_path}. penalizing miner.")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)
                reasons.append("MINER FAILURE: corrupted output video")
                final_scores.append(0.0)  # 0 = miner penalty
                continue

            is_valid_encoding, encoding_msg = validate_dist_encoding_settings(
                dist_path, ref_path, task="compression",
                target_codec=target_codec, codec_mode=codec_mode, target_bitrate=target_bitrate
            )
            if not is_valid_encoding:
                logger.error(f"Invalid encoding settings for distorted video {dist_path}: {encoding_msg}")
                logger.info(f"  Required: {target_codec} codec family, proper profile, yuv420p, MP4 container, 1:1 SAR, {codec_mode} mode")
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)
                final_scores.append(0.0)
                reasons.append(f"invalid encoding settings: {encoding_msg}")
                if dist_path and os.path.exists(dist_path):
                    os.unlink(dist_path)
                continue

            dist_total_frames = get_frame_count(dist_path)
            logger.info(f"Distorted video has {dist_total_frames} frames.")
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 5. Retrieved distorted video frame count in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            if abs(dist_total_frames - ref_total_frames) > FRAME_TOLERANCE:
                logger.error(
                    f"Video length mismatch for pair {idx+1}: ref({ref_total_frames}) != dist({dist_total_frames}) & frame_tolerance({FRAME_TOLERANCE}). Assigning score of 0."
                )
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)  # No compression achieved
                final_scores.append(0.0)
                reasons.append("video length mismatch")
                if dist_path and os.path.exists(dist_path):
                    os.unlink(dist_path)
                continue

            # Get distorted video file size for compression rate calculation
            dist_file_size = os.path.getsize(dist_path)
            logger.info(f"Distorted video file size: {dist_file_size} bytes")

            # Calculate compression rate BEFORE VMAF calculation
            if ref_file_size > 0 and dist_file_size < ref_file_size:
                compression_rate = dist_file_size / ref_file_size
                logger.info(f"Compression rate: {compression_rate:.4f} ({dist_file_size}/{ref_file_size})")
            else:
                compression_rate = 1.0
                logger.info("Reference file size is 0 or distorted file size >= reference, setting compression rate to 1.0")

            # Score VMAF on the full videos. Keep a short bounded window only
            # for the existing color/chroma validation checks.
            ref_fps = get_video_fps(ref_path)
            logger.info(f"Reference video FPS: {ref_fps}")

            color_check_num_frames = max(1, int(ref_fps * 0.5))
            max_color_check_start_frame = max(0, ref_total_frames - color_check_num_frames)
            color_check_start_frame = random.randint(0, max_color_check_start_frame)

            logger.info(
                f"Selected {color_check_num_frames}-frame window starting at frame "
                f"{color_check_start_frame} for color/chroma validation"
            )
            step_time = time.time() - uid_start_time
            logger.info(f"♎️ 6. Selected validation frame window in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

            # Calculate VMAF — Docker-first with Y4M fallback
            dist_y4m_path = None
            try:
                vmaf_start = time.time()
                vmaf_score = None
                base_vmaf_score = None

                if VMAF_FFMPEG_AVAILABLE:
                    try:
                        logger.info("Attempting Docker-based VMAF calculation (libvmaf_cuda)...")
                        vmaf_score, base_vmaf_score = vmaf_metric_ffmpeg(
                            dist_path=dist_path,
                            ref_path=ref_path,
                            skip_frames=0,
                            n_subsample=1,
                            neg_model=True,
                            return_base_model_score=True,
                        )
                        logger.info(
                            f"Docker VMAF NEG calculation succeeded: {vmaf_score}; "
                            f"base VMAF: {base_vmaf_score}"
                        )
                    except Exception as docker_err:
                        logger.warning(f"Docker VMAF failed, falling back to Y4M: {docker_err}")
                        vmaf_score = None
                        base_vmaf_score = None

                if vmaf_score is None:
                    # Fallback: Y4M-based VMAF calculation
                    logger.info("Using Y4M-based VMAF calculation (fallback)...")
                    ref_y4m_path = convert_mp4_to_y4m(ref_path)
                    logger.info("The full reference video has been successfully converted to Y4M format.")
                    step_time = time.time() - uid_start_time
                    logger.info(f"♎️ 9. Converted full reference video to Y4M in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")
                    vmaf_score, dist_y4m_path = calculate_vmaf(ref_y4m_path, dist_path, None, neg_model=True, return_y4m_path=True)
                    if vmaf_score is not None and dist_y4m_path:
                        base_vmaf_score = calculate_base_vmaf(ref_y4m_path, dist_y4m_path)

                vmaf_calc_time = time.time() - vmaf_start
                logger.info(f"☣️☣️ VMAF calculation took {vmaf_calc_time:.2f} seconds.")

                step_time = time.time() - uid_start_time
                logger.info(f"♎️ 10. Completed VMAF calculation in {step_time:.2f} seconds. Total time: {step_time:.2f} seconds.")

                if vmaf_score is not None:
                    append_vmaf_scores(vmaf_score, base_vmaf_score)
                else:
                    vmaf_score = 0.0
                    append_vmaf_scores(vmaf_score, base_vmaf_score)
                logger.info(f"🎾 VMAF NEG score is {vmaf_score}; Base VMAF: {base_vmaf_score}")
            except Exception as e:
                append_vmaf_scores(0.0)
                compression_rates.append(0.9999)   # No compression achieved
                final_scores.append(0.0)
                reasons.append("failed to calculate VMAF score due to video dimension mismatch")
                logger.error(f"Error calculating VMAF score: {e}")
                if dist_y4m_path and os.path.exists(dist_y4m_path):
                    os.unlink(dist_y4m_path)
                continue

            if base_vmaf_score is not None:
                vmaf_model_delta = abs(base_vmaf_score - vmaf_score)
                logger.info(
                    f"VMAF model delta: {vmaf_model_delta:.4f} "
                    f"(maximum allowed: {MAX_VMAF_MODEL_DELTA:.4f})"
                )
                if vmaf_model_delta > MAX_VMAF_MODEL_DELTA:
                    delta_reason = (
                        f"VMAF model delta {vmaf_model_delta:.4f} exceeds maximum "
                        f"{MAX_VMAF_MODEL_DELTA:.4f}"
                    )
                    logger.error(f"UID {uid}: {delta_reason}. Assigning score of 0.")
                    compression_rates.append(compression_rate)
                    final_scores.append(0.0)
                    reasons.append(f"{delta_reason}; {encoding_msg}")
                    continue

            tone_result = check_tone_manipulation_for_compression(
                ref_path=ref_path,
                dist_path=dist_path,
                timestamp_seconds=(color_check_start_frame / ref_fps) if ref_fps else 0.0,
                uid=uid,
                context="organics compression",
                duration_seconds=((ref_total_frames - 1) / ref_fps) if ref_fps else None,
                sample_key=ref_url,
            )
            if tone_result and tone_result.get("detected") is True:
                tone_reason = f"Tone manipulation detected: {tone_result['reason']}"
                logger.error(f"UID {uid}: {tone_reason}. Assigning score of 0.")
                compression_rates.append(compression_rate)
                final_scores.append(0.0)
                reasons.append(f"{tone_reason}; {encoding_msg}")
                continue

            # === COLOR / CHROMA VALIDATION ===
            # Use a bounded window so these guard checks remain fast while VMAF
            # itself is scored over the full videos above.
            logger.info("Using FFmpeg signalstats for bounded color/chroma validation")
            signalstats_start_time = color_check_start_frame / ref_fps
            signalstats_duration = color_check_num_frames / ref_fps
            ref_signal_stats = _get_signalstats(ref_path, start_time=signalstats_start_time, duration=signalstats_duration)
            dist_signal_stats = _get_signalstats(dist_path, start_time=signalstats_start_time, duration=signalstats_duration)

            # Color validation (grayscale detection)
            color_valid, color_reason = validate_color_channels_ffmpeg(ref_signal_stats, dist_signal_stats)
            if not color_valid:
                logger.error(f"UID {uid}: {color_reason}")
                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)
                continue

            logger.info(f"✅ UID {uid}: Color channels validated (ffmpeg) - {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.")

            # Chroma validation (UV reduction detection)
            chroma_valid, chroma_reason = validate_chroma_quality_ffmpeg(ref_signal_stats, dist_signal_stats, threshold=0.7)
            if not chroma_valid:
                logger.error(f"UID {uid}: {chroma_reason}")
                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)
                continue

            logger.info(f"✅ UID {uid}: Chroma quality validated (ffmpeg) - {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

            final_score, compression_component, quality_component, reason = calculate_compression_score(
                vmaf_score=vmaf_score,
                compression_rate=compression_rate,
                vmaf_threshold=vmaf_threshold,
                compression_weight=COMPRESSION_RATE_WEIGHT,
                quality_weight=COMPRESSION_VMAF_WEIGHT,
                soft_threshold_margin=SOFT_THRESHOLD_MARGIN
            )


            #  quality component: 
            logger.info(f"VMAF quality component: {quality_component:.4f}")

            # compression component
            logger.info(f"🎯 Compression score is {compression_component:.4f}")

            # final score
            logger.info(f"🎯 Final score is {final_score:.4f}")
            # vmaf_scores.append(vmaf_score)
            compression_rates.append(compression_rate)
            final_scores.append(final_score)
            reasons.append(f"{reason}; {encoding_msg}")

            step_time = time.time() - uid_start_time
            logger.info(f"🛑 Processed one UID in {step_time:.2f} seconds.")

        except Exception as e:
            error_msg = f"Failed to process video from {dist_url}: {str(e)}"
            logger.error(f"{error_msg}. Assigning score of 0.")
            append_vmaf_scores(0.0)
            compression_rates.append(0.9999)   # No compression achieved
            final_scores.append(0.0)
            reasons.append("failed to process video")

        finally:
            # Clean up resources for this pair
            # ref_cap is not used in this endpoint (using ffprobe instead)
            # dist_cap is not used in this endpoint (using ffprobe instead)
            if ref_path and os.path.exists(ref_path):
                os.unlink(ref_path)
            if dist_path and os.path.exists(dist_path):
                os.unlink(dist_path)
            if ref_y4m_path and os.path.exists(ref_y4m_path):
                os.unlink(ref_y4m_path)
            if dist_y4m_path and os.path.exists(dist_y4m_path):
                os.unlink(dist_y4m_path)

    return OrganicsCompressionScoringResponse(
        vmaf_scores=vmaf_scores,
        base_vmaf_scores=base_vmaf_scores,
        compression_rates=compression_rates,
        final_scores=final_scores,
        reasons=reasons
    )

if __name__ == "__main__":
    import uvicorn
    import argparse
    
    parser = argparse.ArgumentParser(description="Vidaio Subnet Scoring Server")
    parser.add_argument("--port", type=int, default=CONFIG.score.port, help="Port to run the scoring server on")
    args = parser.parse_args()
    
    host = CONFIG.score.host
    port = args.port
    
    logger.info(f"Starting scoring server on {host}:{port}")
    uvicorn.run(app, host=host, port=port)