424 lines
18 KiB
Python
424 lines
18 KiB
Python
# core/video_handler.py
|
|
"""Video resolution detection and encoding logic."""
|
|
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from core.logger_helper import setup_logger
|
|
|
|
logger = setup_logger(Path(__file__).parent.parent / "logs")
|
|
|
|
|
|
def get_source_resolution(input_file: Path) -> tuple:
|
|
"""
|
|
Get source video resolution (width, height).
|
|
Returns tuple: (width, height)
|
|
Skips attached pictures and cover art.
|
|
"""
|
|
try:
|
|
# First, get all video streams and their disposition to find the first non-attached pic
|
|
cmd = [
|
|
"ffprobe", "-v", "error",
|
|
"-select_streams", "v",
|
|
"-show_entries", "stream=width,height,disposition",
|
|
"-of", "default=noprint_wrappers=1",
|
|
str(input_file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', check=False)
|
|
|
|
if result.stdout:
|
|
lines = result.stdout.strip().split("\n")
|
|
# Parse the output to find a non-attached picture video stream
|
|
width = None
|
|
height = None
|
|
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
if line.startswith("width="):
|
|
width_val = int(line.split("=")[1]) if "=" in line else None
|
|
# Look ahead for height and disposition
|
|
height_val = None
|
|
is_attached_pic = False
|
|
|
|
if i + 1 < len(lines):
|
|
next_line = lines[i + 1].strip()
|
|
if next_line.startswith("height="):
|
|
height_val = int(next_line.split("=")[1]) if "=" in next_line else None
|
|
if i + 2 < len(lines):
|
|
disp_line = lines[i + 2].strip()
|
|
if disp_line.startswith("disposition="):
|
|
# Check if attached_pic flag is set to 1
|
|
if "attached_pic=1" in disp_line:
|
|
is_attached_pic = True
|
|
|
|
# If this is a real video stream (not attached pic) and has valid dimensions, use it
|
|
if width_val and height_val and not is_attached_pic:
|
|
width = width_val
|
|
height = height_val
|
|
return (width, height)
|
|
|
|
i += 1
|
|
|
|
# Fallback: if no valid stream found, try simple v:0 selection
|
|
if not width or not height:
|
|
logger.debug("No non-attached-pic video stream found, trying fallback method")
|
|
cmd = [
|
|
"ffprobe", "-v", "error",
|
|
"-select_streams", "v:0",
|
|
"-show_entries", "stream=width,height",
|
|
"-of", "default=noprint_wrappers=1:nokey=1:noprint_wrappers=1",
|
|
str(input_file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', check=False)
|
|
if result.stdout:
|
|
lines = result.stdout.strip().split("\n")
|
|
width = int(lines[0]) if len(lines) > 0 and lines[0].strip() else 1920
|
|
height = int(lines[1]) if len(lines) > 1 and lines[1].strip() else 1080
|
|
return (width, height)
|
|
|
|
logger.warning(f"ffprobe returned no output for {input_file.name}. Defaulting to 1920x1080")
|
|
return (1920, 1080)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to detect source resolution: {e}. Defaulting to 1920x1080")
|
|
return (1920, 1080)
|
|
|
|
|
|
def get_source_bit_depth(input_file: Path) -> int:
|
|
"""
|
|
Detect source video bit depth (8, 10, or 12).
|
|
Returns: 12, 10, or 8 (default)
|
|
Skips attached pictures and cover art.
|
|
"""
|
|
try:
|
|
# Get all video streams with pixel format and disposition
|
|
cmd = [
|
|
"ffprobe", "-v", "error",
|
|
"-select_streams", "v",
|
|
"-show_entries", "stream=pix_fmt,disposition",
|
|
"-of", "default=noprint_wrappers=1",
|
|
str(input_file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', check=False)
|
|
|
|
if result.stdout:
|
|
lines = result.stdout.strip().split("\n")
|
|
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
if line.startswith("pix_fmt="):
|
|
pix_fmt = line.split("=")[1] if "=" in line else None
|
|
# Check next line for disposition
|
|
is_attached_pic = False
|
|
if i + 1 < len(lines):
|
|
disp_line = lines[i + 1].strip()
|
|
if disp_line.startswith("disposition="):
|
|
if "attached_pic=1" in disp_line:
|
|
is_attached_pic = True
|
|
|
|
# If not attached pic, analyze the pixel format
|
|
if pix_fmt and not is_attached_pic:
|
|
pix_fmt_lower = pix_fmt.lower()
|
|
# Check for 12-bit indicators first
|
|
if any(x in pix_fmt_lower for x in ["12le", "12be"]):
|
|
return 12
|
|
# Check for 10-bit indicators
|
|
elif any(x in pix_fmt_lower for x in ["10le", "10be", "p010", "yuv420p10"]):
|
|
return 10
|
|
else:
|
|
return 8
|
|
|
|
i += 1
|
|
|
|
# Fallback to simple method if no streams found
|
|
logger.debug(f"Could not detect bit depth for {input_file.name}. Defaulting to 8-bit")
|
|
return 8
|
|
except Exception as e:
|
|
logger.warning(f"Failed to detect source bit depth: {e}. Defaulting to 8-bit")
|
|
return 8
|
|
|
|
|
|
def is_hdr(input_file: Path) -> bool:
|
|
"""
|
|
Detect if source video is HDR by checking color space and transfer characteristics.
|
|
|
|
HDR indicators:
|
|
- color_space: bt2020 (BT.2020 wide gamut color space)
|
|
- color_transfer: smpte2084 (SMPTE ST 2084 PQ tone mapping for HDR10)
|
|
- color_range: tv (limited range typical for HDR)
|
|
- color_primaries: bt2020 (BT.2020 primaries)
|
|
|
|
Returns: True if source is detected as HDR, False otherwise
|
|
Skips attached pictures and cover art.
|
|
"""
|
|
try:
|
|
# Get video stream color information
|
|
cmd = [
|
|
"ffprobe", "-v", "error",
|
|
"-select_streams", "v",
|
|
"-show_entries", "stream=color_space,color_transfer,color_primaries,color_range,disposition",
|
|
"-of", "default=noprint_wrappers=1",
|
|
str(input_file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', check=False)
|
|
|
|
if result.stdout:
|
|
lines = result.stdout.strip().split("\n")
|
|
|
|
# Look for HDR indicators in the first non-attached-pic video stream
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
|
|
# Check if this is an attached pic (skip if so)
|
|
is_attached_pic = False
|
|
if line.startswith("color_space="):
|
|
# Look ahead for disposition
|
|
j = i + 1
|
|
while j < len(lines) and not lines[j].strip().startswith("color_"):
|
|
if lines[j].strip().startswith("disposition="):
|
|
if "attached_pic=1" in lines[j]:
|
|
is_attached_pic = True
|
|
break
|
|
j += 1
|
|
|
|
if not is_attached_pic:
|
|
# Parse color properties starting from this stream
|
|
color_space = None
|
|
color_transfer = None
|
|
color_primaries = None
|
|
color_range = None
|
|
|
|
k = i
|
|
while k < len(lines) and not lines[k].strip() == "":
|
|
entry_line = lines[k].strip()
|
|
if entry_line.startswith("color_space="):
|
|
color_space = entry_line.split("=")[1] if "=" in entry_line else None
|
|
elif entry_line.startswith("color_transfer="):
|
|
color_transfer = entry_line.split("=")[1] if "=" in entry_line else None
|
|
elif entry_line.startswith("color_primaries="):
|
|
color_primaries = entry_line.split("=")[1] if "=" in entry_line else None
|
|
elif entry_line.startswith("color_range="):
|
|
color_range = entry_line.split("=")[1] if "=" in entry_line else None
|
|
elif entry_line.startswith("disposition="):
|
|
# End of this stream's properties
|
|
break
|
|
k += 1
|
|
|
|
# Check for HDR characteristics
|
|
# HDR typically has: bt2020 color space + smpte2084 transfer + bt2020 primaries
|
|
is_hdr_content = False
|
|
if color_space and "bt2020" in color_space.lower():
|
|
if color_transfer and "smpte2084" in color_transfer.lower():
|
|
is_hdr_content = True
|
|
|
|
logger.debug(f"HDR Detection for {input_file.name}: color_space={color_space}, color_transfer={color_transfer}, color_primaries={color_primaries}, is_hdr={is_hdr_content}")
|
|
return is_hdr_content
|
|
|
|
i += 1
|
|
|
|
# No HDR indicators found
|
|
logger.debug(f"No HDR indicators found for {input_file.name}")
|
|
return False
|
|
except Exception as e:
|
|
logger.warning(f"Failed to detect HDR status: {e}. Assuming SDR")
|
|
return False
|
|
|
|
|
|
def determine_target_resolution(src_width: int, src_height: int, explicit_resolution: str = None) -> tuple:
|
|
"""
|
|
Determine target resolution based on source and explicit override.
|
|
|
|
Returns tuple: (res_width, res_height, target_resolution_label)
|
|
|
|
Logic:
|
|
If explicit_resolution specified: use it as a MAXIMUM (downscale only, never upscale)
|
|
- If source > max: scale down to max
|
|
- If source <= max: preserve source resolution
|
|
Special case for 2160p:
|
|
- Source MUST be actual 4K (2160p or higher) or encoding will be skipped
|
|
- If source is 4K, preserve it (no downscaling)
|
|
Else:
|
|
- If source > 1080p: scale to 1080p (default downscale 4K to 1080p)
|
|
- If source <= 1080p: preserve source resolution
|
|
"""
|
|
if explicit_resolution:
|
|
# User explicitly specified resolution as a maximum threshold
|
|
max_height = int(explicit_resolution)
|
|
|
|
if max_height == 2160:
|
|
# Special handling for 2160p (4K) mode
|
|
# Source MUST be actual 4K (check both width and height for various 4K formats)
|
|
# 4K is defined as: width >= 3840 OR height >= 1440 (accounts for different 4K formats)
|
|
is_4k = src_width >= 3840 or src_height >= 1440
|
|
|
|
if is_4k:
|
|
# Source is 4K or higher - keep at source resolution (don't downscale)
|
|
return (src_width, src_height, "2160")
|
|
else:
|
|
# Source is NOT 4K - return special signal that this should be skipped
|
|
# We return a tuple with a special marker
|
|
logger.warning(f"4K mode requested (--r 2160) but source is only {src_width}x{src_height}. Encoding will be skipped.")
|
|
return (src_width, src_height, "2160_SKIP")
|
|
|
|
elif src_height > max_height:
|
|
# Source is larger than max - downscale to max
|
|
if max_height == 1080:
|
|
return (1920, 1080, "1080")
|
|
elif max_height == 720:
|
|
return (1280, 720, "720")
|
|
else: # 480
|
|
return (854, 480, "480")
|
|
else:
|
|
# Source is <= max - preserve source resolution (no upscaling)
|
|
if src_height <= 720:
|
|
return (src_width, src_height, "720")
|
|
else:
|
|
return (src_width, src_height, "1080")
|
|
else:
|
|
# No explicit resolution - use smart defaults
|
|
# 4K content is downscaled to 1080p by default (unless --r 2160 specified)
|
|
if src_height > 1080:
|
|
# Scale down anything above 1080p to 1080p
|
|
return (1920, 1080, "1080")
|
|
else:
|
|
# Preserve source resolution (480p, 720p, 1080p, etc.)
|
|
if src_height <= 720:
|
|
return (src_width, src_height, "720")
|
|
else:
|
|
return (src_width, src_height, "1080")
|
|
|
|
|
|
def get_subtitle_stream_codecs(input_file: Path) -> dict:
|
|
"""
|
|
Get the codec name for each subtitle stream in the input file.
|
|
Returns a dictionary: {stream_index: codec_name}
|
|
"""
|
|
try:
|
|
cmd = [
|
|
"ffprobe", "-v", "error",
|
|
"-select_streams", "s",
|
|
"-show_entries", "stream=codec_name",
|
|
"-of", "json",
|
|
str(input_file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', check=False)
|
|
|
|
subtitle_codecs = {}
|
|
if result.stdout:
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
for i, stream in enumerate(data.get("streams", [])):
|
|
codec_name = stream.get("codec_name", "unknown")
|
|
subtitle_codecs[i] = codec_name
|
|
except json.JSONDecodeError:
|
|
logger.debug(f"Failed to parse subtitle codecs for {input_file.name}")
|
|
|
|
return subtitle_codecs
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get subtitle stream codecs for {input_file.name}: {e}")
|
|
return {}
|
|
|
|
def has_pgs_subtitles(input_file: Path) -> bool:
|
|
"""
|
|
Check if the input file has PGS (presentation graphics) subtitles.
|
|
PGS is the codec used for bitmap subtitles in Blu-ray discs.
|
|
Returns True if at least one subtitle stream uses 'hdmv_pgs_subtitle' codec.
|
|
"""
|
|
try:
|
|
subtitle_codecs = get_subtitle_stream_codecs(input_file)
|
|
for codec_name in subtitle_codecs.values():
|
|
if codec_name.lower() == "hdmv_pgs_subtitle":
|
|
logger.debug(f"Found PGS subtitle stream in {input_file.name}")
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
logger.warning(f"Failed to check for PGS subtitles in {input_file.name}: {e}")
|
|
return False
|
|
|
|
def has_forced_subtitles(input_file: Path) -> bool:
|
|
"""
|
|
Check if the input file has any subtitles with the forced flag set.
|
|
Returns True if at least one subtitle stream has forced=1 disposition.
|
|
"""
|
|
try:
|
|
# Method 1: Try JSON output (most reliable)
|
|
cmd = [
|
|
"ffprobe", "-v", "error",
|
|
"-select_streams", "s",
|
|
"-show_entries", "stream=disposition",
|
|
"-of", "json",
|
|
str(input_file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', check=False)
|
|
|
|
if result.stdout:
|
|
try:
|
|
data = json.loads(result.stdout)
|
|
for stream in data.get("streams", []):
|
|
disposition = stream.get("disposition", {})
|
|
if isinstance(disposition, dict) and disposition.get("forced") == 1:
|
|
logger.debug(f"Found forced subtitle stream in {input_file.name}")
|
|
return True
|
|
except json.JSONDecodeError:
|
|
logger.debug(f"Failed to parse JSON from ffprobe for {input_file.name}, trying fallback method")
|
|
|
|
# Method 2: Fallback to text search for "forced=1" or "(forced)"
|
|
cmd = [
|
|
"ffprobe", "-v", "info",
|
|
"-select_streams", "s",
|
|
str(input_file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore', check=False)
|
|
|
|
if result.stderr:
|
|
# Look for "(forced)" in the human-readable ffprobe output
|
|
if "(forced)" in result.stderr:
|
|
logger.debug(f"Found (forced) in ffprobe output for {input_file.name}")
|
|
return True
|
|
|
|
return False
|
|
except Exception as e:
|
|
logger.warning(f"Failed to check forced subtitles for {input_file.name}: {e}")
|
|
return False
|
|
|
|
def calculate_crop_dimensions(src_height: int, target_height: int) -> dict:
|
|
"""
|
|
Calculate crop dimensions to center-crop video to target height.
|
|
Maintains width, crops from top and bottom equally.
|
|
|
|
Args:
|
|
src_height: Source video height in pixels
|
|
target_height: Target crop height in pixels
|
|
|
|
Returns:
|
|
dict with:
|
|
- "ffmpeg_filter": FFmpeg crop filter string or empty if no crop needed
|
|
- "crop_top": Pixels to crop from top
|
|
- "crop_bottom": Pixels to crop from bottom
|
|
"""
|
|
if target_height >= src_height or target_height <= 0:
|
|
return {
|
|
"ffmpeg_filter": "",
|
|
"crop_top": 0,
|
|
"crop_bottom": 0
|
|
}
|
|
|
|
# Calculate pixels to remove total
|
|
pixels_to_remove = src_height - target_height
|
|
|
|
# Crop equally from top and bottom (centered crop)
|
|
crop_amount = pixels_to_remove // 2
|
|
|
|
# Note: FFmpeg crop filter is crop=width:height:x:y
|
|
# We use 'in_w' to preserve input width, and y is the vertical offset (crop_amount)
|
|
# For 1920x1080 -> 1920x816: crop=in_w:816:0:132
|
|
# where 132 = (1080 - 816) / 2
|
|
|
|
return {
|
|
"ffmpeg_filter": f"crop=in_w:{target_height}:0:{crop_amount}",
|
|
"crop_top": crop_amount,
|
|
"crop_bottom": crop_amount
|
|
} |