dcosnet
/
OpenTranscode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
OpenTranscode/open-transcode.py

3940 lines
162 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""
OpenCodec Transcode Master — Merged & Refactored
=================================================
Merges best of: trans.py (CTk probe/tooltips), custom-transcoder (QThread/pathlib),
and the bash script's ffmpeg-fallback + WebM concept.
Key refactors over originals:
- Config-driven codec/audio/container profiles (no more nested if/else chains)
- Distro-aware: Arch, Fedora, RHEL/CentOS/Rocky/Alma, openSUSE, NixOS, Debian/Ubuntu
- Per-distro binary search paths, package manager, install hints, encoder name quirks
- FFmpeg encoder library availability probe (greys out unavailable codecs in UI)
- ffprobe pre-validation before encoding
- QThread worker (thread-safe UI, proper signal/slot)
- Runtime av1an flag + version probing
- Dynamic worker count (os.cpu_count - 2)
- pathlib throughout
- Per-file progress tracking
- Container format selection (MKV / WebM)
- AV1/VP9/x265 preset selection
- Configurable input extensions
- Batch delete with summary prompt (not per-file)
"""
import hashlib
import math
import os
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import ctypes
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from PySide6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QComboBox, QCheckBox,
QTextEdit, QFileDialog, QGroupBox, QStatusBar, QMessageBox,
QStyleFactory,
)
from PySide6.QtCore import Qt, QThread, Signal, Slot, QPointF, QRectF
from PySide6.QtGui import (
QFont, QPalette, QColor, QPainter, QPen, QBrush,
QRadialGradient, QFontMetrics,
)
# ──────────────────────────────────────────────
# RADIO KNOB WIDGET (oldschool rotary control)
# ──────────────────────────────────────────────
class RadioKnob(QWidget):
"""
A retro radio-style rotary knob widget.
Supports arc range, tick marks, and a glowing indicator dot.
Rotation: 7 o'clock (min) to 5 o'clock (max) = 300 degrees.
"""
valueChanged = Signal(float)
def __init__(
self,
parent=None,
min_val: float = 0.0,
max_val: float = 100.0,
default_val: float = 50.0,
label: str = "",
unit: str = "",
color: tuple = (42, 130, 218),
num_ticks: int = 17,
tick_labels: list[str] | None = None,
snap_ticks: bool = False,
compact: bool = False,
):
super().__init__(parent)
self.min_val = min_val
self.max_val = max_val
self._value = default_val
self.label = label
self.unit = unit
self.color = QColor(*color)
self.num_ticks = num_ticks
self.tick_labels = tick_labels
self.snap_ticks = snap_ticks
self._dragging = False
self.compact = compact
# Arc geometry: 300-degree sweep, centered at 12 o'clock
self._arc_start = 210.0 # degrees (7 o'clock)
self._arc_span = -300.0 # negative = clockwise
# Scaling factor for compact mode (~70% of full size)
s = 0.70 if compact else 1.0
self._s = s
self.setFixedSize(int(180 * s), int(210 * s))
self.setCursor(Qt.CursorShape.PointingHandCursor)
# --- Public API ---
def value(self) -> float:
return self._value
def setValue(self, v: float):
v = max(self.min_val, min(self.max_val, v))
if self.snap_ticks:
v = self._snap(v)
if v != self._value:
self._value = v
self.update()
self.valueChanged.emit(v)
def intValue(self) -> int:
return int(round(self._value))
def _snap(self, v: float) -> float:
"""Snap to nearest tick."""
step = (self.max_val - self.min_val) / max(1, self.num_ticks - 1)
return round((v - self.min_val) / step) * step + self.min_val
def _val_to_angle(self, v: float) -> float:
"""Map value to angle in degrees (matching the conical gradient)."""
ratio = (v - self.min_val) / (self.max_val - self.min_val) if self.max_val != self.min_val else 0
return self._arc_start + ratio * self._arc_span # goes from 210 -> -90
def _angle_to_val(self, angle_deg: float) -> float:
"""Map angle back to value."""
# Normalize angle relative to arc start
ratio = (angle_deg - self._arc_start) / self._arc_span
ratio = max(0.0, min(1.0, ratio))
v = self.min_val + ratio * (self.max_val - self.min_val)
if self.snap_ticks:
v = self._snap(v)
return v
# --- Painting ---
def paintEvent(self, event):
p = QPainter(self)
p.setRenderHint(QPainter.RenderHint.Antialiasing)
w, h = self.width(), self.height()
s = self._s # scale factor (0.7 for compact, 1.0 for full)
cx = w / 2
cy = h / 2 - 4 * s
outer_r = 70 * s
knob_r = 40 * s
arc_w = max(1, int(8 * s))
tick_w = max(1, 1.5 * s)
bezel_pad = 6 * s
# --- Outer bezel ring ---
bezel_grad = QRadialGradient(cx, cy, outer_r + bezel_pad)
bezel_grad.setColorAt(0.85, QColor(48, 48, 52))
bezel_grad.setColorAt(1.0, QColor(26, 26, 30))
p.setBrush(QBrush(bezel_grad))
p.setPen(Qt.PenStyle.NoPen)
p.drawEllipse(QPointF(cx, cy), outer_r + bezel_pad, outer_r + bezel_pad)
# --- Inactive arc (dark track) ---
p.setPen(QPen(QColor(50, 50, 56), arc_w, Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap))
p.drawArc(QRectF(cx - outer_r, cy - outer_r, outer_r * 2, outer_r * 2),
int(self._arc_start * 16), int(self._arc_span * 16))
# --- Active arc (colored fill up to current value) ---
val_angle = self._val_to_angle(self._value)
active_span = val_angle - self._arc_start
if abs(active_span) > 0.5:
arc_color = QColor(self.color)
p.setPen(QPen(arc_color, arc_w, Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap))
p.drawArc(QRectF(cx - outer_r, cy - outer_r, outer_r * 2, outer_r * 2),
int(self._arc_start * 16), int(active_span * 16))
# --- Tick marks ---
for i in range(self.num_ticks):
t = i / (self.num_ticks - 1) if self.num_ticks > 1 else 0
tick_angle = self._val_to_angle(self.min_val + t * (self.max_val - self.min_val))
tick_rad = tick_angle * math.pi / 180.0
ox = cx + (outer_r + 12 * s) * (-1) * math.sin(tick_rad)
oy = cy + (outer_r + 12 * s) * (-1) * (-math.cos(tick_rad))
ix_ = cx + (outer_r + 3 * s) * (-1) * math.sin(tick_rad)
iy_ = cy + (outer_r + 3 * s) * (-1) * (-math.cos(tick_rad))
p.setPen(QPen(QColor(130, 130, 130), tick_w))
p.drawLine(QPointF(ix_, iy_), QPointF(ox, oy))
# Tick labels (if provided)
if self.tick_labels:
p.setFont(QFont("Sans", max(5, int(7 * s))))
p.setPen(QColor(160, 160, 160))
step = max(1, self.num_ticks // len(self.tick_labels))
label_idx = 0
for i in range(0, self.num_ticks, step):
if label_idx >= len(self.tick_labels):
break
t = i / (self.num_ticks - 1) if self.num_ticks > 1 else 0
tick_angle = self._val_to_angle(self.min_val + t * (self.max_val - self.min_val))
tick_rad = tick_angle * math.pi / 180.0
lx = cx + (outer_r + 24 * s) * (-1) * math.sin(tick_rad)
ly = cy + (outer_r + 24 * s) * (-1) * (-math.cos(tick_rad))
txt = self.tick_labels[label_idx]
fm = QFontMetrics(p.font())
tw = fm.horizontalAdvance(txt)
p.drawText(QPointF(lx - tw / 2, ly + 2 * s), txt)
label_idx += 1
# --- Knob body (dark brushed aluminum) ---
knob_grad = QRadialGradient(cx - 6 * s, cy - 6 * s, knob_r * 1.3)
knob_grad.setColorAt(0.0, QColor(72, 72, 78))
knob_grad.setColorAt(0.5, QColor(50, 50, 55))
knob_grad.setColorAt(1.0, QColor(34, 34, 38))
p.setBrush(QBrush(knob_grad))
p.setPen(QPen(QColor(26, 26, 30), max(1, 1.5 * s)))
p.drawEllipse(QPointF(cx, cy), knob_r, knob_r)
# --- Inner shadow ring ---
inner_shadow = QRadialGradient(cx, cy, knob_r - 2)
inner_shadow.setColorAt(0.85, QColor(0, 0, 0, 0))
inner_shadow.setColorAt(1.0, QColor(0, 0, 0, 60))
p.setBrush(QBrush(inner_shadow))
p.setPen(Qt.PenStyle.NoPen)
p.drawEllipse(QPointF(cx, cy), knob_r - 1, knob_r - 1)
# --- Indicator line (pointer) ---
ptr_angle = self._val_to_angle(self._value)
ptr_rad = ptr_angle * 3.14159265 / 180.0
ptr_len = knob_r - 8 * s
px = cx + ptr_len * (-1) * math.sin(ptr_rad)
py = cy + ptr_len * (-1) * (-math.cos(ptr_rad))
p.setPen(QPen(QColor(255, 255, 255, 220), max(1, 2.5 * s),
Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap))
p.drawLine(QPointF(cx, cy), QPointF(px, py))
# --- Center cap dot ---
cap_r = max(2, 5 * s)
cap_grad = QRadialGradient(cx, cy, cap_r)
cap_grad.setColorAt(0.0, QColor(60, 60, 65))
cap_grad.setColorAt(1.0, QColor(30, 30, 34))
p.setBrush(QBrush(cap_grad))
p.setPen(Qt.PenStyle.NoPen)
p.drawEllipse(QPointF(cx, cy), cap_r, cap_r)
# --- Glow dot at arc tip ---
glow_r = max(3, 10 * s)
glow_x = cx + outer_r * (-1) * math.sin(ptr_rad)
glow_y = cy + outer_r * (-1) * (-math.cos(ptr_rad))
glow = QRadialGradient(glow_x, glow_y, glow_r * 1.2)
glow.setColorAt(0.0, QColor(self.color.red(), self.color.green(), self.color.blue(), 200))
glow.setColorAt(1.0, QColor(self.color.red(), self.color.green(), self.color.blue(), 0))
p.setBrush(QBrush(glow))
p.setPen(Qt.PenStyle.NoPen)
p.drawEllipse(QPointF(glow_x, glow_y), glow_r, glow_r)
p.end()
# --- Label + value text below knob ---
p2 = QPainter(self)
p2.setRenderHint(QPainter.RenderHint.Antialiasing)
# Value line (e.g. "32.0 CRF")
val_font_sz = max(6, int(13 * s))
p2.setFont(QFont("Consolas", val_font_sz, QFont.Weight.Bold))
val_color = QColor(self.color.red(), self.color.green(), self.color.blue())
p2.setPen(val_color)
val_text = f"{self._value:.0f} {self.unit}" if self.unit else f"{self._value:.0f}"
p2.drawText(QRectF(0, h - 38 * s, w, 20 * s), Qt.AlignmentFlag.AlignCenter, val_text)
# Label line (e.g. "Quality")
lbl_font_sz = max(5, int(9 * s))
p2.setFont(QFont("Consolas", lbl_font_sz, QFont.Weight.Bold))
p2.setPen(QColor(160, 160, 160))
p2.drawText(QRectF(0, h - 18 * s, w, 16 * s), Qt.AlignmentFlag.AlignCenter, self.label)
p2.end()
# --- Input handling ---
def mousePressEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self._dragging = True
self._update_from_mouse(event.position())
def mouseMoveEvent(self, event):
if self._dragging:
self._update_from_mouse(event.position())
def mouseReleaseEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self._dragging = False
def wheelEvent(self, event):
delta = event.angleDelta().y()
step = (self.max_val - self.min_val) / max(1, self.num_ticks - 1)
if delta > 0:
self.setValue(self._value + step)
elif delta < 0:
self.setValue(self._value - step)
def _update_from_mouse(self, pos: QPointF):
cx = self.width() / 2
cy = self.height() / 2 - 4 * self._s
dx = pos.x() - cx
dy = pos.y() - cy
angle = math.degrees(math.atan2(dx, -dy)) # 0=north, CW positive
if angle < 0:
angle += 360
# Clamp to arc range: 210..510 (which is 210..360 and 0..150)
# Our arc: 210 degrees to -90 (=270) degrees clockwise
if angle < 210 and angle > 150:
# Dead zone at bottom (between 150 and 210)
# Push to nearest end
angle = 210 if abs(angle - 210) < abs(angle - 510) else 510
if angle > 360:
angle -= 360 # normalize back to 0..360
self.setValue(self._angle_to_val(angle))
# ──────────────────────────────────────────────
# CONFIG-DRIVEN PROFILES (replaces all if/else chains)
# ──────────────────────────────────────────────
@dataclass
class VideoCodecProfile:
label: str # Display name in combo box
av1an_encoder: str # Encoder name passed to --encoder
ffmpeg_encoder: str # Encoder name for pure-ffmpeg fallback (e.g. "libsvtav1")
container: str # Default container extension (mkv or webm)
crf_range: tuple[int, int] # (min, max) valid CRF values
default_crf: int
params_fn: callable # (crf: int, preset: int) -> str (av1an format)
ffmpeg_vargs_fn: callable # (crf: int, preset: int) -> list[str] (ffmpeg -c:v args)
presets: list[str] # Human-readable preset labels
preset_map: dict[str, int] # label -> internal preset value
@dataclass
class AudioProfile:
label: str
params: list[str] # Tokens passed to --audio-params (joined with space)
@dataclass
class ContainerProfile:
label: str
ext: str # e.g. "mkv", "webm"
def _av1_params(crf: int, preset: int) -> str:
"""SVT-AV1 encoder params for av1an's --video-params.
av1an splits the --video-params value by whitespace (``split_whitespace()``)
and passes each resulting token as a separate argument to SvtAv1EncApp.
Therefore the string must contain space-separated ``--flag value`` pairs
that SvtAv1EncApp can parse natively.
Colon-separated ``key=value:key=value`` does NOT work because there are
no whitespace boundaries for av1an to split on — the entire string reaches
SvtAv1EncApp as one opaque argument, producing:
``Maybe missing spacing between tokens``.
"""
return f"--preset {preset} --crf {crf} --keyint 240"
def _vp9_params(crf: int, preset: int) -> str:
"""VP9 encoder params for av1an's --video-params.
av1an splits by whitespace, so we use space-separated --flag=value tokens
that vpxenc parses natively.
"""
cpu_used = max(0, 8 - preset)
return f"--end-usage=q --cq-level={crf} --cpu-used={cpu_used}"
def _x265_params(crf: int, preset: int) -> str:
"""x265 encoder params for av1an's --video-params.
av1an splits by whitespace, so we use space-separated --flag value tokens
that x265 parses natively.
"""
return f"--crf {crf} --preset {preset}"
def _svtav1_ffmpeg_args(crf: int, preset: int) -> list[str]:
"""FFmpeg args for SVT-AV1 (maps av1an preset=0..8 → svtav1 -preset 0..13)."""
# av1an preset range 0-8 maps to SVT-AV1 preset range 0-13
# Scale roughly: 8→0, 6→4, 4→7, 2→10
svt_preset = max(0, min(13, round((8 - preset) * 13 / 8)))
return ["-c:v", "libsvtav1", "-preset", str(svt_preset), "-crf", str(crf),
"-pix_fmt", "yuv420p10le", "-g", "240"]
def _vp9_ffmpeg_args(crf: int, preset: int) -> list[str]:
"""FFmpeg args for VP9 (maps av1an cpu-used 0..8 → -cpu-used 0..8)."""
cpu_used = max(0, min(8, preset))
return ["-c:v", "libvpx-vp9", "-crf", str(crf), "-b:v", "0",
"-cpu-used", str(cpu_used), "-pix_fmt", "yuv420p", "-g", "240",
"-row-mt", "1", "-tiles", "2x2"]
def _x265_ffmpeg_args(crf: int, preset: int) -> list[str]:
"""FFmpeg args for x265 (maps av1an preset 5..10 → x265 -preset)."""
# av1an x265 preset range 5-10 maps to x265 preset names
preset_names = {5: "slow", 7: "medium", 9: "fast", 10: "faster"}
p = preset_names.get(preset, "medium")
return ["-c:v", "libx265", "-preset", p, "-crf", str(crf),
"-pix_fmt", "yuv420p10le", "-g", "240"]
VIDEO_CODECS: list[VideoCodecProfile] = [
VideoCodecProfile(
label="AV1 (SVT-AV1)",
av1an_encoder="svt_av1",
ffmpeg_encoder="libsvtav1",
container="mkv",
crf_range=(18, 52),
default_crf=32,
params_fn=_av1_params,
ffmpeg_vargs_fn=_svtav1_ffmpeg_args,
presets=["Slow (8)", "Medium (6)", "Fast (4)", "Faster (2)"],
preset_map={"Slow (8)": 8, "Medium (6)": 6, "Fast (4)": 4, "Faster (2)": 2},
),
VideoCodecProfile(
label="VP9",
av1an_encoder="vpx",
ffmpeg_encoder="libvpx-vp9",
container="webm",
crf_range=(18, 52),
default_crf=32,
params_fn=_vp9_params,
ffmpeg_vargs_fn=_vp9_ffmpeg_args,
presets=["Slow (0)", "Medium (2)", "Fast (4)", "Faster (6)"],
preset_map={"Slow (0)": 0, "Medium (2)": 2, "Fast (4)": 4, "Faster (6)": 6},
),
VideoCodecProfile(
label="x265 (HEVC)",
av1an_encoder="x265",
ffmpeg_encoder="libx265",
container="mkv",
crf_range=(18, 40),
default_crf=28,
params_fn=_x265_params,
ffmpeg_vargs_fn=_x265_ffmpeg_args,
presets=["Slow (5)", "Medium (7)", "Fast (9)", "Faster (10)"],
preset_map={"Slow (5)": 5, "Medium (7)": 7, "Fast (9)": 9, "Faster (10)": 10},
),
]
AUDIO_PROFILES: list[AudioProfile] = [
AudioProfile(label="Opus (96k)", params=["-c:a", "libopus", "-b:a", "96k"]),
AudioProfile(label="Opus (128k)", params=["-c:a", "libopus", "-b:a", "128k"]),
AudioProfile(label="Opus (64k)", params=["-c:a", "libopus", "-b:a", "64k"]),
AudioProfile(label="Vorbis (128k)", params=["-c:a", "libvorbis", "-b:a", "128k"]),
AudioProfile(label="Vorbis (192k)", params=["-c:a", "libvorbis", "-b:a", "192k"]),
AudioProfile(label="FLAC (lossless)", params=["-c:a", "flac"]),
]
CONTAINER_PROFILES: list[ContainerProfile] = [
ContainerProfile(label="MKV (Matroska)", ext="mkv"),
ContainerProfile(label="WebM", ext="webm"),
]
# ── Resolution presets ──
# Aspect ratios:
# Standard 16:9 -> w/h = 1.778
# Wide 21:9 -> w/h = 2.333
# Ultrawide 32:9 -> w/h = 3.556
@dataclass
class ResolutionProfile:
label: str # Display label in dropdown, e.g. "1080p Wide (2560x1080)"
category: str # Grouping key: "standard", "wide", "ultrawide", "original"
width: Optional[int] # None for "original" (no scaling)
height: Optional[int] # None for "original"
aspect_label: str # "16:9", "21:9", "32:9", "Source"
RESOLUTION_PRESETS: list[ResolutionProfile] = [
# ── Original (no scaling) ──
ResolutionProfile("Original (No Scaling)", "original", None, None, "Source"),
# ── Standard 16:9 ──
ResolutionProfile("480p ( 854x 480)", "standard", 854, 480, "16:9"),
ResolutionProfile("720p (1280x 720)", "standard", 1280, 720, "16:9"),
ResolutionProfile("1080p (1920x1080)", "standard", 1920, 1080, "16:9"),
ResolutionProfile("2K (2560x1440)", "standard", 2560, 1440, "16:9"),
ResolutionProfile("4K (3840x2160)", "standard", 3840, 2160, "16:9"),
# ── Wide 21:9 ──
ResolutionProfile("480p Wide ( 854x 366)", "wide", 854, 366, "21:9"),
ResolutionProfile("720p Wide (1280x 549)", "wide", 1280, 549, "21:9"),
ResolutionProfile("1080p Wide (2560x1080)", "wide", 2560, 1080, "21:9"),
ResolutionProfile("2K Wide (3440x1440)", "wide", 3440, 1440, "21:9"),
ResolutionProfile("4K Wide (5120x2160)", "wide", 5120, 2160, "21:9"),
# ── Ultrawide 32:9 ──
ResolutionProfile("480p UW (1706x 480)", "ultrawide", 1706, 480, "32:9"),
ResolutionProfile("1080p UW (3840x1080)", "ultrawide", 3840, 1080, "32:9"),
ResolutionProfile("2K UW (5120x1440)", "ultrawide", 5120, 1440, "32:9"),
ResolutionProfile("4K UW (7680x2160)", "ultrawide", 7680, 2160, "32:9"),
]
SUBTITLE_OPTIONS = [
("None", None),
("English", "eng"),
]
DEFAULT_INPUT_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".ts", ".m4v", ".flv", ".wmv", ".webm", ".mpg", ".mpeg"}
# ──────────────────────────────────────────────
# CPU TOPOLOGY (physical cores, not hyperthreads)
# ──────────────────────────────────────────────
@dataclass
class CpuTopology:
physical_cores: int
logical_threads: int
threads_per_core: int
model_name: str
def _read_sysfs_cores() -> Optional[tuple[int, int]]:
"""
Read /sys/devices/system/cpu/cpu*/topology/ to count unique
(physical_package_id, core_id) pairs — i.e. physical cores.
Returns (physical_cores, logical_threads) or None.
"""
cpu_base = Path("/sys/devices/system/cpu")
if not cpu_base.exists():
return None
unique_cores: set[tuple[str, str]] = set()
logical = 0
for cpu_dir in sorted(cpu_base.glob("cpu[0-9]*")):
core_id_file = cpu_dir / "topology" / "core_id"
pkg_id_file = cpu_dir / "topology" / "physical_package_id"
if core_id_file.exists() and pkg_id_file.exists():
try:
pkg = pkg_id_file.read_text().strip()
core = core_id_file.read_text().strip()
unique_cores.add((pkg, core))
logical += 1
except Exception:
pass
if unique_cores and logical:
return (len(unique_cores), logical)
return None
def _read_lscpu_cores() -> Optional[tuple[int, int]]:
"""Fallback: parse lscpu -p=CORE,SOCKET for unique physical cores."""
if not shutil.which("lscpu"):
return None
try:
res = subprocess.run(
["lscpu", "-p=CORE,SOCKET"],
capture_output=True, text=True, timeout=5,
)
lines = [l.strip() for l in res.stdout.strip().splitlines() if l.strip() and not l.startswith("#")]
if lines:
unique = set(lines)
return (len(unique), len(lines))
except Exception:
pass
return None
def detect_cpu_topology() -> CpuTopology:
"""
Detect physical CPU topology. Prefers /sys filesystem, falls back
to lscpu, then estimates from os.cpu_count().
"""
logical = os.cpu_count() or 1
physical = logical
# Try /sys first (most reliable)
result = _read_sysfs_cores()
if result:
physical, logical = result
else:
# Try lscpu
result = _read_lscpu_cores()
if result:
physical, logical = result
else:
# Estimate: assume 2 threads/core if cpu_count > 2 and is even
if logical > 2 and logical % 2 == 0:
physical = logical // 2
tpc = logical // physical if physical > 0 else 1
# Try to get CPU model name
model = "Unknown CPU"
model_file = Path("/proc/cpuinfo")
if model_file.exists():
for line in model_file.read_text(errors="replace").splitlines():
if line.startswith("model name"):
model = line.split(":", 1)[1].strip()
break
else:
# Non-x86 / non-Linux: try lscpu
if shutil.which("lscpu"):
try:
res = subprocess.run(["lscpu"], capture_output=True, text=True, timeout=5)
for line in res.stdout.splitlines():
if "Model name" in line:
model = line.split(":", 1)[1].strip()
break
except Exception:
pass
return CpuTopology(
physical_cores=physical,
logical_threads=logical,
threads_per_core=tpc,
model_name=model,
)
# ──────────────────────────────────────────────
# DISTRO DETECTION & PROFILES
# ──────────────────────────────────────────────
@dataclass
class DistroProfile:
family: str # Canonical family: arch, debian, redhat, suse, nixos, unknown
name: str # Pretty name: "Arch Linux", "Fedora 40", etc.
version_id: str # e.g. "40", "15.6", "24.05"
pkg_manager: str # e.g. "pacman", "dnf", "zypper", "apt", "nix"
install_cmd_template: str # e.g. "sudo pacman -S {packages}"
binary_extra_paths: list[str] # Distro-specific dirs to search for binaries
av1an_known_encoder_names: list[str] # Names this distro's av1an build may accept
ffmpeg_pkg: str # Package name providing ffmpeg
av1an_pkg: str # Package name providing av1an
notes: str # Distro-specific quirks worth showing the user
# Runtime dependency packages (key = generic name, value = distro package name)
dep_pkgs: dict[str, str] = field(default_factory=dict)
# Binaries that av1an invokes directly (not via ffmpeg)
encoder_binaries: dict[str, list[str]] = field(default_factory=dict)
# VSScript package name — on most distros this is bundled into 'vapoursynth',
# but Debian/Ubuntu split it into a separate -script-dev package.
# If set, this takes priority over dep_pkgs["vapoursynth"] for the VS check.
vsscript_pkg: str = ""
def _read_os_release() -> dict[str, str]:
"""Parse /etc/os-release into a dict. Falls back to empty dict."""
os_release = Path("/etc/os-release")
fallback = Path("/usr/lib/os-release")
target = os_release if os_release.exists() else fallback
if not target.exists():
return {}
data = {}
for line in target.read_text(encoding="utf-8", errors="replace").splitlines():
line = line.strip()
if "=" in line and not line.startswith("#"):
key, _, val = line.partition("=")
data[key.strip()] = val.strip().strip('"')
return data
def detect_distro() -> DistroProfile:
"""
Detect the running Linux distribution via /etc/os-release.
Returns a DistroProfile with distro-specific package manager,
install commands, binary search paths, and known quirks.
"""
info = _read_os_release()
id_like = info.get("ID_LIKE", "").lower().split()
distro_id = info.get("ID", "").lower()
pretty = info.get("PRETTY_NAME", info.get("NAME", platform.system()))
version = info.get("VERSION_ID", "?")
# --- Classify into families ---
if distro_id in ("arch", "manjaro", "endeavouros", "garuda", "cachyos") or "arch" in id_like:
return DistroProfile(
family="arch",
name=pretty,
version_id=version,
pkg_manager="pacman",
install_cmd_template="sudo pacman -S {packages}",
binary_extra_paths=[
"/usr/bin",
"/usr/local/bin",
"~/.local/bin", # AUR user installs
"~/.cargo/bin", # Rust-based av1an installs
],
av1an_known_encoder_names=["svt_av1", "svt", "svt-av1", "aom", "rav1e", "vpx", "x265"],
ffmpeg_pkg="ffmpeg",
av1an_pkg="av1an",
dep_pkgs={
"vapoursynth": "vapoursynth",
"svt-av1": "svt-av1",
"x265": "x265",
"vpx": "libvpx",
"opus": "libopus",
"vorbis": "libvorbis",
"flac": "flac",
},
encoder_binaries={
"svt_av1": ["SvtAv1EncApp", "svt_av1"],
"vpx": ["vpxenc"],
"x265": ["x265"],
},
notes=(
"Arch/Manjaro: av1an is in the AUR (yay -S av1an) or community repo. "
"SVT-AV1 encoder name is typically 'svt_av1'. "
"Cargo-installed av1an may live in ~/.cargo/bin."
),
)
if distro_id in ("fedora",) or "fedora" in id_like:
return DistroProfile(
family="redhat",
name=pretty,
version_id=version,
pkg_manager="dnf",
install_cmd_template="sudo dnf install {packages}",
binary_extra_paths=[
"/usr/bin",
"/usr/local/bin",
"~/.cargo/bin",
],
av1an_known_encoder_names=["svt_av1", "svt", "aom", "rav1e", "vpx", "x265"],
ffmpeg_pkg="ffmpeg",
av1an_pkg="av1an",
dep_pkgs={
"vapoursynth": "vapoursynth",
"svt-av1": "svt-av1",
"x265": "x265",
"vpx": "libvpx-tools",
"opus": "opus",
"vorbis": "libvorbis",
"flac": "flac",
},
encoder_binaries={
"svt_av1": ["SvtAv1EncApp", "svt_av1"],
"vpx": ["vpxenc"],
"x265": ["x265"],
},
notes=(
"Fedora: av1an may require COPR enablement first: "
"sudo dnf copr enable sergiomb/av1an (or build from source). "
"SVT-AV1 is in the main repos as 'svt-av1'. "
"Ensure RPM Fusion is enabled for full codec support."
),
)
if distro_id in ("rhel", "centos", "rocky", "almalinux", "ol") or "rhel" in id_like or "centos" in id_like:
return DistroProfile(
family="redhat",
name=pretty,
version_id=version,
pkg_manager="dnf" if Path("/usr/bin/dnf").exists() else "yum",
install_cmd_template=(
"sudo dnf install {packages}"
if Path("/usr/bin/dnf").exists()
else "sudo yum install {packages}"
),
binary_extra_paths=[
"/usr/bin",
"/usr/local/bin",
"~/.cargo/bin",
],
av1an_known_encoder_names=["svt_av1", "svt", "aom", "rav1e", "vpx", "x265"],
ffmpeg_pkg="ffmpeg",
av1an_pkg="av1an",
dep_pkgs={
"vapoursynth": "vapoursynth",
"svt-av1": "svt-av1",
"x265": "x265",
"vpx": "libvpx-tools",
"opus": "opus",
"vorbis": "libvorbis",
"flac": "flac",
},
encoder_binaries={
"svt_av1": ["SvtAv1EncApp", "svt_av1"],
"vpx": ["vpxenc"],
"x265": ["x265"],
},
notes=(
"RHEL/CentOS/Rocky/Alma: av1an is NOT in default repos. "
"Options: (1) cargo install av1an, (2) build from GitHub source, "
"(3) use pre-built binary from releases. "
"Enable EPEL + RPM Fusion for FFmpeg codec support."
),
)
if distro_id in ("opensuse-leap", "opensuse-tumbleweed", "sles") or "suse" in id_like:
return DistroProfile(
family="suse",
name=pretty,
version_id=version,
pkg_manager="zypper",
install_cmd_template="sudo zypper install {packages}",
binary_extra_paths=[
"/usr/bin",
"/usr/local/bin",
"~/.cargo/bin",
],
av1an_known_encoder_names=["svt_av1", "svt", "aom", "rav1e", "vpx", "x265"],
ffmpeg_pkg="ffmpeg",
av1an_pkg="av1an",
dep_pkgs={
"vapoursynth": "vapoursynth",
"svt-av1": "svt-av1",
"x265": "x265",
"vpx": "libvpx",
"opus": "libopus",
"vorbis": "libvorbis",
"flac": "flac",
},
encoder_binaries={
"svt_av1": ["SvtAv1EncApp", "svt_av1"],
"vpx": ["vpxenc"],
"x265": ["x265"],
},
notes=(
"openSUSE: av1an may be available via OBS (Open Build Service). "
"Check: https://build.opensuse.org/package/show/multimedia:apps/av1an. "
"Packman repo provides FFmpeg with full codec support."
),
)
if "nixos" in distro_id or "nixos" in " ".join(id_like):
return DistroProfile(
family="nixos",
name=pretty,
version_id=version,
pkg_manager="nix",
install_cmd_template="nix-shell -p {packages}",
binary_extra_paths=[
"/run/current-system/sw/bin",
"~/.nix-profile/bin",
],
av1an_known_encoder_names=["svt_av1", "svt", "aom", "rav1e", "vpx", "x265"],
ffmpeg_pkg="ffmpeg",
av1an_pkg="av1an",
dep_pkgs={
"vapoursynth": "vapoursynth",
"svt-av1": "svt-av1",
"x265": "x265",
"vpx": "libvpx",
"opus": "opus",
"vorbis": "libvorbis",
"flac": "flac",
},
encoder_binaries={
"svt_av1": ["SvtAv1EncApp", "svt_av1"],
"vpx": ["vpxenc"],
"x265": ["x265"],
},
notes=(
"NixOS: Use 'nix-shell -p ffmpeg av1an' or add to configuration.nix. "
"Binaries live under /run/current-system/sw/bin or ~/.nix-profile/bin. "
"av1an CLI flags may differ from other distros depending on the nixpkgs channel."
),
)
# Debian / Ubuntu / Mint / Pop!_OS / etc.
if distro_id in ("debian", "ubuntu", "linuxmint", "pop") or "debian" in id_like:
return DistroProfile(
family="debian",
name=pretty,
version_id=version,
pkg_manager="apt",
install_cmd_template="sudo apt install {packages}",
binary_extra_paths=[
"/usr/bin",
"/usr/local/bin",
"~/.cargo/bin",
],
av1an_known_encoder_names=["svt_av1", "svt", "aom", "rav1e", "vpx", "x265"],
ffmpeg_pkg="ffmpeg",
av1an_pkg="av1an",
dep_pkgs={
"vapoursynth": "vapoursynth",
"svt-av1": "svtav1",
"x265": "x265",
"vpx": "libvpx-tools",
"opus": "libopus-dev",
"vorbis": "libvorbis-dev",
"flac": "flac",
},
encoder_binaries={
"svt_av1": ["SvtAv1EncApp", "svt_av1"],
"vpx": ["vpxenc"],
"x265": ["x265"],
},
vsscript_pkg="libvapoursynth-script-dev",
notes=(
"Debian/Ubuntu: av1an is in the repos (apt install av1an). "
"Debian repo builds may use 'svt' as encoder name instead of 'svt_av1'. "
"VSScript is in a separate package: libvapoursynth-script-dev. "
"For newer builds, consider cargo install av1an."
),
)
# Fallback: unknown distro
return DistroProfile(
family="unknown",
name=pretty,
version_id=version,
pkg_manager="unknown",
install_cmd_template="# Unknown distro — install ffmpeg and av1an manually",
binary_extra_paths=["/usr/bin", "/usr/local/bin", "~/.cargo/bin", "~/.local/bin"],
av1an_known_encoder_names=["svt_av1", "svt", "aom", "rav1e", "vpx", "x265"],
ffmpeg_pkg="ffmpeg",
av1an_pkg="av1an",
dep_pkgs={},
encoder_binaries={
"svt_av1": ["SvtAv1EncApp", "svt_av1"],
"vpx": ["vpxenc"],
"x265": ["x265"],
},
notes="Unknown distro detected. Ensure ffmpeg and av1an are in PATH.",
)
# ──────────────────────────────────────────────
# ENVIRONMENT PROBE (distro-aware, extended)
# ──────────────────────────────────────────────
@dataclass
class EnvProbe:
distro: DistroProfile = field(default_factory=lambda: DistroProfile(
family="unknown", name="Unknown", version_id="?",
pkg_manager="unknown", install_cmd_template="",
binary_extra_paths=[], av1an_known_encoder_names=[],
ffmpeg_pkg="ffmpeg", av1an_pkg="av1an", notes=""
))
av1an_path: Optional[str] = None
ffmpeg_path: Optional[str] = None
ffprobe_path: Optional[str] = None
av1an_flags: dict = field(default_factory=dict)
av1an_version: Optional[str] = None
ffmpeg_version: Optional[str] = None
ffmpeg_libs: dict = field(default_factory=dict) # lib name -> bool (available)
runtime_deps: dict = field(default_factory=dict) # dep name -> bool (present)
missing_dep_pkgs: list[str] = field(default_factory=list) # distro pkg names to install
vs_version: Optional[str] = None # VapourSynth version string (for diagnostics)
vs_script_lib: Optional[str] = None # path to libvapoursynth-script.so that passed
cpu: CpuTopology = field(default_factory=lambda: CpuTopology(1, 1, 1, "Unknown"))
errors: list[str] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
@property
def ready(self) -> bool:
return (self.av1an_path is not None and self.ffmpeg_path is not None
and not self.errors and not self.missing_dep_pkgs)
@property
def dep_install_hint(self) -> str:
"""Generate a distro-specific install command for missing runtime deps."""
if not self.missing_dep_pkgs or self.distro.family == "unknown":
return ""
return self.distro.install_cmd_template.format(packages=" ".join(self.missing_dep_pkgs))
@property
def install_hint(self) -> str:
"""Generate a distro-specific install command for missing packages."""
missing = []
if self.av1an_path is None:
missing.append(self.distro.av1an_pkg)
if self.ffmpeg_path is None:
missing.append(self.distro.ffmpeg_pkg)
if not missing:
return ""
return self.distro.install_cmd_template.format(packages=" ".join(missing))
def _find_binary(name: str, distro: DistroProfile) -> Optional[str]:
"""
Search for a binary in: (1) standard PATH via shutil.which, then
(2) distro-specific extra paths (expanded ~). Returns first match.
"""
# Standard PATH search
found = shutil.which(name)
if found:
return found
# Distro-specific extra paths
for raw_path in distro.binary_extra_paths:
expanded = Path(raw_path).expanduser()
candidate = expanded / name
if candidate.is_file() and os.access(candidate, os.X_OK):
return str(candidate)
return None
def _probe_ffmpeg_libs(ffmpeg_bin: str) -> dict[str, bool]:
"""Check which encoder/decoder libraries ffmpeg was compiled with.
Runs ffmpeg -encoders ONCE and greps for all known encoder names.
Each entry: (key, [search_strings]) — any match = available.
"""
try:
res = subprocess.run(
[ffmpeg_bin, "-encoders"],
capture_output=True, text=True, timeout=10,
)
output = res.stdout
except Exception:
output = ""
checks = [
("libsvtav1", ["libsvt_av1", "svt_av1"]),
("libaom", ["libaom_av1", "aom_av1"]),
("libvpx", ["libvpx-vp9", "libvpx_vp9", "vpx_vp9"]),
("libx265", ["libx265"]),
("libopus", ["libopus"]),
("libvorbis", ["libvorbis"]),
("flac", ["flac"]),
]
libs = {}
for lib_name, search_strings in checks:
libs[lib_name] = any(s in output for s in search_strings)
return libs
def _probe_av1an_version(av1an_bin: str) -> Optional[str]:
"""Extract av1an version string."""
try:
# Try --version first, fall back to parsing --help header
for args in (["--version"], ["--help"]):
res = subprocess.run(
[av1an_bin] + args,
capture_output=True, text=True, timeout=10,
)
output = res.stdout or res.stderr
match = re.search(r"av1an\s+([\d.]+(?:-\w+)?)", output, re.IGNORECASE)
if match:
return match.group(1)
if res.stdout.strip(): # If --version produced output but no version match
return res.stdout.strip().splitlines()[0][:60]
except Exception:
pass
return None
def _probe_ffmpeg_version(ffmpeg_bin: str) -> Optional[str]:
"""Extract ffmpeg version string."""
try:
res = subprocess.run(
[ffmpeg_bin, "-version"],
capture_output=True, text=True, timeout=10,
)
first_line = res.stdout.splitlines()[0] if res.stdout else ""
match = re.search(r"ffmpeg version (\S+)", first_line)
return match.group(1) if match else first_line[:60]
except Exception:
return None
def _probe_runtime_deps(distro: DistroProfile) -> tuple[dict[str, bool], list[str]]:
"""Check runtime dependencies that av1an needs to function.
Returns (deps_dict, missing_pkg_names).
Checks:
- VapourSynth + VSScript (av1an loads libvapoursynth-script.so via dlopen
to get the VSScript API — without this it panics with
'Failed to get VSScript API')
- Encoder binaries that av1an invokes directly (svt_av1, x265, vpxenc)
"""
deps: dict[str, bool] = {}
missing_pkgs: list[str] = []
# --- VapourSynth + VSScript (critical: av1an will panic without it) ---
# av1an is a Rust binary that dlopen's libvapoursynth-script.so and calls
# vsscript_init() / vsscript_createScript() / etc. It does NOT use the
# Python vapoursynth module. The shared library and the VSScript API
# library can be packaged separately on some distros (e.g. Debian has
# libvapoursynth-script-dev). We must check what av1an actually loads.
#
# IMPORTANT: We do NOT call vsscript_init() in our probe. VSScript's init
# internally calls Py_Initialize(), which crashes/fails when Python is
# already running (our probe runs inside a Python subprocess). Instead,
# we verify the shared library exists AND can be dlopen'd (CDLL constructor
# resolves all .so dependencies). If it loads, it will work for av1an.
vs_ok = False
vs_detail = ""
vs_ver_str = ""
vs_lib_path = None
# --- Step 1: Direct filesystem check (most reliable) ---
# Check well-known install paths. Works even if ldconfig cache is stale.
_vs_script_search = [
"/usr/lib/libvapoursynth-script.so",
"/usr/lib/libvapoursynth_script.so",
"/usr/lib64/libvapoursynth-script.so",
"/usr/lib/x86_64-linux-gnu/libvapoursynth-script.so",
"/usr/local/lib/libvapoursynth-script.so",
]
for p in _vs_script_search:
if Path(p).is_file():
vs_lib_path = p
break
# --- Step 2: Glob search on known lib dirs ---
if not vs_lib_path:
for lib_dir in ("/usr/lib", "/usr/lib64", "/usr/local/lib",
"/usr/lib/x86_64-linux-gnu"):
d = Path(lib_dir)
if d.is_dir():
matches = list(d.glob("libvapoursynth-script.so*"))
# Prefer unversioned .so over .so.0 (dev symlink)
for m in sorted(matches, key=lambda p: p.name):
vs_lib_path = str(m)
break
if vs_lib_path:
break
# --- Step 3: ldconfig -p ---
if not vs_lib_path:
try:
res = subprocess.run(
["ldconfig", "-p"], capture_output=True, text=True, timeout=5,
)
for line in res.stdout.splitlines():
if "libvapoursynth-script" in line or "libvapoursynth_script" in line:
parts = line.split("=>")
if len(parts) >= 2:
vs_lib_path = parts[1].strip().split()[0]
break
except Exception:
pass
# --- Step 4: ctypes.util.find_library ---
if not vs_lib_path:
try:
for name in ("vapoursynth-script", "vapoursynth_script"):
found = ctypes.util.find_library(name)
if found:
vs_lib_path = found
break
except Exception:
pass
# --- Step 5: Distro-specific package file listing ---
if not vs_lib_path:
pkg_query = {
"arch": ["pacman", "-Ql", "vapoursynth"],
"debian": ["dpkg", "-L", "vapoursynth"],
"redhat": ["rpm", "-ql", "vapoursynth"],
"suse": ["rpm", "-ql", "vapoursynth"],
}
query_cmd = pkg_query.get(distro.family)
if query_cmd:
try:
res = subprocess.run(
query_cmd, capture_output=True, text=True, timeout=10,
)
for line in res.stdout.splitlines():
line = line.strip()
# Skip directory entries and grab .so files
if "libvapoursynth-script" in line and line.endswith(".so"):
vs_lib_path = line
break
if "libvapoursynth-script" in line and ".so." in line and not vs_lib_path:
vs_lib_path = line # versioned .so as fallback
except Exception:
pass
# --- Step 6: dlopen smoke test (diagnostic only, NOT a gate) ---
# We do NOT gate on dlopen success. The library's constructor may call
# Py_Initialize() which conflicts with our Python subprocess, causing a
# silent segfault. av1an loads this library in its own fresh Rust process
# where no Python is running — so it works there even if our probe crashes.
# We only use dlopen to produce an optional warning.
vs_dlopen_warning = ""
if vs_lib_path:
try:
_escaped = vs_lib_path.replace("'", "\\'")
probe_code = (
"import ctypes; "
f"try: h = ctypes.CDLL('{_escaped}'); print('LOAD_OK') "
f"except OSError as e: print(f'LOAD_FAIL|{{e}}') "
f"except Exception as e: print(f'LOAD_OTHER|{{e}}') "
)
res = subprocess.run(
[sys.executable, "-c", probe_code],
capture_output=True, text=True, timeout=10,
)
out = res.stdout.strip()
if out == "LOAD_OK":
vs_ok = True
elif out:
vs_dlopen_warning = f"dlopen test failed: {out}"
vs_ok = True # file exists — let av1an try in its own process
else:
# subprocess produced no output — likely segfault in library
# constructor (Py_Initialize conflict). File still exists.
vs_dlopen_warning = "dlopen test produced no output (likely segfault in library constructor — not a problem for av1an)"
vs_ok = True
except subprocess.TimeoutExpired:
vs_dlopen_warning = "dlopen test timed out (library may have hanging constructor)"
vs_ok = True
except Exception as e:
vs_dlopen_warning = f"dlopen probe error: {e}"
vs_ok = True
# Final gate: library file was found on disk
if vs_lib_path and not vs_ok:
vs_ok = True # file found on disk is sufficient
if vs_ok:
vs_detail = vs_lib_path or "found"
# Try to get VapourSynth version from the core lib for diagnostics
try:
ver_probe = (
"import ctypes, ctypes.util; "
"_lib = ctypes.util.find_library('vapoursynth'); "
"if not _lib: "
" import subprocess as _sp; "
" _r = _sp.run(['ldconfig','-p'], capture_output=True, text=True, timeout=5); "
" _m = [l.split('=>')[1].strip().split()[0] for l in _r.stdout.splitlines() "
" if 'libvapoursynth.so.' in l and 'script' not in l]; "
" _lib = _m[0] if _m else None; "
"if _lib: "
" try: "
" _h = ctypes.CDLL(_lib); "
" _fn = _h.vapoursynth_version; "
" _fn.restype = ctypes.c_int; "
" print(_fn()) "
" except: pass "
)
res = subprocess.run(
[sys.executable, "-c", ver_probe],
capture_output=True, text=True, timeout=10,
)
ver_out = res.stdout.strip()
if ver_out and ver_out.isdigit() and int(ver_out) > 0:
vs_ver_str = f"R{ver_out}"
except Exception:
pass
else:
if not vs_detail:
vs_detail = "libvapoursynth-script.so not found (checked filesystem, ldconfig, and package manager)"
deps["vapoursynth"] = vs_ok
if not vs_ok:
# Determine which package(s) to suggest.
# Most distros bundle VSScript into the main 'vapoursynth' package,
# but some split it (Debian/Ubuntu: libvapoursynth-script-dev).
# Use the dedicated vsscript_pkg field if set, else fall back to dep_pkgs.
if distro.vsscript_pkg:
missing_pkgs.append(distro.vsscript_pkg)
elif "vapoursynth" in distro.dep_pkgs:
missing_pkgs.append(distro.dep_pkgs["vapoursynth"])
deps["vs_detail"] = False # extra key for the diagnostic message
else:
deps["vs_detail"] = True
# --- Encoder binaries (av1an invokes these directly, not via ffmpeg) ---
for enc_key, binary_names in distro.encoder_binaries.items():
found = False
for bin_name in binary_names:
if _find_binary(bin_name, distro) is not None:
found = True
break
deps[enc_key] = found
if not found:
# Map encoder key to dep_pkgs key
dep_key_map = {"svt_av1": "svt-av1", "vpx": "vpx", "x265": "x265"}
dep_key = dep_key_map.get(enc_key, enc_key)
if dep_key in distro.dep_pkgs:
pkg_name = distro.dep_pkgs[dep_key]
if pkg_name not in missing_pkgs:
missing_pkgs.append(pkg_name)
# --- ffprobe (needed for input file validation) ---
# Already checked in probe_environment() for the main binary, but let's
# make sure the dep dict reflects it for consistency.
# (ffprobe_path is set separately in probe_environment)
return deps, missing_pkgs, vs_detail, vs_ver_str, vs_dlopen_warning
def probe_environment() -> EnvProbe:
"""
Distro-aware binary detection + av1an flag compatibility probe +
ffmpeg library availability check.
"""
distro = detect_distro()
result = EnvProbe(distro=distro)
result.cpu = detect_cpu_topology()
cpu = result.cpu
result.warnings.append(f"Detected distro: {distro.name} (family={distro.family}, v{distro.version_id})")
result.warnings.append(
f"CPU: {cpu.model_name} — {cpu.physical_cores} physical cores x {cpu.threads_per_core} threads = {cpu.logical_threads} logical"
)
# --- Binary detection (distro-aware path search) ---
for name, attr in [("av1an", "av1an_path"), ("ffmpeg", "ffmpeg_path"), ("ffprobe", "ffprobe_path")]:
path = _find_binary(name, distro)
if path is None:
result.errors.append(f"Missing binary: {name}")
else:
setattr(result, attr, path)
# --- Install hint for missing binaries ---
if result.install_hint:
result.warnings.append(f"Install command: {result.install_hint}")
# --- FFmpeg version + library probe ---
if result.ffmpeg_path:
result.ffmpeg_version = _probe_ffmpeg_version(result.ffmpeg_path)
if result.ffmpeg_version:
result.warnings.append(f"FFmpeg version: {result.ffmpeg_version}")
result.ffmpeg_libs = _probe_ffmpeg_libs(result.ffmpeg_path)
# Warn about missing AUDIO libs (video codecs are handled by av1an's own
# encoder binaries — ffmpeg's video encoder list is irrelevant)
audio_lib_warnings = {
"Opus": "libopus",
"Vorbis": "libvorbis",
"FLAC": "flac",
}
for codec_label, lib_name in audio_lib_warnings.items():
if not result.ffmpeg_libs.get(lib_name, False):
result.warnings.append(f"FFmpeg missing encoder: {lib_name} ({codec_label} audio will not work)")
# --- Av1an version ---
if result.av1an_path:
result.av1an_version = _probe_av1an_version(result.av1an_path)
if result.av1an_version:
result.warnings.append(f"av1an version: {result.av1an_version}")
# --- Av1an flag compatibility probe ---
if result.av1an_path:
try:
help_out = subprocess.run(
[result.av1an_path, "--help"],
capture_output=True, text=True, timeout=15,
).stdout
result.av1an_flags = {
"worker": "--workers" if "--workers" in help_out else "-w",
"video_params": "--video-params" if "--video-params" in help_out else "-v",
"audio_params": "--audio-params" if "--audio-params" in help_out else "-a",
}
# Detect which encoder names this av1an build actually accepts.
# Substring matching on --help is unreliable (e.g. "svt" appears in
# descriptions but the real name may be "svtav1" or "svt_av1").
# Instead, pass a bogus encoder name and parse the clap error which
# lists all valid values.
svt_name = _detect_av1an_svt_encoder(result.av1an_path)
if svt_name:
result.av1an_flags["svt_name"] = svt_name
result.warnings.append(f"av1an SVT-AV1 encoder name: '{svt_name}'")
else:
# Absolute fallback — should rarely be needed
result.av1an_flags["svt_name"] = "svt_av1"
result.warnings.append("av1an SVT-AV1 encoder name: 'svt_av1' (fallback, not auto-detected)")
# Check for chunk-method availability (differs by av1an version/distro)
if "--chunk-method" in help_out:
result.av1an_flags["has_chunk_method"] = True
# Check for --temp flag (lets us relocate av1an work dir out of user folders)
if "--temp" in help_out:
result.av1an_flags["has_temp"] = True
result.av1an_flags["temp_flag"] = "--temp"
elif "-T" in help_out:
result.av1an_flags["has_temp"] = True
result.av1an_flags["temp_flag"] = "-T"
# Check for -s/segments flag (newer av1an)
if "-s" in help_out or "--scenes" in help_out:
result.av1an_flags["has_scenes"] = True
# Detect concat method: prefer mkvmerge, fall back to ffmpeg
if shutil.which("mkvmerge"):
result.av1an_flags["concat_method"] = "mkvmerge"
else:
result.av1an_flags["concat_method"] = "ffmpeg"
except Exception as e:
result.errors.append(f"av1an probe failed: {e}")
# --- Distro-specific notes ---
if distro.notes:
result.warnings.append(f"Distro note: {distro.notes}")
# --- Runtime dependency probe (vapoursynth, encoder binaries) ---
if result.av1an_path:
deps, missing_pkgs, vs_detail, vs_ver, vs_dlopen_warn = _probe_runtime_deps(distro)
result.runtime_deps = deps
result.missing_dep_pkgs = missing_pkgs
if deps.get("vapoursynth"):
result.vs_version = vs_ver
result.vs_script_lib = vs_detail
# Log VapourSynth/VSScript with extra detail
vs_status = "OK" if deps.get("vapoursynth") else "MISSING"
result.warnings.append(f"Dependency: vapoursynth (VSScript API) = {vs_status}")
if deps.get("vapoursynth"):
# vs_detail is the library path on success
result.warnings.append(f" VSScript lib: {vs_detail}")
if vs_dlopen_warn:
result.warnings.append(f" dlopen note: {vs_dlopen_warn}")
else:
# vs_detail is the failure reason
result.warnings.append(f" Reason: {vs_detail}")
# Log encoder binary deps (skip vs_detail key)
for dep_name, present in deps.items():
if dep_name in ("vapoursynth", "vs_detail"):
continue
status = "OK" if present else "MISSING"
result.warnings.append(f"Dependency: {dep_name} = {status}")
if missing_pkgs:
hint = result.dep_install_hint
result.errors.append(
f"Missing runtime dependencies: {', '.join(missing_pkgs)}"
)
if hint:
result.errors.append(f" FIX: {hint}")
return result
# ──────────────────────────────────────────────
# FFPREPBE VALIDATION
# ──────────────────────────────────────────────
def ffprobe_validate(filepath: Path, ffprobe_bin: str) -> Optional[dict]:
"""Returns stream info dict or None if invalid/unreadable."""
try:
res = subprocess.run(
[ffprobe_bin, "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", str(filepath)],
capture_output=True, text=True, timeout=30,
)
if res.returncode != 0:
return None
import json
return json.loads(res.stdout)
except Exception:
return None
def _verify_output_resolution(output_path: Path, ffprobe_bin: str, target_w: int, target_h: int) -> bool:
"""Verify that an encoded file actually has the requested output resolution.
Returns True if the output matches (or is within 2px due to force_divisible_by=2),
False otherwise.
"""
import json
try:
res = subprocess.run(
[ffprobe_bin, "-v", "quiet", "-print_format", "json",
"-show_streams", "-select_streams", "v:0", str(output_path)],
capture_output=True, text=True, timeout=15,
)
if res.returncode != 0:
return True # can't verify, don't block
data = json.loads(res.stdout)
streams = data.get("streams", [])
if not streams:
return True
ow = int(streams[0].get("width", 0) or 0)
oh = int(streams[0].get("height", 0) or 0)
# Allow 2px tolerance (force_divisible_by=2 rounding)
if abs(ow - target_w) <= 2 and abs(oh - target_h) <= 2:
return True
return False
except Exception:
return True # can't verify, don't block
def _detect_av1an_svt_encoder(av1an_bin: str) -> str | None:
"""Determine the exact encoder name av1an accepts for SVT-AV1.
Strategy (in order):
1. Run ``av1an --encoder __PROBE__`` and parse clap's error for
``[possible values: ...]``.
2. Parse ``--help`` for ``[default: <name>]`` next to ``--encoder``.
3. Regex fallback on the error output.
"""
import re
try:
# --- Method 1: clap error with possible values ---
res = subprocess.run(
[av1an_bin, "--encoder", "__PROBE_TEST__"],
capture_output=True, text=True, timeout=10,
)
stderr = res.stderr or ""
stdout = res.stdout or ""
combined = stderr + stdout
m = re.search(r"\[possible values:\s*([^\]]+)\]", combined)
if m:
values = [v.strip().rstrip(',') for v in m.group(1).split()]
for v in values:
if "svt" in v.lower():
return v
# --- Method 2: parse --help for encoder default value ---
help_res = subprocess.run(
[av1an_bin, "--help"],
capture_output=True, text=True, timeout=10,
)
help_text = (help_res.stdout or "") + (help_res.stderr or "")
# Look for pattern: --encoder <ENCODER> ... [default: svt-av1]
m2 = re.search(
r"--encoder\s+<ENCODER>.*?\[default:\s*(\S+?)\]",
help_text, re.DOTALL,
)
if m2:
return m2.group(1)
# --- Method 3: regex fallback on the error output ---
for line in combined.splitlines():
for token in re.findall(r"\bsvt[a-z_-]*av1[a-z_-]*\b", line, re.IGNORECASE):
return token
for token in re.findall(r"\bsvtav1\b", line, re.IGNORECASE):
return token
return None
except Exception:
return None
def _av1an_env() -> dict[str, str]:
"""Build an env dict for subprocess that includes ~/.local/lib in LD_LIBRARY_PATH.
When VapourSynth is built from git and installed to ~/.local/, the linker
won't find libvapoursynth-script.so unless LD_LIBRARY_PATH points there.
This function ensures every av1an invocation inherits that path.
"""
env = os.environ.copy()
local_lib = str(Path.home() / ".local" / "lib")
existing = env.get("LD_LIBRARY_PATH", "")
if local_lib not in existing:
env["LD_LIBRARY_PATH"] = f"{local_lib}:{existing}".rstrip(":")
return env
def _av1an_vsscript_smoke_test(
av1an_bin: str,
ffmpeg_bin: str,
av1an_flags: dict,
svt_name: str = "svt_av1",
timeout: int = 30,
) -> tuple[bool, str]:
"""Pre-flight test: create a tiny video and try to run av1an on it.
This catches 'Failed to get VSScript API' panics BEFORE the real queue
starts. File-existence checks for libvapoursynth-script.so pass even
when the ABI is incompatible (av1an's Rust vapoursynth crate built
against a different VS version). Only actually invoking av1an reveals
the mismatch.
Returns (ok, detail_message).
ok=True -> av1an initialized VSScript successfully.
ok=False -> av1an panicked or failed; detail_message explains why.
"""
import tempfile
with tempfile.TemporaryDirectory(prefix="av1an_smoke_") as tmpdir:
test_in = Path(tmpdir) / "test_smoke.mkv"
test_out = Path(tmpdir) / "test_smoke_out.mkv"
# Create a 1-second 64x64 black video (video-only is enough to
# trigger VSScript init in av1an — no audio needed).
gen_cmd = [
ffmpeg_bin,
"-f", "lavfi", "-i", "color=c=black:s=64x64:d=1:r=24",
"-t", "1", "-pix_fmt", "yuv420p", "-an", "-y", str(test_in),
]
try:
res = subprocess.run(gen_cmd, capture_output=True, text=True, timeout=15)
if res.returncode != 0:
return False, f"ffmpeg test-video failed (rc={res.returncode}): {(res.stderr or '')[-200:]}"
except Exception as e:
return False, f"Could not generate smoke test video: {e}"
if not test_in.exists():
return False, "Smoke test video was not created by ffmpeg"
# Build minimal av1an command
worker_flag = av1an_flags.get("worker", "--workers")
vparams_flag = av1an_flags.get("video_params", "--video-params")
aparams_flag = av1an_flags.get("audio_params", "--audio-params")
cmd = [
av1an_bin,
"-i", str(test_in),
worker_flag, "1",
"--encoder", svt_name,
vparams_flag, "--preset 8 --crf 40 --keyint 240",
"-o", str(test_out),
]
# Use chunk-method select if available (triggers VSScript init)
if av1an_flags.get("has_chunk_method"):
cmd.extend(["--chunk-method", "select"])
try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout,
env=_av1an_env())
stderr = res.stderr or ""
if res.returncode == 0 and test_out.exists():
# av1an succeeded — VSScript works
test_out.unlink(missing_ok=True)
return True, "av1an VSScript init OK"
if "Failed to get VSScript API" in stderr:
return False, "VSScript_API_INCOMPAT"
# Check for invalid encoder name — means probe was wrong
if "invalid value" in stderr and "--encoder" in stderr:
return False, f"INVALID_ENCODER: {stderr[-200:]}"
# Other failure — could be missing encoder binary, etc.
tail = stderr[-300:] if stderr else "(no stderr)"
return True, f"av1an smoke test had non-VSScript error (rc={res.returncode}): {tail}"
except subprocess.TimeoutExpired:
return True, "av1an smoke test timed out (not a VSScript issue)"
except Exception as e:
return True, f"av1an smoke test exception (not a VSScript issue): {e}"
# ──────────────────────────────────────────────
# TEMP DIRECTORY MANAGEMENT
# ──────────────────────────────────────────────
_APP_CACHE_DIR: Path | None = None
def _get_app_temp_dir() -> Path:
"""Return the shared temp directory for all intermediate files.
Priority:
1. ``~/.cache/OpenTranscode/tmp/`` (XDG-compliant, persistent across reboots)
2. ``/tmp/OpenTranscode/`` (fallback if home cache is unwritable)
The directory is created on first call. All temp intermediates
(pre-scaled MKVs, av1an work dirs) go here so the user's video
folders stay clean.
"""
global _APP_CACHE_DIR
if _APP_CACHE_DIR is not None:
return _APP_CACHE_DIR
# Try XDG cache dir first
xdg_cache = os.environ.get("XDG_CACHE_HOME", "")
if xdg_cache:
candidate = Path(xdg_cache) / "OpenTranscode" / "tmp"
else:
candidate = Path.home() / ".cache" / "OpenTranscode" / "tmp"
try:
candidate.mkdir(parents=True, exist_ok=True)
_APP_CACHE_DIR = candidate
return _APP_CACHE_DIR
except (OSError, PermissionError):
pass
# Fallback: /tmp/OpenTranscode
fallback = Path("/tmp/OpenTranscode")
try:
fallback.mkdir(parents=True, exist_ok=True)
_APP_CACHE_DIR = fallback
return _APP_CACHE_DIR
except (OSError, PermissionError):
# Last resort: system temp
_APP_CACHE_DIR = Path(tempfile.gettempdir()) / "OpenTranscode"
_APP_CACHE_DIR.mkdir(parents=True, exist_ok=True)
return _APP_CACHE_DIR
def _temp_path_for(file_path: Path, suffix: str = ".scaled_tmp.mkv") -> Path:
"""Build a unique temp path for *file_path* inside the app temp dir.
Uses a short hash of the original absolute path to avoid collisions
when files in different subdirs share the same stem.
"""
tmp_dir = _get_app_temp_dir()
# Hash the absolute source path for uniqueness
path_hash = hashlib.sha256(str(file_path.resolve()).encode()).hexdigest()[:12]
return tmp_dir / f"{file_path.stem}.{path_hash}{suffix}"
# ──────────────────────────────────────────────
# ENCODER WORKER (QThread, from PySide6 ver, extended)
# ──────────────────────────────────────────────
class EncoderWorker(QThread):
log_msg = Signal(str)
progress_msg = Signal(str, int, int) # (filename, current, total)
finished_queue = Signal(int, int) # (success_count, fail_count)
def __init__(
self,
in_dir: Path,
out_dir: Path,
video_codec: VideoCodecProfile,
audio_profile: AudioProfile,
container: ContainerProfile,
crf: int,
preset_label: str,
delete_source: bool,
env: EnvProbe,
extensions: set[str],
resolution: ResolutionProfile,
audio_level_db: float = 0.0,
use_ffmpeg_fallback: bool = False,
subtitle_lang: Optional[str] = None,
):
super().__init__()
self.in_dir = in_dir
self.out_dir = out_dir
self.video_codec = video_codec
self.audio_profile = audio_profile
self.container = container
self.crf = crf
self.preset_val = video_codec.preset_map.get(preset_label, 6)
self.delete_source = delete_source
self.env = env
self.extensions = extensions
self.resolution = resolution
self.audio_level_db = audio_level_db
self.use_ffmpeg_fallback = use_ffmpeg_fallback
self.subtitle_lang = subtitle_lang
self._stop = False
self._current_temps: list[Path] = [] # temps for the file currently being processed
self._sources_to_delete: list[Path] = [] # sources deferred for deletion after final cleanup
self.success_count = 0
self.fail_count = 0
# Resolve temp dir once per worker (shared across all files in the queue)
self._temp_dir = _get_app_temp_dir()
def _ffmpeg_fallback_encode(
self,
file_path: Path,
encode_input: Path,
output_f: Path,
) -> bool:
"""Encode a single file using pure ffmpeg (av1an fallback path).
Used when av1an cannot initialize VapourSynth. No chunk-parallel
mode, but ffmpeg uses multithreaded encoding internally.
Returns True on success, False on failure.
"""
# Check if ffmpeg has the video encoder we need
ffmpeg_enc = self.video_codec.ffmpeg_encoder
# Map our ffmpeg_encoder name to the ffmpeg_libs key
ffmpeg_lib_key = {
"libsvtav1": "libsvtav1",
"libaom-av1": "libaom",
"libvpx-vp9": "libvpx",
"libx265": "libx265",
}.get(ffmpeg_enc, ffmpeg_enc)
if not self.env.ffmpeg_libs.get(ffmpeg_lib_key, False):
self.log_msg.emit(
f" FATAL: ffmpeg does not have '{ffmpeg_enc}' encoder. "
f"Cannot fall back. Install a ffmpeg build with {ffmpeg_enc} support."
)
return False
v_args = self.video_codec.ffmpeg_vargs_fn(self.crf, self.preset_val)
# Belt-and-suspenders: if a target resolution is set, inject -vf scale
# directly into the ffmpeg command. This guarantees the output resolution
# matches the dropdown even if the intermediate pre-scale was bypassed.
vf_scale_args: list[str] = []
if self.resolution.width is not None and self.resolution.height is not None:
vf_scale_args = [
"-vf", (
f"scale={self.resolution.width}:{self.resolution.height}:"
f"force_original_aspect_ratio=decrease:force_divisible_by=2"
),
]
# Audio args from profile
audio_args = list(self.audio_profile.params)
if abs(self.audio_level_db) > 0.01:
per_file_gain = self._analyze_audio_loudness(file_path)
if per_file_gain is not None and abs(per_file_gain) > 0.01:
audio_args.extend(["-af", f"volume={per_file_gain:+.1f}dB"])
else:
static_db = f"{self.audio_level_db:+.1f}".replace("+", "")
audio_args.extend(["-af", f"volume={static_db}dB"])
cmd = [
self.env.ffmpeg_path,
"-i", str(encode_input),
] + vf_scale_args + v_args + audio_args + [
"-movflags", "+faststart", # useful for MKV/WebM streaming
"-y",
str(output_f),
]
try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=7200)
if res.returncode == 0 and output_f.exists():
src_size = file_path.stat().st_size
out_size = output_f.stat().st_size
ratio = out_size / src_size if src_size > 0 else 0
if out_size > (src_size * 0.05):
return True
else:
self.log_msg.emit(
f" INTEGRITY: output only {ratio * 100:.1f}% of source."
)
output_f.unlink(missing_ok=True)
return False
else:
stderr_snip = (res.stderr or "")[-300:]
self.log_msg.emit(
f" ffmpeg error (rc={res.returncode}): {stderr_snip.strip()}"
)
return False
except subprocess.TimeoutExpired:
self.log_msg.emit(f" TIMEOUT: ffmpeg exceeded 2h limit.")
return False
except Exception as e:
self.log_msg.emit(f" SYSTEM ERROR: {e}")
return False
def run(self):
# Use physical core count, leave 1 for OS/UI breathing room
worker_count = max(1, self.env.cpu.physical_cores - 1)
phys = self.env.cpu.physical_cores
logical = self.env.cpu.logical_threads
# Collect all valid files first (for progress tracking)
# Exclude our own temp intermediates from previous failed runs.
all_files = sorted(
f for f in self.in_dir.rglob("*")
if f.is_file()
and f.suffix.lower() in self.extensions
and not f.name.endswith(".scaled_tmp.mkv")
)
total = len(all_files)
if total == 0:
self.log_msg.emit("INFO: No matching files found in source directory.")
self.finished_queue.emit(0, 0)
return
# ── Mode banner ──
# use_ffmpeg_fallback is set by the main thread's pre-flight check.
# If True, the main thread already logged the fallback reason.
if self.use_ffmpeg_fallback:
self.log_msg.emit(
f"FFmpeg fallback: {self.video_codec.ffmpeg_encoder} on {phys} cores "
f"(single-pass, no chunk-parallel)"
)
else:
self.log_msg.emit(
f"Chunk-parallel: {worker_count} workers on {phys} physical cores "
f"({logical} logical threads, {self.env.cpu.threads_per_core}T/core)"
)
self.log_msg.emit(f"Found {total} file(s) to process.")
self.log_msg.emit(f"Temp dir: {self._temp_dir}")
# ── Pre-scan: show each file's source → output resolution ──
needs_scale = (
self.resolution.width is not None
and self.resolution.height is not None
)
if needs_scale:
self.log_msg.emit(f"Output resolution: {self.resolution.width}x{self.resolution.height} ({self.resolution.aspect_label})")
else:
self.log_msg.emit("Output resolution: Original (no scaling)")
self.log_msg.emit("─── FILE RESOLUTION MAP ───")
self._file_res_map: dict[Path, tuple] = {} # file -> (src_w, src_h, out_w, out_h)
if self.env.ffprobe_path:
for f in all_files:
info = ffprobe_validate(f, self.env.ffprobe_path)
sw, sh = None, None
if info:
for s in info.get("streams", []):
if s.get("codec_type") == "video":
sw = int(s.get("width", 0) or 0)
sh = int(s.get("height", 0) or 0)
break
if sw and sh:
ow, oh = (self.resolution.width, self.resolution.height) if needs_scale else (sw, sh)
self._file_res_map[f] = (sw, sh, ow, oh)
arrow = "->" if needs_scale else "="
action = "" if needs_scale or sw == ow else " (no change)"
self.log_msg.emit(f" {f.name:<40s} {sw:>5}x{sh:<5} {arrow} {ow:>5}x{oh}{action}")
else:
self._file_res_map[f] = (None, None, self.resolution.width if needs_scale else None, self.resolution.height if needs_scale else None)
self.log_msg.emit(f" {f.name:<40s} (unknown resolution)")
else:
self.log_msg.emit(" (ffprobe unavailable — resolution map skipped)")
self.log_msg.emit("───────────────────────────")
scale_filter = (
f"scale={self.resolution.width}:{self.resolution.height}:"
f"force_original_aspect_ratio=decrease:force_divisible_by=2,"
f"pad={self.resolution.width}:{self.resolution.height}:(ow-iw)/2:(oh-ih)/2"
) if needs_scale else ""
for idx, file_path in enumerate(all_files, 1):
if self._stop:
self.log_msg.emit("STOP: Aborted by user.")
break
self.progress_msg.emit(file_path.name, idx, total)
self.log_msg.emit(f"[{idx}/{total}] Encoding: {file_path.name}")
# --- ffprobe pre-validation ---
# Reuse pre-scanned dimensions if available, otherwise probe now
prescan = self._file_res_map.get(file_path)
src_w, src_h = (prescan[0], prescan[1]) if prescan else (None, None)
info = None
if self.env.ffprobe_path:
info = ffprobe_validate(file_path, self.env.ffprobe_path)
if info is None:
self.log_msg.emit(f"WARN: ffprobe could not read {file_path.name} — attempting encode anyway.")
else:
duration = float(info.get("format", {}).get("duration", 0))
has_video = any(s.get("codec_type") == "video" for s in info.get("streams", []))
if not has_video:
self.log_msg.emit(f"SKIP: {file_path.name} has no video stream.")
self.fail_count += 1
continue
if duration < 0.5:
self.log_msg.emit(f"SKIP: {file_path.name} is too short ({duration:.1f}s).")
self.fail_count += 1
continue
# Extract dims if pre-scan didn't have them
if not src_w or not src_h:
for s in info.get("streams", []):
if s.get("codec_type") == "video":
src_w = int(s.get("width", 0) or 0)
src_h = int(s.get("height", 0) or 0)
break
# --- Determine actual output resolution ---
if src_w and src_h:
out_w, out_h = src_w, src_h
if needs_scale:
out_w, out_h = self.resolution.width, self.resolution.height
self.log_msg.emit(f" Source: {src_w}x{src_h} -> Output: {out_w}x{out_h}")
else:
if needs_scale:
self.log_msg.emit(f" Source: unknown -> Output: {self.resolution.width}x{self.resolution.height}")
else:
self.log_msg.emit(f" Source: unknown -> Output: original")
# --- Pre-scale with ffmpeg if target resolution selected ---
# ALL intermediates (scaled files, av1an work dirs) go to the app
# temp directory so the user's video folders stay clean.
encode_input = file_path
av1an_work: Path | None = None
if needs_scale:
try:
temp_scaled = _temp_path_for(file_path, ".scaled_tmp.mkv")
self._current_temps.append(temp_scaled)
# Use libx265 lossless for the intermediate — NOT ffv1.
# ffv1 is not supported by VapourSynth source plugins (bestsource,
# ffms2, lsmash), so av1an's chunking pipeline produces an empty
# pipe and the encoder emits "Fatal: Failed to open input file".
# libx265 -crf 0 is bit-for-bit lossless, fast at ultrafast preset,
# and HEVC-in-MKV is universally supported by every VS plugin.
scale_cmd = [
self.env.ffmpeg_path,
"-i", str(file_path),
"-vf", scale_filter,
"-c:v", "libx265",
"-crf", "0",
"-preset", "ultrafast",
"-pix_fmt", "yuv420p", # force 8-bit 4:2:0
"-y",
str(temp_scaled),
]
self.log_msg.emit(f" Scaling {src_w or '?'}x{src_h or '?'} -> {self.resolution.width}x{self.resolution.height}...")
scale_res = subprocess.run(
scale_cmd, capture_output=True, text=True, timeout=1800,
)
if scale_res.returncode == 0 and temp_scaled.exists():
encode_input = temp_scaled
scaled_size = temp_scaled.stat().st_size / 1_048_576
self.log_msg.emit(f" Pre-scale OK ({scaled_size:.1f} MB intermediate)")
else:
stderr_snip = (scale_res.stderr or "")[-200:]
self.log_msg.emit(
f" FAIL: Pre-scale failed for {file_path.name} (rc={scale_res.returncode}). "
f"Output resolution must match selected {self.resolution.width}x{self.resolution.height}."
)
if stderr_snip.strip():
self.log_msg.emit(f" ffmpeg stderr: {stderr_snip.strip()}")
temp_scaled.unlink(missing_ok=True)
self._cleanup_current_temps()
self.fail_count += 1
continue
except Exception as e:
self.log_msg.emit(
f" FAIL: Pre-scale error for {file_path.name}: {e}. "
f"Cannot guarantee output resolution {self.resolution.width}x{self.resolution.height}."
)
self._cleanup_current_temps()
self.fail_count += 1
continue
# --- Ensure av1an work dir lands in the temp directory ---
# av1an creates its work dir as {input_path}.av1an by default.
# We do NOT use av1an's --temp flag because it causes "Error: End of file"
# during scene detection when the input file is in the same directory
# as --temp (av1an 0.5.2-unstable). Instead, we ensure the -i argument
# always points into the temp dir (pre-scaled files already live there;
# for no-scale we create a symlink).
if not encode_input.is_relative_to(self._temp_dir):
symlink_path = _temp_path_for(file_path, encode_input.suffix)
try:
symlink_path.unlink(missing_ok=True)
symlink_path.symlink_to(file_path.resolve())
self._current_temps.append(symlink_path)
encode_input = symlink_path
except OSError as e:
self.log_msg.emit(
f" WARN: Could not create symlink in temp dir: {e}. "
f"av1an work dir will be created next to source file."
)
# Track the work dir where av1an will actually create it
av1an_work = Path(f"{encode_input}.av1an")
self._current_temps.append(av1an_work)
# --- Build output path (preserve directory structure) ---
rel_path = file_path.relative_to(self.in_dir)
target_dir = self.out_dir / rel_path.parent
target_dir.mkdir(parents=True, exist_ok=True)
ext = self.container.ext
# Always add resolution suffix when a target resolution is selected
res_suffix = f"_{self.resolution.width}x{self.resolution.height}" if needs_scale else ""
output_f = target_dir / f"{file_path.stem}{res_suffix}_archived.{ext}"
# ── Choose encode path: av1an or ffmpeg fallback ──
if self.use_ffmpeg_fallback:
# ── Pure ffmpeg encode path ──
self.log_msg.emit(f" Mode: ffmpeg ({self.video_codec.ffmpeg_encoder})")
success = self._ffmpeg_fallback_encode(
file_path, encode_input, output_f,
)
if success:
# Post-encode resolution verification
if needs_scale and self.env.ffprobe_path:
if not _verify_output_resolution(
output_f, self.env.ffprobe_path,
self.resolution.width, self.resolution.height,
):
self.log_msg.emit(
f" FAIL: Output resolution verification failed for {file_path.name}. "
f"Expected {self.resolution.width}x{self.resolution.height}."
)
output_f.unlink(missing_ok=True)
self._cleanup_current_temps()
self.fail_count += 1
continue
src_size = file_path.stat().st_size
out_size = output_f.stat().st_size
ratio = out_size / src_size if src_size > 0 else 0
# Duration integrity check (>= 95% of source)
dur_ok = True
dur_info = ""
if self.env.ffprobe_path:
src_dur = ffprobe_duration(file_path, self.env.ffprobe_path)
out_dur = ffprobe_duration(output_f, self.env.ffprobe_path)
if src_dur and out_dur:
dur_ratio = out_dur / src_dur
dur_ok = dur_ratio >= 0.95
dur_info = f", duration {out_dur:.1f}s/{src_dur:.1f}s ({dur_ratio * 100:.0f}%)"
if not dur_ok:
self.fail_count += 1
self.log_msg.emit(f"INTEGRITY FAIL: {file_path.name} — duration{dur_info}")
output_f.unlink(missing_ok=True)
self._cleanup_current_temps()
continue
# Mux subtitle if requested (needs source file intact)
if self.subtitle_lang:
self._mux_subtitle(file_path, output_f)
self.success_count += 1
self.log_msg.emit(
f"SUCCESS: {file_path.name} "
f"({src_size / 1_048_576:.1f}MB -> {out_size / 1_048_576:.1f}MB, "
f"{ratio * 100:.0f}%{dur_info})"
)
# Defer source deletion until after final cleanup
if self.delete_source:
self._sources_to_delete.append(file_path)
self._cleanup_current_temps()
else:
self.fail_count += 1
self._cleanup_current_temps()
continue
# ── av1an encode path (original) ──
# Resolve encoder name with probe data
enc = self.video_codec.av1an_encoder
if enc in ("svt_av1", "svt") and "svt_name" in self.env.av1an_flags:
enc = self.env.av1an_flags["svt_name"]
# Build params via config table (no if/else)
v_params = self.video_codec.params_fn(self.crf, self.preset_val)
# Audio params: dual-pass normalization per file, or simple volume
audio_parts = list(self.audio_profile.params)
if abs(self.audio_level_db) > 0.01:
per_file_gain = self._analyze_audio_loudness(file_path)
if per_file_gain is not None and abs(per_file_gain) > 0.01:
audio_parts.extend(["-af", f"volume={per_file_gain:+.1f}dB"])
else:
# Fallback to knob's static value if analysis failed
static_db = f"{self.audio_level_db:+.1f}".replace("+", "")
audio_parts.extend(["-af", f"volume={static_db}dB"])
self.log_msg.emit(f" Audio: static gain {self.audio_level_db:+.1f} dB (analysis unavailable)")
audio_str = " ".join(audio_parts)
cmd = [
self.env.av1an_path,
"-i", str(encode_input),
self.env.av1an_flags.get("worker", "--workers"), str(worker_count),
]
# Chunk method: let av1an auto-select.
# The intermediate (when pre-scaling) is now HEVC lossless in MKV,
# which all VS plugins handle correctly, so auto-select is safe.
chunk_method = self.env.av1an_flags.get("chunk_method_override")
if chunk_method:
cmd.extend(["--chunk-method", chunk_method])
self.log_msg.emit(f" Chunking: {chunk_method or 'auto'} (av1an default if no override)")
cmd.extend([
"--encoder", enc,
self.env.av1an_flags.get("video_params", "--video-params"), v_params,
self.env.av1an_flags.get("audio_params", "--audio-params"), audio_str,
"--concat", self.env.av1an_flags.get("concat_method", "ffmpeg"),
"-o", str(output_f),
])
# Log the full av1an command for debugging
self.log_msg.emit(f" CMD: {' '.join(cmd)}")
try:
res = subprocess.run(cmd, capture_output=True, text=True, timeout=7200,
env=_av1an_env())
if res.returncode == 0 and output_f.exists():
src_size = file_path.stat().st_size
out_size = output_f.stat().st_size
ratio = out_size / src_size if src_size > 0 else 0
# Integrity: output must be at least 5% of source
if out_size > (src_size * 0.05):
# Post-encode resolution verification
if needs_scale and self.env.ffprobe_path:
if not _verify_output_resolution(
output_f, self.env.ffprobe_path,
self.resolution.width, self.resolution.height,
):
self.fail_count += 1
self.log_msg.emit(
f" FAIL: Output resolution verification failed for {file_path.name}. "
f"Expected {self.resolution.width}x{self.resolution.height}."
)
output_f.unlink(missing_ok=True)
self._cleanup_current_temps()
continue
# Duration integrity check (>= 95% of source)
dur_ok = True
dur_info = ""
if self.env.ffprobe_path:
src_dur = ffprobe_duration(file_path, self.env.ffprobe_path)
out_dur = ffprobe_duration(output_f, self.env.ffprobe_path)
if src_dur and out_dur:
dur_ratio = out_dur / src_dur
dur_ok = dur_ratio >= 0.95
dur_info = f", duration {out_dur:.1f}s/{src_dur:.1f}s ({dur_ratio * 100:.0f}%)"
if not dur_ok:
self.fail_count += 1
self.log_msg.emit(f"INTEGRITY FAIL: {file_path.name} — duration{dur_info}")
output_f.unlink(missing_ok=True)
self._cleanup_current_temps()
continue
# Mux subtitle if requested (needs source file intact)
if self.subtitle_lang:
self._mux_subtitle(file_path, output_f)
self.success_count += 1
self.log_msg.emit(
f"SUCCESS: {file_path.name} "
f"({src_size / 1_048_576:.1f}MB -> {out_size / 1_048_576:.1f}MB, "
f"{ratio * 100:.0f}%{dur_info})"
)
# Defer source deletion until after final cleanup
if self.delete_source:
self._sources_to_delete.append(file_path)
else:
self.fail_count += 1
self.log_msg.emit(
f"ERROR: Integrity check failed for {file_path.name} "
f"(output only {ratio * 100:.1f}% of source)."
)
# Remove corrupt output
output_f.unlink(missing_ok=True)
else:
self.fail_count += 1
stderr_full = res.stderr or ""
stderr_tail = stderr_full[-300:]
self.log_msg.emit(f"FAIL: {file_path.name} (code {res.returncode}) | {stderr_tail}")
# Detect known av1an crash patterns and provide actionable fixes
if "Failed to get VSScript API" in stderr_full:
self.log_msg.emit("")
self.log_msg.emit("DIAGNOSIS: av1an cannot initialize VapourSynth.")
self.log_msg.emit(
" This means the av1an binary was compiled against a different "
"VapourSynth version than what is currently installed."
)
distro = self.env.distro
if distro.family == "arch":
self.log_msg.emit(
" FIX (Arch): Rebuild av1an against current VapourSynth:"
)
self.log_msg.emit(" yay -S av1an (AUR rebuild)")
self.log_msg.emit(" or: cargo install av1an --force --locked")
self.log_msg.emit(
" VapourSynth R77+ changed the VSScript API. "
"Your av1an binary needs to be recompiled."
)
elif distro.family == "debian":
self.log_msg.emit(
" FIX (Debian/Ubuntu): Ensure matching versions:"
)
self.log_msg.emit(" sudo apt install vapoursynth libvapoursynth-script-dev av1an")
else:
self.log_msg.emit(
" FIX: Reinstall av1an to rebuild against your current "
"VapourSynth. Check your distro's package manager."
)
self.log_msg.emit("")
# Stop the queue — all remaining files will hit the same crash
self._stop = True
self.log_msg.emit("STOP: Skipping remaining files (same VSScript issue).")
elif "No usable encoder found" in stderr_full:
self.log_msg.emit("")
self.log_msg.emit("DIAGNOSIS: av1an cannot find the encoder binary.")
self.log_msg.emit(
" Verify the encoder is installed and in PATH."
)
self._stop = True
self.log_msg.emit("STOP: Skipping remaining files (same encoder issue).")
elif "not a valid input file" in stderr_full or "could not open input" in stderr_full.lower():
self.log_msg.emit("")
self.log_msg.emit("DIAGNOSIS: av1an cannot read this input file.")
self.log_msg.emit(" Try playing the file with ffplay to verify it's not corrupt.")
except subprocess.TimeoutExpired:
self.fail_count += 1
self.log_msg.emit(f"TIMEOUT: {file_path.name} exceeded 2h limit.")
except Exception as e:
self.fail_count += 1
self.log_msg.emit(f"SYSTEM ERROR: {file_path.name} — {e}")
finally:
# Always clean this file's temps before moving to next
self._cleanup_current_temps()
# ── Final cleanup pass: residual sweep ──
self._final_cleanup_sweep()
# ── Deferred source deletion (only after all cleanup is done) ──
if self._sources_to_delete:
deleted = 0
for src in self._sources_to_delete:
try:
if src.exists():
src.unlink()
deleted += 1
except Exception:
pass
self.log_msg.emit(f"CLEANED: Removed {deleted} source file(s) after verified transcode.")
self._sources_to_delete.clear()
if self.use_ffmpeg_fallback:
self.log_msg.emit(
f"QUEUE COMPLETE (ffmpeg fallback). Success: {self.success_count}, Failed: {self.fail_count}."
)
else:
self.log_msg.emit(f"QUEUE COMPLETE. Success: {self.success_count}, Failed: {self.fail_count}.")
self.finished_queue.emit(self.success_count, self.fail_count)
def _cleanup_current_temps(self):
"""Remove all tracked temp files/dirs for the current file.
Resilient: each removal is try/except'd individually so one bad path
doesn't block the rest. Clears the tracking list when done.
"""
for tf in self._current_temps:
try:
if tf.is_dir():
shutil.rmtree(str(tf), ignore_errors=True)
elif tf.exists():
tf.unlink()
except Exception:
pass
self._current_temps.clear()
def _final_cleanup_sweep(self):
"""Residual sweep to catch any orphaned temp files.
Primary target: the app temp directory (~/.cache/OpenTranscode/tmp/).
Since av1an --temp creates dirs named by input stem (no fixed suffix),
we clean the ENTIRE temp directory contents as a nuclear option.
Also scans in_dir/out_dir as a safety net for legacy temp files.
"""
swept = 0
# Nuclear cleanup of the app temp dir — remove everything in it.
# This is safe because the dir ONLY contains our intermediates.
if self._temp_dir.is_dir():
for hit in self._temp_dir.iterdir():
try:
if hit.is_dir():
shutil.rmtree(str(hit), ignore_errors=True)
else:
hit.unlink(missing_ok=True)
swept += 1
except Exception:
pass
# Safety-net sweep of user directories (for legacy temp files
# written by older versions that placed temps next to source files)
legacy_patterns = ["*.scaled_tmp.mkv", "*.av1an", "*_encodes", "*.*.av1an"]
for search_dir in (self.in_dir, self.out_dir):
if not search_dir.is_dir():
continue
for pattern in legacy_patterns:
for hit in search_dir.rglob(pattern):
try:
if hit.is_dir():
shutil.rmtree(str(hit), ignore_errors=True)
else:
hit.unlink(missing_ok=True)
swept += 1
except Exception:
pass
# Also clean any orphans still in _current_temps (e.g. stop/crash mid-loop)
self._cleanup_current_temps()
if swept:
self.log_msg.emit(f"CLEANUP: Swept {swept} residual temp file(s)/dir(s).")
# ── Audio loudness analysis (dual-pass normalization) ──
def _analyze_audio_loudness(self, file_path: Path) -> Optional[float]:
"""Dual-pass loudnorm analysis for a single file.
Pass 1: Run loudnorm in analysis-only mode to measure the file's current
integrated loudness (I) and true peak (TP).
Returns the dB gain to apply, or None if analysis fails (falls back to
the knob's static value).
"""
if not self.env.ffmpeg_path:
return None
if abs(self.audio_level_db) < 0.01:
return None # knob is at 0 — no normalization requested
target_lufs = self.audio_level_db # knob value IS the target LUFS
try:
# Pass 1: analyze current loudness
analysis_cmd = [
self.env.ffmpeg_path,
"-i", str(file_path),
"-af", (
f"loudnorm=I={target_lufs}:TP=-1.5:LRA=11:"
f"print_format=json"
),
"-f", "null", "-",
]
res = subprocess.run(
analysis_cmd, capture_output=True, text=True, timeout=120,
)
# Parse the JSON stats from stderr (loudnorm prints to stderr)
stderr = res.stderr or ""
# Find the JSON block
json_match = re.search(r'\{[^{}]*"input_i"[^{}]*\}', stderr, re.DOTALL)
if not json_match:
return None
import json
stats = json.loads(json_match.group())
input_i = float(stats.get("input_i", "-99"))
input_tp = float(stats.get("input_tp", "-99"))
target_tp = float(stats.get("target_tp", "-1.5"))
# If file is already silent or near-silent, skip
if input_i <= -70:
return None
# Compute the gain loudnorm would apply
gain_db = target_lufs - input_i
# Pass 2 concept: check if applying this gain would push peaks
# above our ceiling. The ceiling is target_tp (default -1.5 dBTP).
# We want 15% headroom below that ceiling.
headroom_db = abs(target_tp) * 0.15
peak_ceiling = target_tp + headroom_db
# If the file's true peak + gain would exceed the ceiling, clamp
projected_peak = input_tp + gain_db
if projected_peak > peak_ceiling:
gain_db = peak_ceiling - input_tp
self.log_msg.emit(
f" Audio: {input_i:.1f} LUFS -> {target_lufs:.1f} LUFS "
f"(gain {gain_db:+.1f} dB, peak {input_tp:.1f} -> "
f"{input_tp + gain_db:.1f} dBTP)"
)
return gain_db
except Exception as e:
self.log_msg.emit(f" Audio: loudnorm analysis failed ({e}), using knob value")
return None
# ── Subtitle extraction & muxing ──
def _find_subtitle_stream(self, source: Path, lang: str) -> tuple[Optional[int], str]:
"""Find subtitle stream in source matching language code.
Prefers forced disposition tracks. Returns (stream_index, codec_name)."""
if not self.env.ffprobe_path:
return (None, "")
info = ffprobe_validate(source, self.env.ffprobe_path)
if not info:
return (None, "")
forced_match = None
any_match = None
for stream in info.get("streams", []):
if stream.get("codec_type") != "subtitle":
continue
tags = stream.get("tags", {})
if tags.get("language", "").lower() != lang.lower():
continue
idx = stream.get("index")
codec = stream.get("codec_name", "")
disposition = stream.get("disposition", {})
if disposition.get("forced") and forced_match is None:
forced_match = (idx, codec)
if any_match is None:
any_match = (idx, codec)
return forced_match if forced_match else (any_match or (None, ""))
def _mux_subtitle(self, source: Path, output: Path):
"""Mux a subtitle track from source into the encoded output (soft sub).
Uses stream copy for MKV; converts to WebVTT for WebM containers."""
sub_idx, sub_codec = self._find_subtitle_stream(source, self.subtitle_lang)
if sub_idx is None:
self.log_msg.emit(f" SUBS: No {self.subtitle_lang} subtitle found in {source.name}")
return
# WebM only supports WebVTT natively; MKV carries any subtitle codec
is_webm = output.suffix.lower() == ".webm"
sub_codec_flag = "copy" if not is_webm else "webvtt"
tmp_out = output.with_suffix(output.suffix + ".submux_tmp")
try:
cmd = [
self.env.ffmpeg_path,
"-i", str(output), # encoded output (video + audio)
"-i", str(source), # original source (subtitle source)
"-map", "0", # all streams from encoded output
"-map", "-0:s", # strip any subtitle from output
"-map", f"1:{sub_idx}", # subtitle from source
"-c:v", "copy",
"-c:a", "copy",
"-c:s", sub_codec_flag,
"-y",
str(tmp_out),
]
res = subprocess.run(cmd, capture_output=True, text=True)
if res.returncode == 0 and tmp_out.exists() and tmp_out.stat().st_size > 0:
output.unlink()
tmp_out.rename(output)
self.log_msg.emit(
f" SUBS: Muxed {self.subtitle_lang} sub ({sub_codec}) into {output.name}"
)
else:
tmp_out.unlink(missing_ok=True)
tail = (res.stderr or "")[-200:]
self.log_msg.emit(f" SUBS WARN: Remux failed for {output.name}: {tail}")
except Exception as e:
tmp_out.unlink(missing_ok=True)
self.log_msg.emit(f" SUBS ERROR: {e}")
def stop(self):
self._stop = True
# ──────────────────────────────────────────────
# SOURCE BUILD WORKER — compile VS + av1an from git
# ──────────────────────────────────────────────
class SourceBuildWorker(QThread):
"""Builds VapourSynth and/or av1an from git to resolve ABI mismatches.
Runs in a background thread. Emits progress via log_msg.
When done, emits build_done(success, message).
Everything installs to the user's home directory (no sudo for install):
VapourSynth → ~/.local/lib/ (av1an finds it via LD_LIBRARY_PATH)
av1an → ~/.cargo/bin/ (already in PATH)
Only build-dependency installation (pacman -S) may need sudo.
"""
log_msg = Signal(str)
build_done = Signal(bool, str) # (success, detail)
def __init__(self, build_vs: bool = True, build_av1an: bool = True):
super().__init__()
self.build_vs = build_vs
self.build_av1an = build_av1an
self._stop = False
def _run_cmd(self, cmd, cwd=None, timeout=600, label=""):
"""Run a command, log output, return (returncode, combined_output)."""
self.log_msg.emit(f" $ {' '.join(cmd[:6])}{'...' if len(cmd)>6 else ''}")
try:
r = subprocess.run(cmd, capture_output=True, text=True,
timeout=timeout, cwd=cwd)
# Log last few lines of stderr for diagnostics
if r.stderr:
for line in r.stderr.strip().splitlines()[-5:]:
self.log_msg.emit(f" {line}")
if r.returncode != 0 and r.stdout:
for line in r.stdout.strip().splitlines()[-3:]:
self.log_msg.emit(f" {line}")
return r.returncode, (r.stdout or "") + (r.stderr or "")
except subprocess.TimeoutExpired:
self.log_msg.emit(f" TIMEOUT ({timeout}s) running: {label or cmd[0]}")
return -1, f"timeout after {timeout}s"
except Exception as e:
self.log_msg.emit(f" ERROR: {e}")
return -1, str(e)
def _sudo_cmd(self, cmd, timeout=120, label=""):
"""Run a command with sudo (or pkexec as graphical fallback)."""
# Try pkexec first (graphical polkit prompt — works in desktop sessions)
pkexec = shutil.which("pkexec")
if pkexec:
return self._run_cmd([pkexec] + cmd, timeout=timeout, label=label or cmd[0])
# Fall back to sudo (needs a terminal; may fail silently)
return self._run_cmd(["sudo"] + cmd, timeout=timeout, label=label or cmd[0])
def run(self):
try:
# ── Install build dependencies (may need one sudo prompt) ──
self.log_msg.emit("")
self.log_msg.emit("=== Installing build dependencies ===")
all_deps = [
"meson", "ninja", "gcc", "pkg-config", "git",
"nasm", "yasm", "cmake", "python",
]
need_rust = self.build_av1an and not shutil.which("cargo")
if need_rust:
all_deps.append("rust")
# Only invoke sudo if at least one dep is missing
missing = [d for d in all_deps if not shutil.which(d)]
if missing:
self.log_msg.emit(f" Missing: {', '.join(missing)} — installing via pacman")
rc, _ = self._sudo_cmd(
["pacman", "-S", "--needed", "--noconfirm"] + all_deps,
timeout=300, label="pacman build-deps",
)
if rc != 0:
self.log_msg.emit(" (some deps may already be installed — continuing)")
else:
self.log_msg.emit(" All build dependencies already installed.")
# Ensure cargo is in PATH after potential install
os.environ["PATH"] = os.environ.get("PATH", "") + ":/usr/bin:/root/.cargo/bin:" + str(Path.home() / ".cargo" / "bin")
if not shutil.which("cargo") and self.build_av1an:
self.log_msg.emit(" FATAL: cargo not found after deps install. Aborting.")
self.build_done.emit(False, "Rust/cargo not available")
return
# ── Build & install VapourSynth to ~/.local (NO sudo needed) ──
if self.build_vs:
self._build_vapoursynth()
# ── Build av1an to ~/.cargo/bin (NO sudo needed) ──
if self.build_av1an:
self._build_av1an()
# ── Ensure LD_LIBRARY_PATH includes local VS libs ──
local_lib = str(Path.home() / ".local" / "lib")
existing_ld = os.environ.get("LD_LIBRARY_PATH", "")
if local_lib not in existing_ld:
os.environ["LD_LIBRARY_PATH"] = f"{local_lib}:{existing_ld}".rstrip(":")
self.log_msg.emit(f" Set LD_LIBRARY_PATH to include {local_lib}")
self.log_msg.emit("")
self.log_msg.emit("=== Source build complete ===")
self.build_done.emit(True, "Build and install completed (local ~/.local/).")
except Exception as e:
self.log_msg.emit(f"BUILD FAILED: {e}")
self.build_done.emit(False, str(e))
def _build_vapoursynth(self):
"""Clone, build, and install VapourSynth to ~/.local/ (no sudo needed)."""
self.log_msg.emit("")
self.log_msg.emit("=== Building VapourSynth from git ===")
self.log_msg.emit(" Install target: ~/.local/ (no system-wide changes)")
build_dir = Path("/tmp/vapoursynth-git-build")
local_prefix = str(Path.home() / ".local")
if build_dir.exists():
self.log_msg.emit(f" Cleaning old build directory...")
import shutil as _shutil
_shutil.rmtree(build_dir, ignore_errors=True)
# Clone (shallow — faster)
rc, out = self._run_cmd(
["git", "clone", "--depth", "1",
"https://github.com/vapoursynth/vapoursynth.git",
str(build_dir)],
timeout=120, label="git clone vapoursynth",
)
if rc != 0:
raise Exception(f"git clone VapourSynth failed: {out[-300:]}")
# Meson setup — install to ~/.local so it doesn't touch system dirs
self.log_msg.emit(" Configuring with meson (--prefix=~/.local)...")
rc, out = self._run_cmd(
["meson", "setup", "build",
f"--prefix={local_prefix}", "--libdir=lib"],
cwd=str(build_dir), timeout=120, label="meson setup",
)
if rc != 0:
raise Exception(f"meson setup failed: {out[-500:]}")
# Build
self.log_msg.emit(" Compiling VapourSynth (this may take a few minutes)...")
rc, out = self._run_cmd(
["ninja", "-C", "build", "-j", str(max(1, os.cpu_count() or 2))],
cwd=str(build_dir), timeout=900, label="ninja build",
)
if rc != 0:
raise Exception(f"ninja build failed: {out[-500:]}")
# Install to ~/.local/ — NO sudo needed (user owns this directory)
self.log_msg.emit(" Installing VapourSynth to ~/.local/ ...")
rc, out = self._run_cmd(
["ninja", "-C", "build", "install"],
cwd=str(build_dir), timeout=120, label="ninja install",
)
if rc != 0:
raise Exception(f"ninja install failed: {out[-500:]}")
self.log_msg.emit(f" VapourSynth installed to {local_prefix}/ (libs in {local_prefix}/lib/)")
# Cleanup build directory
import shutil as _shutil
_shutil.rmtree(build_dir, ignore_errors=True)
def _build_av1an(self):
"""Clone and build av1an from git. Installs to ~/.cargo/bin/ (no sudo needed)."""
self.log_msg.emit("")
self.log_msg.emit("=== Building av1an from git ===")
self.log_msg.emit(" Install target: ~/.cargo/bin/ (no system-wide changes)")
# Ensure cargo is in PATH
cargo_bin = shutil.which("cargo")
if not cargo_bin:
# Common locations
for p in [Path.home() / ".cargo" / "bin" / "cargo", "/usr/bin/cargo"]:
if p.exists():
os.environ["PATH"] = os.environ.get("PATH", "") + f":{p.parent}"
cargo_bin = str(p)
break
if not cargo_bin:
raise Exception("cargo not found — cannot build av1an")
self.log_msg.emit(f" Using cargo at: {cargo_bin}")
self.log_msg.emit(" Compiling av1an (this may take 10-30 minutes)...")
rc, out = self._run_cmd(
["cargo", "install", "av1an",
"--git", "https://github.com/master-of-zen/av1an",
"--force", "--root", str(Path.home() / ".cargo")],
timeout=3600, label="cargo install av1an",
)
if rc != 0:
raise Exception(f"cargo install av1an failed: {out[-500:]}")
new_av1an = Path.home() / ".cargo" / "bin" / "av1an"
if new_av1an.exists():
self.log_msg.emit(f" av1an installed: {new_av1an}")
else:
self.log_msg.emit(" WARNING: av1an binary not found at expected path after build.")
def stop(self):
self._stop = True
# ──────────────────────────────────────────────
# MAIN WINDOW (merged UI from all 3)
# ──────────────────────────────────────────────
# ──────────────────────────────────────────────
# RETRO-FUTURISTIC MEDIA CONSOLE THEME
# ──────────────────────────────────────────────
# Brushed aluminum, amber/green LED displays,
# beveled metallic panels, VU meters, spectrum bars.
# Modernized with: rounded corners, subtle glow, glassmorphism hints,
# information-dense DAW-style layout.
MMD3_QSS = """
/* ── Global ── */
QMainWindow, QWidget#central {
background-color: #1a1a1e;
}
/* ── Group Boxes — brushed aluminum panels ── */
QGroupBox {
font-family: 'Segoe UI', 'Ubuntu', sans-serif;
font-size: 10px;
font-weight: bold;
color: #8a8a8a;
border: 1px solid #3a3a40;
border-radius: 8px;
margin-top: 14px;
padding: 14px 10px 10px 10px;
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #2c2c32, stop:0.5 #27272c, stop:1 #222228);
}
QGroupBox::title {
subcontrol-origin: margin;
subcontrol-position: top left;
padding: 2px 10px;
color: #666;
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #2c2c32, stop:1 #222228);
border-radius: 4px;
}
/* ── Labels ── */
QLabel {
color: #999;
font-size: 10px;
font-family: 'Segoe UI', 'Ubuntu', sans-serif;
}
/* ── Line Edits — recessed aluminum wells ── */
QLineEdit {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #18181c, stop:1 #141418);
border: 1px solid #333;
border-radius: 4px;
padding: 5px 8px;
color: #d4aa50; /* amber LED */
font-family: 'Consolas', 'DejaVu Sans Mono', 'Ubuntu Mono', monospace;
font-size: 11px;
selection-background-color: #d4aa50;
selection-color: #000;
}
QLineEdit:focus {
border-color: #d4aa50;
}
/* ── Combo Boxes ── */
QComboBox {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #1e1e24, stop:1 #1a1a20);
border: 1px solid #3a3a40;
border-radius: 4px;
padding: 4px 8px;
color: #c8c8c8;
font-family: 'Segoe UI', 'Ubuntu', sans-serif;
font-size: 11px;
min-height: 24px;
}
QComboBox:hover {
border-color: #555;
}
QComboBox:focus {
border-color: #d4aa50;
}
QComboBox::drop-down {
border: none;
width: 22px;
}
QComboBox::down-arrow {
image: none;
border-left: 4px solid transparent;
border-right: 4px solid transparent;
border-top: 6px solid #888;
margin-right: 6px;
}
QComboBox QAbstractItemView {
background: #1e1e24;
border: 1px solid #3a3a40;
border-radius: 4px;
color: #c8c8c8;
selection-background-color: #3a3a48;
selection-color: #d4aa50;
padding: 4px;
}
QComboBox item {
min-height: 22px;
padding: 2px 8px;
}
/* ── Buttons — beveled metallic (MMD3 transport style) ── */
QPushButton {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #404048, stop:0.15 #38383f,
stop:0.85 #2e2e35, stop:1 #28282e);
border: 1px solid #4a4a52;
border-bottom-color: #1a1a1e;
border-radius: 5px;
padding: 6px 16px;
color: #d0d0d0;
font-family: 'Segoe UI', 'Ubuntu', sans-serif;
font-size: 11px;
font-weight: bold;
}
QPushButton:hover {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #4a4a54, stop:0.15 #424248,
stop:0.85 #363640, stop:1 #303038);
border-color: #5a5a64;
color: #fff;
}
QPushButton:pressed {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #28282e, stop:1 #3a3a42);
border-bottom-color: #4a4a52;
border-top-color: #1a1a1e;
}
QPushButton:disabled {
background: #222228;
border-color: #2a2a30;
color: #555;
}
/* Primary action button — amber glow */
QPushButton#btnRun {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #3a3428, stop:0.15 #332e22,
stop:0.85 #2a261c, stop:1 #221e16);
border: 1px solid #5a4a30;
border-bottom-color: #1a1608;
color: #d4aa50;
font-size: 13px;
letter-spacing: 2px;
}
QPushButton#btnRun:hover {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #4a4030, stop:0.15 #423828,
stop:0.85 #3a3020, stop:1 #322a1a);
border-color: #d4aa50;
color: #f0d080;
}
QPushButton#btnRun:disabled {
background: #22201a;
border-color: #2a2820;
color: #5a4a30;
}
/* Stop button — red danger */
QPushButton#btnStop {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #3a2222, stop:0.15 #321c1c,
stop:0.85 #2a1616, stop:1 #221010);
border: 1px solid #5a3030;
border-bottom-color: #1a0808;
color: #e05050;
font-size: 13px;
letter-spacing: 2px;
}
QPushButton#btnStop:hover {
border-color: #e05050;
color: #ff7070;
}
QPushButton#btnStop:disabled {
background: #221a1a;
border-color: #2a2020;
color: #5a3030;
}
/* Rebuild-from-git button — muted teal */
QPushButton#btnRebuild {
background: qlineargradient(x1:0, y1:0, x2:0, y2:1,
stop:0 #1e2e2e, stop:0.15 #1a2a2a,
stop:0.85 #162424, stop:1 #121e1e);
border: 1px solid #2a5050;
border-bottom-color: #0e1818;
color: #50b0b0;
font-size: 10px;
letter-spacing: 1px;
}
QPushButton#btnRebuild:hover {
border-color: #50b0b0;
color: #70d0d0;
}
QPushButton#btnRebuild:disabled {
background: #1a1e1e;
border-color: #222828;
color: #304040;
}
/* Browse buttons — small, subdued */
QPushButton#btnBrowse {
font-size: 9px;
padding: 4px 10px;
letter-spacing: 1px;
}
/* ── Check Boxes ── */
QCheckBox {
color: #999;
font-size: 10px;
spacing: 8px;
font-family: 'Segoe UI', 'Ubuntu', sans-serif;
}
QCheckBox::indicator {
width: 16px;
height: 16px;
border-radius: 3px;
border: 1px solid #444;
background: #1a1a1e;
}
QCheckBox::indicator:checked {
background: #d4aa50;
border-color: #b8903a;
}
QCheckBox#dangerCheck {
color: #c05050;
font-weight: bold;
}
QCheckBox#dangerCheck::indicator:checked {
background: #c04040;
border-color: #a03030;
}
/* ── Text Edit (log) — LED terminal display ── */
QTextEdit#logBox {
background: #0a0a0c;
border: 2px solid #1e1e24;
border-radius: 6px;
color: #40d060; /* green phosphor LED */
font-family: 'Consolas', 'DejaVu Sans Mono', 'Ubuntu Mono', monospace;
font-size: 11px;
padding: 8px;
}
/* ── Status Bar — LED readout strip ── */
QStatusBar {
background: #0e0e12;
border-top: 1px solid #2a2a30;
font-family: 'Consolas', 'DejaVu Sans Mono', 'Ubuntu Mono', monospace;
font-size: 10px;
color: #d4aa50;
padding: 2px 8px;
}
QStatusBar QLabel {
color: #d4aa50;
font-family: 'Consolas', 'DejaVu Sans Mono', 'Ubuntu Mono', monospace;
font-size: 10px;
}
/* ── Tooltips ── */
QToolTip {
background: #2a2a30;
color: #c8c8c8;
border: 1px solid #444;
border-radius: 4px;
padding: 6px;
font-size: 10px;
}
/* ── Scrollbars — thin, dark ── */
QScrollBar:vertical {
background: #141418;
width: 10px;
border-radius: 5px;
margin: 0;
}
QScrollBar::handle:vertical {
background: #3a3a42;
border-radius: 5px;
min-height: 30px;
}
QScrollBar::handle:vertical:hover {
background: #4a4a54;
}
QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical {
height: 0;
}
QScrollBar:horizontal {
background: #141418;
height: 10px;
border-radius: 5px;
}
QScrollBar::handle:horizontal {
background: #3a3a42;
border-radius: 5px;
min-width: 30px;
}
QScrollBar::handle:horizontal:hover {
background: #4a4a54;
}
QScrollBar::add-line:horizontal, QScrollBar::sub-line:horizontal {
width: 0;
}
"""
class OpenCodecMaster(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("OpenTranscode — dcos.net")
self.resize(1100, 920)
self.worker: Optional[EncoderWorker] = None
self.env: Optional[EnvProbe] = None
self._pending_deletes: list[Path] = []
self._apply_mmd3_theme()
self._build_ui()
# Probe environment after UI is up
QTimer = __import__("PySide6.QtCore", fromlist=["QTimer"]).QTimer
QTimer.singleShot(500, self._probe_and_init)
# ── UI Construction ──
def _build_ui(self):
central = QWidget()
central.setObjectName("central")
self.setCentralWidget(central)
root = QVBoxLayout(central)
root.setContentsMargins(10, 6, 10, 4)
root.setSpacing(4)
# ── Header ──
header = QWidget()
header_lay = QVBoxLayout(header)
header_lay.setContentsMargins(0, 0, 0, 0)
header_lay.setSpacing(0)
title = QLabel("OpenTranscode")
title.setFont(QFont("Segoe UI", 22, QFont.Weight.Bold))
title.setAlignment(Qt.AlignmentFlag.AlignCenter)
title.setStyleSheet("color: #d4aa50; letter-spacing: 4px;")
header_lay.addWidget(title)
subtitle = QLabel('dcos.net // concurrent open-source transcoding')
subtitle.setFont(QFont("Consolas", 8))
subtitle.setAlignment(Qt.AlignmentFlag.AlignCenter)
subtitle.setStyleSheet("color: #555; letter-spacing: 2px;")
header_lay.addWidget(subtitle)
accent = QWidget()
accent.setFixedHeight(1)
accent.setStyleSheet("background: qlineargradient(x1:0, y1:0, x2:1, y2:0,"
"stop:0 transparent, stop:0.15 #d4aa5044,"
"stop:0.5 #d4aa5088, stop:0.85 #d4aa5044, stop:1 transparent);")
header_lay.addWidget(accent)
root.addWidget(header)
# ── Paths ──
path_grp = QGroupBox("Paths")
path_lay = QVBoxLayout(path_grp)
path_lay.setSpacing(2)
path_lay.setContentsMargins(10, 14, 10, 8)
self.in_path_edit = QLineEdit(str(Path.home() / "Videos" / "INCOMING"))
self.out_path_edit = QLineEdit(str(Path.home() / "Videos" / "ARCHIVE"))
for label_text, line_edit in [
("IN:", self.in_path_edit),
("OUT:", self.out_path_edit),
]:
row = QHBoxLayout()
row.setSpacing(6)
lbl = QLabel(label_text)
lbl.setFixedWidth(28)
lbl.setStyleSheet("color: #d4aa50; font-family: 'Consolas', monospace; font-weight: bold; font-size: 10px;")
row.addWidget(lbl)
row.addWidget(line_edit, 1)
btn_browse = QPushButton("...")
btn_browse.setObjectName("btnBrowse")
btn_browse.setFixedSize(30, 22)
btn_browse.setToolTip("Browse")
btn_browse.clicked.connect(
lambda checked, le=line_edit, is_dir=True: self._browse(le, is_dir)
)
row.addWidget(btn_browse)
path_lay.addLayout(row)
root.addWidget(path_grp)
# ── Encoder Chain ──
codec_grp = QGroupBox("Encoder Chain")
codec_lay = QHBoxLayout(codec_grp)
codec_lay.setSpacing(8)
codec_lay.setContentsMargins(10, 14, 10, 8)
for col_idx, (label, combo_items, slot) in enumerate([
("VIDEO", [vc.label for vc in VIDEO_CODECS], self._on_codec_changed),
("PRESET", [], None),
("AUDIO", [ap.label for ap in AUDIO_PROFILES], None),
("CONTAINER", [cp.label for cp in CONTAINER_PROFILES], self._on_container_changed),
("RESOLUTION", [], self._on_resolution_changed),
("SUBS", [so[0] for so in SUBTITLE_OPTIONS], None),
]):
col = QVBoxLayout()
col.setSpacing(1)
lbl = QLabel(label)
lbl.setStyleSheet("color: #666; font-size: 7px; letter-spacing: 1px;")
col.addWidget(lbl)
combo = QComboBox()
combo.setFixedHeight(24)
if combo_items:
combo.addItems(combo_items)
if slot:
combo.currentIndexChanged.connect(slot)
col.addWidget(combo)
codec_lay.addLayout(col)
if label == "VIDEO":
self.codec_combo = combo
elif label == "PRESET":
self.preset_combo = combo
self._populate_presets(0)
self.preset_combo.setCurrentIndex(1)
elif label == "AUDIO":
self.audio_combo = combo
elif label == "CONTAINER":
self.container_combo = combo
elif label == "RESOLUTION":
self.resolution_combo = combo
self._populate_resolution_combo()
elif label == "SUBS":
self.subs_combo = combo
root.addWidget(codec_grp)
# ── Side panel: compact knobs ──
knobs_panel = QWidget()
knobs_panel.setFixedWidth(170)
knobs_lay = QVBoxLayout(knobs_panel)
knobs_lay.setContentsMargins(6, 8, 6, 8)
knobs_lay.setSpacing(6)
knobs_lay.setAlignment(Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignHCenter)
# CRF Knob — amber
self.crf_knob = RadioKnob(
min_val=18, max_val=52, default_val=32,
label="Quality",
unit="CRF",
color=(212, 170, 80),
num_ticks=18,
tick_labels=["18", "28", "38", "52"],
snap_ticks=True,
compact=True,
)
self.crf_knob.valueChanged.connect(self._on_crf_knob_changed)
knobs_lay.addWidget(self.crf_knob, 0, Qt.AlignmentFlag.AlignHCenter)
# Volume Knob — green (dual-pass loudnorm target)
self.vol_knob = RadioKnob(
min_val=-20.0, max_val=6.0, default_val=0.0,
label="LUFS",
unit="dB",
color=(64, 208, 96),
num_ticks=27,
tick_labels=["-20", "-10", "0", "+6"],
snap_ticks=True,
compact=True,
)
self.vol_knob.setToolTip(
"Dual-pass audio normalization target (EBU R128 LUFS).\n"
"0 = off (pass-through).\n"
"Each file is analyzed individually: loudnorm measures its\n"
"current LUFS and true peak, then computes the exact gain\n"
"to hit this target. If the gain would push peaks above\n"
"-1.5 dBTP, gain is reduced to keep 15%% headroom.\n"
"Common targets: -14 (streaming), -16 (broadcast), -23 (cinema)."
)
self.vol_knob.valueChanged.connect(self._on_vol_knob_changed)
knobs_lay.addWidget(self.vol_knob, 0, Qt.AlignmentFlag.AlignHCenter)
# ── Options row ──
opt_row = QHBoxLayout()
opt_row.setSpacing(8)
opt_lbl = QLabel("FILTER")
opt_lbl.setFixedWidth(44)
opt_lbl.setStyleSheet("color: #666; font-size: 7px; letter-spacing: 1px;")
opt_row.addWidget(opt_lbl)
self.ext_edit = QLineEdit(", ".join(sorted(DEFAULT_INPUT_EXTENSIONS)))
self.ext_edit.setFixedHeight(22)
self.ext_edit.setToolTip("File extensions to process. Separate with commas.")
opt_row.addWidget(self.ext_edit)
self.del_check = QCheckBox("Delete source after verify")
self.del_check.setObjectName("dangerCheck")
self.del_check.setToolTip(
"Sources are only deleted after all files finish and cleanup passes.\n"
"If any file fails, you will be prompted before deletion."
)
opt_row.addWidget(self.del_check)
root.addLayout(opt_row)
# ── Log + Knobs: horizontal split ──
mid_split = QHBoxLayout()
mid_split.setSpacing(6)
# Log: LED terminal (takes remaining space)
self.log_box = QTextEdit()
self.log_box.setObjectName("logBox")
self.log_box.setReadOnly(True)
mid_split.addWidget(self.log_box, 1)
# Knobs panel on the right
mid_split.addWidget(knobs_panel)
root.addLayout(mid_split, 1)
# ── Status Bar: LED readout ──
self.status = QStatusBar()
self.setStatusBar(self.status)
self.status_label = QLabel(" INITIALIZING...")
self.status_label.setStyleSheet(
"color: #d4aa50; font-family: 'Consolas', 'DejaVu Sans Mono', monospace; font-size: 10px;"
)
self.status.addWidget(self.status_label, 1)
# ── Transport Buttons ──
btn_lay = QHBoxLayout()
btn_lay.setSpacing(8)
self.btn_run = QPushButton(" > ENCODE")
self.btn_run.setObjectName("btnRun")
self.btn_run.setFixedHeight(40)
self.btn_run.setEnabled(False)
self.btn_run.clicked.connect(self._start_process)
btn_lay.addWidget(self.btn_run)
self.btn_stop = QPushButton(" [] STOP")
self.btn_stop.setObjectName("btnStop")
self.btn_stop.setFixedHeight(40)
self.btn_stop.clicked.connect(self._stop_process)
self.btn_stop.setEnabled(False)
btn_lay.addWidget(self.btn_stop)
self.btn_rebuild = QPushButton(" <> REBUILD FROM GIT")
self.btn_rebuild.setObjectName("btnRebuild")
self.btn_rebuild.setFixedHeight(40)
self.btn_rebuild.setToolTip(
"Compile VapourSynth + av1an from git source.\n"
"Resolves ABI/version mismatch when package managers\n"
"install incompatible versions."
)
self.btn_rebuild.clicked.connect(self._manual_rebuild)
self.btn_rebuild.setEnabled(False)
btn_lay.addWidget(self.btn_rebuild)
root.addLayout(btn_lay)
# ── Footer ──
footer = QWidget()
footer_lay = QHBoxLayout(footer)
footer_lay.setContentsMargins(6, 4, 6, 2)
footer_lay.setSpacing(0)
link_lbl = QLabel(
'<a href="http://git.dcos.net/dcosnet/OpenTranscode" '
'style="color: #888; text-decoration: none;">Visit Homepage</a>'
)
link_lbl.setTextInteractionFlags(Qt.TextInteractionFlag.TextBrowserInteraction)
link_lbl.setOpenExternalLinks(True)
link_lbl.setStyleSheet("font-size: 8px;")
footer_lay.addWidget(link_lbl)
footer_lay.addStretch()
copy_lbl = QLabel(
'AGPL-3.0 | Jeremy Anderson - <a href="http://dcos.net" '
'style="color: #888; text-decoration: none;">dcos.net</a> (c) 2026'
)
copy_lbl.setTextInteractionFlags(Qt.TextInteractionFlag.TextBrowserInteraction)
copy_lbl.setOpenExternalLinks(True)
copy_lbl.setAlignment(Qt.AlignmentFlag.AlignRight)
copy_lbl.setStyleSheet("color: #555; font-size: 8px;")
footer_lay.addWidget(copy_lbl)
root.addWidget(footer)
def _apply_mmd3_theme(self):
self.setStyle(QStyleFactory.create("Fusion"))
self.setStyleSheet(MMD3_QSS)
# Palette as fallback for things QSS doesn't cover
p = QPalette()
p.setColor(QPalette.ColorRole.Window, QColor(26, 26, 30))
p.setColor(QPalette.ColorRole.WindowText, QColor(200, 200, 200))
p.setColor(QPalette.ColorRole.Base, QColor(20, 20, 24))
p.setColor(QPalette.ColorRole.AlternateBase, QColor(40, 40, 46))
p.setColor(QPalette.ColorRole.ToolTipBase, QColor(30, 30, 36))
p.setColor(QPalette.ColorRole.ToolTipText, QColor(200, 200, 200))
p.setColor(QPalette.ColorRole.Text, QColor(200, 200, 200))
p.setColor(QPalette.ColorRole.Button, QColor(40, 40, 46))
p.setColor(QPalette.ColorRole.ButtonText, QColor(200, 200, 200))
p.setColor(QPalette.ColorRole.Highlight, QColor(212, 170, 80))
p.setColor(QPalette.ColorRole.HighlightedText, QColor(0, 0, 0))
QApplication.instance().setPalette(p)
# ── Slots ──
@Slot()
def _on_codec_changed(self, idx: int):
self._populate_presets(idx)
profile = VIDEO_CODECS[idx]
lo, hi = profile.crf_range
self.crf_knob.min_val = lo
self.crf_knob.max_val = hi
self.crf_knob.setValue(float(profile.default_crf))
# Auto-select best container
for i, cp in enumerate(CONTAINER_PROFILES):
if cp.ext == profile.container:
self.container_combo.blockSignals(True)
self.container_combo.setCurrentIndex(i)
self.container_combo.blockSignals(False)
break
def _populate_presets(self, codec_idx: int):
self.preset_combo.blockSignals(True)
self.preset_combo.clear()
if 0 <= codec_idx < len(VIDEO_CODECS):
self.preset_combo.addItems(VIDEO_CODECS[codec_idx].presets)
self.preset_combo.blockSignals(False)
@Slot()
def _on_container_changed(self, idx: int):
if idx >= 0:
ext = CONTAINER_PROFILES[idx].ext
self.log(f"Container set to: {ext}")
def _populate_resolution_combo(self):
"""Populate resolution dropdown with separator headers per category."""
# Maps combo box position -> RESOLUTION_PRESETS index.
# Separators occupy combo positions too, so we must track them.
self._res_preset_indices: dict[int, int] = {} # combo_pos -> preset index
last_cat = None
combo_pos = 0
for i, rp in enumerate(RESOLUTION_PRESETS):
if rp.category != last_cat:
if last_cat is not None:
self.resolution_combo.insertSeparator(combo_pos)
combo_pos += 1 # separator takes a slot
last_cat = rp.category
self.resolution_combo.addItem(rp.label)
self._res_preset_indices[combo_pos] = i
combo_pos += 1
def _get_current_resolution(self) -> ResolutionProfile:
"""Get the ResolutionProfile for the current combo selection, handling separators."""
combo_idx = self.resolution_combo.currentIndex()
preset_i = self._res_preset_indices.get(combo_idx)
if preset_i is not None:
return RESOLUTION_PRESETS[preset_i]
return RESOLUTION_PRESETS[0]
@Slot()
def _on_resolution_changed(self, idx: int):
rp = self._get_current_resolution()
if rp.width is not None:
self._log(
f"Resolution: {rp.width}x{rp.height} ({rp.aspect_label}) — "
f"files will be pre-scaled with ffmpeg before encoding."
)
else:
self._log("Resolution: Original (no scaling).")
@Slot(float)
def _on_crf_knob_changed(self, val: float):
direction = "higher quality" if val < 28 else ("balanced" if val < 38 else "smaller file")
self._log(f"CRF: {val:.0f} ({direction})")
@Slot(float)
def _on_vol_knob_changed(self, val: float):
if abs(val) < 0.01:
self._log("Audio normalization: OFF (pass-through)")
else:
direction = "louder" if val > 0 else "quieter"
self._log(f"Audio normalization: {val:+.1f} dB ({direction})")
@Slot()
def _browse(self, line_edit: QLineEdit, is_dir: bool = True):
if is_dir:
path = QFileDialog.getExistingDirectory(self, "Select Directory")
if path:
line_edit.setText(path)
def _log(self, msg: str):
self.log_box.append(f"> {msg}")
sb = self.log_box.verticalScrollBar()
sb.setValue(sb.maximum())
# ── Environment Probe ──
def _probe_and_init(self):
self.env = probe_environment()
# --- Distro banner ---
distro = self.env.distro
self._log(f"Distro: {distro.name} (family={distro.family}, v{distro.version_id})")
self._log(f"Package manager: {distro.pkg_manager}")
# --- Warnings (info-level, not errors) ---
for w in self.env.warnings:
self._log(f"INFO: {w}")
# --- Hard errors ---
if not self.env.av1an_path:
self._log("CRITICAL: 'av1an' not found in PATH or distro-specific paths.")
if self.env.install_hint:
self._log(f" TRY: {self.env.install_hint}")
self.status_label.setText(f"NOT READY — missing av1an ({distro.family})")
return
if not self.env.ffmpeg_path:
self._log("CRITICAL: 'ffmpeg' not found in PATH or distro-specific paths.")
if self.env.install_hint:
self._log(f" TRY: {self.env.install_hint}")
self.status_label.setText(f"NOT READY — missing ffmpeg ({distro.family})")
return
if self.env.errors:
for e in self.env.errors:
self._log(f"ERROR: {e}")
# If there are still errors after logging (e.g. missing runtime deps), block start
if self.env.errors:
dep_count = len(self.env.missing_dep_pkgs)
if dep_count:
self.status_label.setText(
f"NOT READY — {dep_count} runtime dep(s) missing. See log."
)
return
# --- Probe results ---
flag_info = ", ".join(f"{k}={v}" for k, v in self.env.av1an_flags.items() if k != "has_chunk_method" and k != "has_scenes")
self._log(f"av1an: {self.env.av1an_path} (v{self.env.av1an_version or '?'})")
if flag_info:
self._log(f" Flags: {flag_info}")
if self.env.ffmpeg_version:
self._log(f"ffmpeg: {self.env.ffmpeg_path} (v{self.env.ffmpeg_version})")
# --- FFmpeg encoder library summary (audio-relevant only for our purposes) ---
available_libs = [name for name, present in self.env.ffmpeg_libs.items() if present]
missing_audio = [name for name, present in self.env.ffmpeg_libs.items()
if not present and name in ("libopus", "libvorbis", "flac")]
if available_libs:
self._log(f" FFmpeg encoders available: {', '.join(available_libs)}")
if missing_audio:
self._log(f" FFmpeg audio encoders MISSING: {', '.join(missing_audio)}")
self._log(f" Some audio codec options may fail. Check distro package: {distro.ffmpeg_pkg}")
# --- Disable unavailable codec options in UI ---
self._disable_unavailable_codecs()
worker_count = max(1, self.env.cpu.physical_cores - 1)
cpu = self.env.cpu
self._log(
f"Chunk-parallel mode: {worker_count} av1an workers "
f"({cpu.physical_cores} physical cores, {cpu.logical_threads} logical, "
f"{cpu.threads_per_core}T/core)"
)
self.btn_run.setEnabled(True)
self.btn_run.setText("START PROCESSING")
self.btn_rebuild.setEnabled(True) # available after successful probe
vs_info = f" | VS{self.env.vs_version}" if self.env.vs_version else ""
# Show ffmpeg video encoder availability (for fallback)
fb_encs = []
for vc in VIDEO_CODECS:
lib_key = {"libsvtav1": "libsvtav1", "libvpx-vp9": "libvpx",
"libx265": "libx265"}.get(vc.ffmpeg_encoder, vc.ffmpeg_encoder)
if self.env.ffmpeg_libs.get(lib_key, False):
fb_encs.append(vc.ffmpeg_encoder)
fb_info = f" | ffmpeg-fb:{'+'.join(fb_encs)}" if fb_encs else ""
self.status_label.setText(
f"{distro.name} | {cpu.physical_cores}C/{cpu.logical_threads}T | "
f"av1an v{self.env.av1an_version or '?'} | ffmpeg v{self.env.ffmpeg_version or '?'}{vs_info}{fb_info}"
)
def _disable_unavailable_codecs(self):
"""Grey out AUDIO codec combos whose FFmpeg library is missing.
Video codecs are NOT disabled here because av1an uses its own
encoder binaries (svt_av1, vpx, x265) — it does not rely on
ffmpeg's encoder list for video.
"""
libs = self.env.ffmpeg_libs
# Audio codecs — these ARE handled by ffmpeg, so probe is valid
audio_lib_map = {
0: "libopus", # Opus (96k)
1: "libopus", # Opus (128k)
2: "libopus", # Opus (64k)
3: "libvorbis", # Vorbis (128k)
4: "libvorbis", # Vorbis (192k)
5: "flac", # FLAC (lossless)
}
for idx, lib_name in audio_lib_map.items():
if idx < self.audio_combo.count():
if not libs.get(lib_name, False):
self.audio_combo.model().item(idx).setEnabled(False)
self.audio_combo.model().item(idx).setToolTip(
f"DISABLED: FFmpeg missing {lib_name} encoder."
)
if self.audio_combo.currentIndex() == idx:
self.audio_combo.setCurrentIndex(0) # Fallback to Opus (96k)
# ── Process Control ──
def _parse_extensions(self) -> set[str]:
raw = self.ext_edit.text()
exts = set()
for part in raw.split(","):
part = part.strip().lower()
if not part.startswith("."):
part = "." + part
if part:
exts.add(part)
return exts or DEFAULT_INPUT_EXTENSIONS
@Slot()
def _start_process(self):
in_dir = Path(self.in_path_edit.text())
out_dir = Path(self.out_path_edit.text())
if not in_dir.is_dir():
self._log(f"ERROR: Source directory does not exist: {in_dir}")
return
if in_dir == out_dir:
self._log("ERROR: Source and output directories must be different.")
return
# If delete is enabled, collect files first for batch confirmation
if self.del_check.isChecked():
extensions = self._parse_extensions()
candidates = [f for f in in_dir.rglob("*") if f.is_file() and f.suffix.lower() in extensions and not f.name.endswith(".scaled_tmp.mkv")]
if candidates:
total_size = sum(f.stat().st_size for f in candidates)
reply = QMessageBox.question(
self, "Confirm Batch Delete",
f"This will delete {len(candidates)} source file(s) after successful transcode.\n"
f"Total size: {total_size / 1_073_741_824:.2f} GB\n\n"
f"Proceed?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No,
)
if reply != QMessageBox.StandardButton.Yes:
self._log("Cancelled: Delete not confirmed.")
return
# ── Pre-flight: av1an VSScript smoke test (main thread — can show dialogs) ──
use_ffmpeg_fallback = False
skip_encode = False
if self.env.av1an_path and self.env.ffmpeg_path:
self._log("Pre-flight: testing av1an + VapourSynth compatibility...")
QApplication.processEvents() # keep UI responsive
svt_name = self.env.av1an_flags.get("svt_name", "svt_av1")
ok, detail = _av1an_vsscript_smoke_test(
self.env.av1an_path,
self.env.ffmpeg_path,
self.env.av1an_flags,
svt_name=svt_name,
)
if not ok and "VSScript_API_INCOMPAT" in detail:
# VSScript ABI mismatch detected — offer rebuild or fallback
use_ffmpeg_fallback = self._handle_vs_incompat()
if not use_ffmpeg_fallback:
# User chose rebuild or cancel — don't start encoding
return
elif not ok and "INVALID_ENCODER" in detail:
# Encoder name detection was wrong — re-probe and retry once
self._log(f" WARN: Encoder name probe was incorrect. Re-detecting...")
QApplication.processEvents()
new_name = _detect_av1an_svt_encoder(self.env.av1an_path)
if new_name:
self.env.av1an_flags["svt_name"] = new_name
self._log(f" Re-detected SVT-AV1 encoder name: '{new_name}'")
# Retry smoke test with corrected name
ok2, detail2 = _av1an_vsscript_smoke_test(
self.env.av1an_path, self.env.ffmpeg_path,
self.env.av1an_flags, svt_name=new_name,
)
if ok2:
self._log(" OK: av1an + VapourSynth working correctly.")
else:
self._log(f" FAIL: Still failing after re-detect: {detail2}")
return
else:
self._log(" FAIL: Could not determine valid encoder name. Check av1an --help manually.")
return
elif ok:
self._log(" OK: av1an + VapourSynth working correctly.")
else:
# Non-VSScript error (e.g. missing encoder binary) — let per-file loop handle it
self._log(f" WARN: smoke test issue (non-fatal): {detail}")
codec_idx = self.codec_combo.currentIndex()
audio_idx = self.audio_combo.currentIndex()
container_idx = self.container_combo.currentIndex()
# Safety: clamp codec_idx to valid range
if not (0 <= codec_idx < len(VIDEO_CODECS)):
self._log(f"ERROR: Invalid codec index {codec_idx}. Resetting to AV1 (SVT-AV1).")
codec_idx = 0
self.codec_combo.blockSignals(True)
self.codec_combo.setCurrentIndex(0)
self.codec_combo.blockSignals(False)
selected_codec = VIDEO_CODECS[codec_idx]
self._log(f"Codec: {selected_codec.label} (av1an encoder: {selected_codec.av1an_encoder})")
self.worker = EncoderWorker(
in_dir=in_dir,
out_dir=out_dir,
video_codec=selected_codec,
audio_profile=AUDIO_PROFILES[audio_idx],
container=CONTAINER_PROFILES[container_idx],
crf=self.crf_knob.intValue(),
preset_label=self.preset_combo.currentText(),
delete_source=self.del_check.isChecked(),
env=self.env,
extensions=self._parse_extensions(),
resolution=self._get_current_resolution(),
audio_level_db=self.vol_knob.value(),
use_ffmpeg_fallback=use_ffmpeg_fallback,
subtitle_lang=SUBTITLE_OPTIONS[self.subs_combo.currentIndex()][1],
)
self.worker.log_msg.connect(self._log)
self.worker.progress_msg.connect(self._on_progress)
self.worker.finished_queue.connect(self._on_finished)
self.btn_run.setEnabled(False)
self.btn_run.setText("RUNNING...")
self.btn_stop.setEnabled(True)
self.btn_rebuild.setEnabled(False)
self.worker.start()
def _handle_vs_incompat(self) -> bool:
"""Handle detected VSScript ABI incompatibility.
Shows a dialog with options:
1. Rebuild VapourSynth + av1an from git (resolves root cause)
2. Use ffmpeg fallback (works now, no chunk-parallel)
3. Cancel
Returns True if we should use ffmpeg fallback (option 2),
False if user cancelled or chose to rebuild (rebuild
starts async and does NOT return here — the user
will click ENCODE again after it completes).
"""
self._log(" FAIL: av1an cannot initialize VSScript API.")
self._log(" The av1an binary was compiled against a different VapourSynth version.")
# Check ffmpeg fallback availability
codec_idx = self.codec_combo.currentIndex()
video_codec = VIDEO_CODECS[codec_idx]
ffmpeg_enc = video_codec.ffmpeg_encoder
ffmpeg_lib_key = {
"libsvtav1": "libsvtav1",
"libaom-av1": "libaom",
"libvpx-vp9": "libvpx",
"libx265": "libx265",
}.get(ffmpeg_enc, ffmpeg_enc)
fallback_possible = self.env.ffmpeg_libs.get(ffmpeg_lib_key, False)
if fallback_possible:
btn_rebuild = QPushButton(" Rebuild from Git ")
btn_rebuild.setObjectName("btnRebuild")
btn_fallback = QPushButton(" Use ffmpeg Fallback ")
btn_fallback.setObjectName("btnRun")
btn_cancel = QPushButton(" Cancel ")
btn_cancel.setObjectName("btnStop")
dlg = QMessageBox(self)
dlg.setWindowTitle("av1an + VapourSynth Version Mismatch")
dlg.setText(
"av1an cannot initialize VapourSynth — the installed versions\n"
"have an ABI incompatibility (common with distro packages).\n\n"
f"Choose how to proceed:"
)
dlg.setInformativeText(
"• Rebuild from Git — compiles both from source (~10-30 min).\n"
" Fixes the root cause. Requires sudo for install.\n"
f"• ffmpeg Fallback — encode with ffmpeg ({ffmpeg_enc}) now.\n"
" No chunk-parallel mode but output quality is identical."
)
dlg.addButton(btn_rebuild, QMessageBox.ButtonRole.AcceptRole)
dlg.addButton(btn_fallback, QMessageBox.ButtonRole.YesRole)
dlg.addButton(btn_cancel, QMessageBox.ButtonRole.RejectRole)
dlg.exec()
clicked = dlg.clickedButton()
if clicked == btn_rebuild:
self._log("")
self._log("User chose: Rebuild VapourSynth + av1an from git.")
self._start_git_rebuild()
return False # don't start encoding — user will retry after build
elif clicked == btn_fallback:
self._log("")
self._log(f"FALLBACK: Switching to pure ffmpeg ({ffmpeg_enc}) encoding.")
self._log(
" Note: ffmpeg single-pass mode (no chunk-parallel). "
"Slower for large files but produces identical output."
)
self._log(" Use the REBUILD FROM GIT button to fix av1an for chunk-parallel mode.")
self._log("")
return True
else:
# Cancel
self._log("Cancelled by user.")
return False
else:
# No ffmpeg fallback available — offer rebuild or hard cancel
btn_rebuild = QPushButton(" Rebuild from Git ")
btn_rebuild.setObjectName("btnRebuild")
btn_cancel = QPushButton(" Cancel ")
btn_cancel.setObjectName("btnStop")
dlg = QMessageBox(self)
dlg.setWindowTitle("av1an + VapourSynth Version Mismatch")
dlg.setText(
"av1an cannot initialize VapourSynth — ABI incompatibility.\n\n"
f"ffmpeg also lacks '{ffmpeg_enc}' — no fallback possible.\n"
"You must rebuild to proceed."
)
dlg.setIcon(QMessageBox.Icon.Critical)
dlg.addButton(btn_rebuild, QMessageBox.ButtonRole.AcceptRole)
dlg.addButton(btn_cancel, QMessageBox.ButtonRole.RejectRole)
dlg.exec()
clicked = dlg.clickedButton()
if clicked == btn_rebuild:
self._log("")
self._log("User chose: Rebuild VapourSynth + av1an from git (no fallback available).")
self._start_git_rebuild()
else:
self._log("Cancelled by user.")
return False
def _start_git_rebuild(self, build_vs: bool = True, build_av1an: bool = True):
"""Start the SourceBuildWorker thread."""
self._log("Starting source build (VapourSynth + av1an from git)...")
self._log("Builds to ~/.local/ and ~/.cargo/bin/ — sudo only if build deps are missing.")
self.btn_run.setEnabled(False)
self.btn_rebuild.setEnabled(False)
self.btn_stop.setEnabled(False)
self.status_label.setText("Building from git... (see log)")
self._build_worker = SourceBuildWorker(build_vs=build_vs, build_av1an=build_av1an)
self._build_worker.log_msg.connect(self._log)
self._build_worker.build_done.connect(self._on_build_done)
self._build_worker.start()
@Slot(bool, str)
def _on_build_done(self, success: bool, message: str):
"""Called when SourceBuildWorker finishes."""
self._log("")
if success:
self._log(f"BUILD SUCCESS: {message}")
self._log("Re-probing environment to pick up new binaries...")
QApplication.processEvents()
# Ensure LD_LIBRARY_PATH is set in the main process too
# (SourceBuildWorker sets it in its thread, but we need it here)
local_lib = str(Path.home() / ".local" / "lib")
existing_ld = os.environ.get("LD_LIBRARY_PATH", "")
if local_lib not in existing_ld:
os.environ["LD_LIBRARY_PATH"] = f"{local_lib}:{existing_ld}".rstrip(":")
# Re-probe environment with fresh data
self.env = probe_environment()
# Run smoke test again to verify the fix
if self.env.av1an_path and self.env.ffmpeg_path:
svt_name = self.env.av1an_flags.get("svt_name", "svt_av1")
ok, detail = _av1an_vsscript_smoke_test(
self.env.av1an_path,
self.env.ffmpeg_path,
self.env.av1an_flags,
svt_name=svt_name,
)
if ok:
self._log("VERIFIED: av1an + VapourSynth now working correctly!")
self._log("Click START PROCESSING to encode.")
elif "INVALID_ENCODER" in detail:
# Re-probe encoder name with the fresh binary
self._log(" Re-detecting encoder name from fresh build...")
new_name = _detect_av1an_svt_encoder(self.env.av1an_path)
if new_name and new_name != svt_name:
self.env.av1an_flags["svt_name"] = new_name
self._log(f" Corrected encoder name: '{svt_name}' -> '{new_name}'")
ok2, detail2 = _av1an_vsscript_smoke_test(
self.env.av1an_path, self.env.ffmpeg_path,
self.env.av1an_flags, svt_name=new_name,
)
if ok2:
self._log("VERIFIED: av1an + VapourSynth now working correctly!")
self._log("Click START PROCESSING to encode.")
else:
self._log(f"WARNING: Smoke test still fails: {detail2}")
else:
self._log(f"WARNING: Could not auto-fix encoder name. Smoke test: {detail}")
else:
self._log(f"WARNING: Build completed but smoke test still fails: {detail}")
self._log("You may need to log out/in or restart the app for library changes to take effect.")
# Update status bar
distro = self.env.distro
cpu = self.env.cpu
vs_info = f" | VS{self.env.vs_version}" if self.env.vs_version else ""
fb_encs = []
for vc in VIDEO_CODECS:
lib_key = {"libsvtav1": "libsvtav1", "libvpx-vp9": "libvpx",
"libx265": "libx265"}.get(vc.ffmpeg_encoder, vc.ffmpeg_encoder)
if self.env.ffmpeg_libs.get(lib_key, False):
fb_encs.append(vc.ffmpeg_encoder)
fb_info = f" | ffmpeg-fb:{'+'.join(fb_encs)}" if fb_encs else ""
self.status_label.setText(
f"{distro.name} | {cpu.physical_cores}C/{cpu.logical_threads}T | "
f"av1an v{self.env.av1an_version or '?'} | ffmpeg v{self.env.ffmpeg_version or '?'}{vs_info}{fb_info}"
)
else:
self._log(f"BUILD FAILED: {message}")
self._log("Try running the build manually in a terminal, or use ffmpeg fallback.")
self.status_label.setText("Build failed — check log")
self.btn_run.setEnabled(True)
self.btn_rebuild.setEnabled(True)
@Slot()
def _manual_rebuild(self):
"""Handle the REBUILD FROM GIT button click (manual trigger)."""
btn_vs_av1an = QPushButton(" VapourSynth + av1an ")
btn_vs_av1an.setObjectName("btnRebuild")
btn_vs_only = QPushButton(" VapourSynth only ")
btn_vs_only.setObjectName("btnRebuild")
btn_av1an_only = QPushButton(" av1an only ")
btn_av1an_only.setObjectName("btnRebuild")
btn_cancel = QPushButton(" Cancel ")
btn_cancel.setObjectName("btnStop")
dlg = QMessageBox(self)
dlg.setWindowTitle("Rebuild from Git")
dlg.setText(
"Select which components to rebuild from git source.\n\n"
"• VapourSynth — installs to /usr (needs sudo)\n"
"• av1an — builds via cargo, copies to /usr/bin (needs sudo)\n\n"
"Build time: VapourSynth ~2-5 min, av1an ~10-30 min"
)
dlg.addButton(btn_vs_av1an, QMessageBox.ButtonRole.AcceptRole)
dlg.addButton(btn_vs_only, QMessageBox.ButtonRole.YesRole)
dlg.addButton(btn_av1an_only, QMessageBox.ButtonRole.NoRole)
dlg.addButton(btn_cancel, QMessageBox.ButtonRole.RejectRole)
dlg.exec()
clicked = dlg.clickedButton()
if clicked == btn_vs_av1an:
self._start_git_rebuild(build_vs=True, build_av1an=True)
elif clicked == btn_vs_only:
self._start_git_rebuild(build_vs=True, build_av1an=False)
elif clicked == btn_av1an_only:
self._start_git_rebuild(build_vs=False, build_av1an=True)
@Slot(str, int, int)
def _on_progress(self, filename: str, current: int, total: int):
self.status_label.setText(f"Processing {current}/{total}: {filename}")
@Slot(int, int)
def _on_finished(self, ok: int, fail: int):
self.btn_run.setEnabled(True)
self.btn_run.setText("START PROCESSING")
self.btn_stop.setEnabled(False)
self.status_label.setText(f"Done — {ok} succeeded, {fail} failed")
if fail > 0:
self._log(f"WARNING: {fail} file(s) failed. Check log above for details.")
if ok > 0:
self._log(f"All {ok} file(s) archived successfully.")
@Slot()
def _stop_process(self):
if self.worker and self.worker.isRunning():
self._log("STOP: Exiting queue after current file finishes...")
self.worker.stop()
self.btn_stop.setEnabled(False)
# ──────────────────────────────────────────────
# ENTRY POINT
# ──────────────────────────────────────────────
if __name__ == "__main__":
app = QApplication(sys.argv)
window = OpenCodecMaster()
window.show()
sys.exit(app.exec())
Reference in New Issue View Git Blame Copy Permalink