diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 0000000..f5f3a63 --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,66 @@ +# Builds the Windows .exe and the Linux binary. Runs on GitHub Actions as-is; +# Gitea/Forgejo Actions also read this folder but need a Windows runner registered. +name: Build + +on: + push: + tags: ["v*"] + workflow_dispatch: + +jobs: + build: + strategy: + fail-fast: false + matrix: + # ubuntu-22.04 keeps the glibc requirement at 2.35 so the binary runs on older distros. + os: [windows-latest, ubuntu-22.04] + runs-on: ${{ matrix.os }} + env: + QT_QPA_PLATFORM: offscreen + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install Qt runtime libraries (bundled into the Linux binary) + if: runner.os == 'Linux' + run: | + sudo apt-get update + sudo apt-get install -y libegl1 libxkbcommon-x11-0 libxcb-cursor0 libxcb-icccm4 \ + libxcb-image0 libxcb-keysyms1 libxcb-randr0 libxcb-render-util0 libxcb-shape0 libxcb-xkb1 + + - name: Install Python dependencies + run: python -m pip install -r requirements-dev.txt + + - name: Download ffmpeg + run: python scripts/fetch_ffmpeg.py + + - name: Run tests + run: python -m pytest -q + env: + FFWRAP_FFMPEG_DIR: ffmpeg/bin + + - name: Build executable + run: python build.py --ffmpeg-dir ffmpeg/bin + + - name: Smoke test the built executable + shell: bash + run: python scripts/smoke_test.py dist/package/FFmpegWrapper/FFmpegWrapper$([ "$RUNNER_OS" = Windows ] && echo .exe) smoke-${{ runner.os }}.png + + - uses: actions/upload-artifact@v4 + with: + name: FFmpegWrapper-${{ runner.os }} + path: | + dist/FFmpegWrapper-*.zip + dist/FFmpegWrapper-*.tar.gz + smoke-${{ runner.os }}.png + + - name: Attach to release + if: startsWith(github.ref, 'refs/tags/') + uses: softprops/action-gh-release@v2 + with: + files: | + dist/FFmpegWrapper-*.zip + dist/FFmpegWrapper-*.tar.gz diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..563b7fd --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +__pycache__/ +*.pyc +.venv/ +venv/ +build/ +dist/ +.pytest_cache/ +*.part.* +ffmpeg/ diff --git a/README.md b/README.md index e69de29..c0a9e82 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,108 @@ +# FFmpeg Wrapper + +A desktop GUI for common ffmpeg jobs: remuxing MKV files to Jellyfin-friendly MP4s, choosing which +audio tracks to keep, stripping audio, re-encoding, and scaling. Works on single files or whole batches. +It's a single executable for Windows and Linux, built with Python, Qt (PySide6) and PyInstaller. + +![icon](assets/icon.png) + +## Features + +- **Single file or batch.** Use *Add Files*, *Add Folder* (recursive), or drag and drop. Files are + probed in the background and shown with duration, video info, and audio and subtitle counts. +- **Target container.** MP4, MKV, MOV, M4V, WebM, AVI or MPEG-TS. The encoder lists only show codecs + the chosen container can hold. +- **Audio track picker.** Every audio track in the selected file is listed with its language, codec, + channels and title. Tick the tracks you want and drag them into order. The first ticked track + becomes the default. **Apply to All Files** copies your choice to the rest of the queue, + matching tracks by language, codec and layout, so a season of episodes takes one click. +- **Strip audio.** Set Audio mode to *Strip audio*. +- **Keep streams unaltered.** *Copy* mode remuxes video and/or audio without re-encoding. *Auto* + copies whatever the container supports and re-encodes only what it can't hold (for example + DTS/TrueHD to AAC for MP4). +- **Up/downscale video.** Pick 360p up to 8K, or a custom WxH, with aspect ratio kept and Lanczos + scaling. Portrait and rotated phone videos are handled. You can also change the frame rate. + Encoders: H.264, HEVC, AV1, VP9, MPEG-4, plus NVENC, QuickSync and AMF if your ffmpeg has them. + There are CRF quality and speed preset controls. +- **Up/downmix audio.** Change channels (mono, stereo, 5.1, 7.1), sample rate, bitrate, and codec + (AAC, AC-3, E-AC-3, Opus, MP3, Vorbis, FLAC, ALAC, PCM). +- **Output folder.** Save next to the source or in a chosen folder. You can add a filename suffix + and choose what happens when the output file already exists (keep both, overwrite, or skip). + The source file is never overwritten. +- **Subtitles** are kept where possible. MP4 and MOV turn text subs into `mov_text`, and image subs + (PGS) are dropped with a note in the log. MKV keeps everything, including font attachments. +- MP4/MOV outputs get `+faststart`, and HEVC is tagged `hvc1` for Apple and TV compatibility. +- The window shows per-file and overall progress, has a **Stop** button, and keeps a full log. + **Show Command** prints the exact ffmpeg command so you can copy it. +- ffmpeg writes to `name.part.ext` and the file is renamed only on success, so a failed or + cancelled job never leaves a half-written file. + +## Getting ffmpeg + +The app looks for `ffmpeg` and `ffprobe` in this order: + +1. The folder set in **Settings > Set FFmpeg Location...** +2. Next to the executable (or in `ffmpeg/`, `ffmpeg/bin/` or `bin/` beside it) +3. Your `PATH` + +The release archives (`FFmpegWrapper--windows-x64.zip` / `-linux-x64.tar.gz`) already +include ffmpeg and ffprobe next to the executable. If you use the bare executable, install ffmpeg +(`sudo apt install ffmpeg`, `winget install ffmpeg`, ...) or point the app at it. + +## Running from source + +```bash +python -m venv .venv +.venv/bin/pip install -r requirements-dev.txt # Windows: .venv\Scripts\pip ... +python scripts/fetch_ffmpeg.py # optional: downloads ffmpeg into ./ffmpeg/bin +python run.py +``` + +Tests: `python -m pytest`. The integration tests run real conversions when ffmpeg is found, either +on `PATH` or through `FFWRAP_FFMPEG_DIR=ffmpeg/bin`. + +## Building the executables + +PyInstaller does not cross-compile. Build the Windows `.exe` on Windows and the Linux binary on Linux. + +```bash +pip install -r requirements-dev.txt +python build.py # -> dist/FFmpegWrapper or dist/FFmpegWrapper.exe +python build.py --ffmpeg-dir ffmpeg/bin # also bundles ffmpeg + ffprobe into a .zip / .tar.gz +python scripts/smoke_test.py dist/FFmpegWrapper # optional: headless start-up check +``` + +`--onedir` builds a folder instead of a single file. It starts faster because nothing has to be +unpacked on launch. + +**CI:** [.github/workflows/build.yml](.github/workflows/build.yml) builds, tests and smoke-tests both +platforms on every `v*` tag (or a manual run) and attaches the archives to the release. Gitea and +Forgejo Actions can run it too, but they need a Windows runner registered. + +### Notes + +- **Linux:** the binary needs glibc 2.35 or newer (Ubuntu 22.04+, Debian 12+, Fedora 36+, + Mint 21+). It bundles the Qt XCB libraries, but uses the system's OpenGL/EGL driver libraries. + To get a menu entry, copy `assets/icon.png` and write a `.desktop` file that points at the binary. +- **PySide6 is pinned below 6.12.** PySide6 6.12.0 mis-counts references to Python's `None` + (in `QComboBox.clear()`/`currentData()`), which crashes the app on Python 3.11. Python 3.12+ is + not affected, but keep the pin until a fixed PySide6 ships. +- **Licensing:** the bundled ffmpeg builds come from + [BtbN/FFmpeg-Builds](https://github.com/BtbN/FFmpeg-Builds) and are GPL. Keep that in mind if you + redistribute the archives. +- Settings (last options, window size, ffmpeg location) are saved per user through Qt's QSettings: + the registry on Windows, `~/.config/ffwrap/FFmpegWrapper.conf` on Linux. + +## Project layout + +``` +ffwrap/ + formats.py container/encoder tables and compatibility rules (no Qt) + probe.py ffprobe -> MediaInfo/Stream dataclasses + command.py Options -> ffmpeg arguments, output naming, audio-track matching (no Qt) + tools.py locating ffmpeg/ffprobe, encoder detection + gui/ main window, queue model, QProcess batch runner +build.py PyInstaller build + packaging +scripts/ fetch_ffmpeg.py, smoke_test.py, make_icon.py +tests/ unit tests + real-ffmpeg integration tests +``` diff --git a/assets/icon.ico b/assets/icon.ico new file mode 100644 index 0000000..3da706d Binary files /dev/null and b/assets/icon.ico differ diff --git a/assets/icon.png b/assets/icon.png new file mode 100644 index 0000000..b726b07 Binary files /dev/null and b/assets/icon.png differ diff --git a/build.py b/build.py new file mode 100644 index 0000000..5edd9d6 --- /dev/null +++ b/build.py @@ -0,0 +1,82 @@ +"""Build a single-file GUI executable for the current OS with PyInstaller. + + python build.py # -> dist/FFmpegWrapper(.exe) + python build.py --ffmpeg-dir DIR # also copy ffmpeg + ffprobe from DIR next to the exe + # and pack everything into a .zip / .tar.gz + +PyInstaller cannot cross-compile: run this on Windows to get the .exe and on +Linux to get the Linux binary (or use the CI workflow, which does both). +""" + +from __future__ import annotations + +import argparse +import shutil +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +NAME = "FFmpegWrapper" +IS_WINDOWS = sys.platform.startswith("win") +EXE = ".exe" if IS_WINDOWS else "" + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--ffmpeg-dir", type=Path, help="folder containing ffmpeg and ffprobe to bundle") + ap.add_argument("--onedir", action="store_true", help="build a folder instead of a single file (faster start-up)") + args = ap.parse_args() + + import PyInstaller.__main__ + + from ffwrap import __version__ + + sep = ";" if IS_WINDOWS else ":" + pyi_args = [ + str(ROOT / "run.py"), + "--name", NAME, + "--windowed", + "--noconfirm", + "--clean", + "--onedir" if args.onedir else "--onefile", + "--icon", str(ROOT / "assets" / "icon.ico"), + "--add-data", f"{ROOT / 'assets' / 'icon.png'}{sep}assets", + "--distpath", str(ROOT / "dist"), + "--workpath", str(ROOT / "build"), + "--specpath", str(ROOT / "build"), + # Not needed and only add weight. + "--exclude-module", "tkinter", + "--exclude-module", "unittest", + "--exclude-module", "pydoc", + ] + PyInstaller.__main__.run(pyi_args) + + dist = ROOT / "dist" + built = dist / NAME if args.onedir else dist / f"{NAME}{EXE}" + print(f"\nBuilt {built}") + + if args.ffmpeg_dir: + # Stage + ffmpeg + ffprobe in one folder and archive it. + stage = dist / "package" / NAME + shutil.rmtree(stage.parent, ignore_errors=True) + if args.onedir: + shutil.copytree(built, stage) + else: + stage.mkdir(parents=True) + shutil.copy2(built, stage) + for tool in ("ffmpeg", "ffprobe"): + src = args.ffmpeg_dir / f"{tool}{EXE}" + if not src.is_file(): + print(f"error: {src} not found", file=sys.stderr) + return 1 + shutil.copy2(src, stage) + plat = "windows-x64" if IS_WINDOWS else "linux-x64" + archive = shutil.make_archive( + str(dist / f"{NAME}-{__version__}-{plat}"), "zip" if IS_WINDOWS else "gztar", + root_dir=stage.parent, base_dir=NAME) + print(f"Packed {archive}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ffwrap/__init__.py b/ffwrap/__init__.py new file mode 100644 index 0000000..f677803 --- /dev/null +++ b/ffwrap/__init__.py @@ -0,0 +1,4 @@ +"""FFmpeg Wrapper - a small GUI front-end for common ffmpeg conversions.""" + +__version__ = "1.0.0" +APP_NAME = "FFmpeg Wrapper" diff --git a/ffwrap/__main__.py b/ffwrap/__main__.py new file mode 100644 index 0000000..ac99fd1 --- /dev/null +++ b/ffwrap/__main__.py @@ -0,0 +1,5 @@ +import sys + +from .app import main + +sys.exit(main()) diff --git a/ffwrap/app.py b/ffwrap/app.py new file mode 100644 index 0000000..dd79bb5 --- /dev/null +++ b/ffwrap/app.py @@ -0,0 +1,45 @@ +"""Application entry point.""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path + + +def main() -> int: + from PySide6.QtGui import QIcon + from PySide6.QtWidgets import QApplication + + from . import APP_NAME + from .gui.main_window import MainWindow + from .tools import app_dir + + app = QApplication(sys.argv) + app.setApplicationName(APP_NAME) + app.setOrganizationName("ffwrap") + app.setStyle("Fusion") + + # Bundled by PyInstaller into sys._MEIPASS, or next to the sources. + base = Path(getattr(sys, "_MEIPASS", app_dir())) + icon = base / "assets" / "icon.png" + if icon.exists(): + app.setWindowIcon(QIcon(str(icon))) + + win = MainWindow() + win.show() + # Files passed on the command line (or dropped onto the executable). + args = [Path(a) for a in sys.argv[1:] if not a.startswith("-")] + if args: + win.add_paths(args) + + # Build verification: FFWRAP_SMOKE_TEST=shot.png saves a screenshot after start-up and exits. + shot = os.environ.get("FFWRAP_SMOKE_TEST") + if shot: + from PySide6.QtCore import QTimer + + def finish(): + win.grab().save(shot) + app.exit(0 if win.tools else 3) + QTimer.singleShot(4000, finish) + return app.exec() diff --git a/ffwrap/command.py b/ffwrap/command.py new file mode 100644 index 0000000..812516b --- /dev/null +++ b/ffwrap/command.py @@ -0,0 +1,366 @@ +"""Turn user options + probe results into an ffmpeg command line. + +Pure Python (no Qt) so it can be unit tested. +""" + +from __future__ import annotations + +import os +import shlex +import subprocess +from dataclasses import asdict, dataclass, field, fields +from pathlib import Path + +from .formats import ( + AUDIO_ENCODERS_BY_NAME, + CONTAINERS, + CUSTOM_RESOLUTION, + VIDEO_ENCODERS_BY_NAME, + container_accepts_audio, + container_accepts_video, + fps_expr, + subtitle_plan, +) +from .probe import MediaInfo, Stream +from .tools import IS_WINDOWS + +VIDEO_MODES = { + "copy": "Copy (keep original, no re-encode)", + "auto": "Auto (copy if compatible, else re-encode)", + "encode": "Re-encode", +} +AUDIO_MODES = { + "copy": "Copy (keep original, no re-encode)", + "auto": "Auto (copy if compatible, else re-encode)", + "encode": "Re-encode", + "strip": "Strip audio (no audio in output)", +} +EXISTS_POLICIES = { + "rename": "Add a number (keep both)", + "overwrite": "Overwrite", + "skip": "Skip file", +} + + +@dataclass +class Options: + container: str = "mp4" + output_dir: str = "" # empty = next to the source file + suffix: str = "" + if_exists: str = "rename" + + video_mode: str = "copy" + video_encoder: str = "libx264" + resolution: int = 0 # target height, 0 = original, -1 = custom + custom_width: int = 0 + custom_height: int = 0 + no_upscale: bool = False + fps: str = "Original" + quality: int = 20 + speed: int = 2 + + audio_mode: str = "auto" + audio_encoder: str = "aac" + audio_bitrate: int = 0 # kbps, 0 = auto + audio_channels: int = 0 # 0 = original + audio_sample_rate: int = 0 # Hz, 0 = original + + subtitles: bool = True + + def to_dict(self) -> dict: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict) -> "Options": + opts = cls() + for f in fields(cls): + if f.name in data: + try: + setattr(opts, f.name, type(getattr(opts, f.name))(data[f.name])) + except (TypeError, ValueError): + pass + return opts + + +@dataclass +class BuiltCommand: + args: list[str] + notes: list[str] = field(default_factory=list) # human readable decisions / warnings + + +# -------------------------------------------------------------------------- +# Video scaling +# -------------------------------------------------------------------------- + +def scale_filter(video: Stream, opts: Options) -> str | None: + """Return an ffmpeg ``scale`` filter for the requested resolution, or None.""" + size = video.display_size + sw, sh = size if size else (0, 0) + + if opts.resolution == CUSTOM_RESOLUTION: + w, h = max(0, opts.custom_width), max(0, opts.custom_height) + if not w and not h: + return None + if w and h: + factor = min(w / sw, h / sh) if size else 0 + expr = f"scale={w}:{h}:force_original_aspect_ratio=decrease:force_divisible_by=2" + elif w: + factor = w / sw if size else 0 + expr = f"scale={w - w % 2}:-2" + else: + factor = h / sh if size else 0 + expr = f"scale=-2:{h - h % 2}" + elif opts.resolution > 0: + target = opts.resolution + if size and sh > sw: + # Portrait video: "1080p" means 1080 pixels wide. + factor = target / sw + expr = f"scale={target}:-2" + else: + factor = target / sh if size else 0 + expr = f"scale=-2:{target}" + else: + return None + + if size: + if abs(factor - 1) < 1e-6: + return None + if opts.no_upscale and factor > 1: + return None + return expr + ":flags=lanczos" + + +# -------------------------------------------------------------------------- +# Command builder +# -------------------------------------------------------------------------- + +def build_command( + ffmpeg: str, + info: MediaInfo, + opts: Options, + audio_indices: list[int], + output: Path, +) -> BuiltCommand: + container = CONTAINERS[opts.container] + notes: list[str] = [] + args = [ + ffmpeg, "-hide_banner", "-nostdin", "-y", + "-loglevel", "warning", + "-progress", "pipe:1", "-nostats", + "-i", str(info.path), + ] + maps: list[str] = [] + codec_args: list[str] = [] + + # ---- video ----------------------------------------------------------- + video = info.video + if video is not None: + maps += ["-map", f"0:{video.index}"] + filters = [] + scale = scale_filter(video, opts) + if scale: + filters.append(scale) + fps = fps_expr(opts.fps) + if fps: + filters.append(f"fps={fps}") + + if opts.video_mode == "copy": + encode = False + if not container_accepts_video(container, video.codec): + notes.append(f"Warning: {video.codec} video may not be supported in {container.ext} when copying") + elif opts.video_mode == "auto": + encode = bool(filters) or not container_accepts_video(container, video.codec) + else: + encode = True + + if encode: + enc = VIDEO_ENCODERS_BY_NAME[opts.video_encoder] + if filters: + codec_args += ["-vf", ",".join(filters)] + codec_args += enc.args(opts.quality, opts.speed) + out_codec = enc.codec + notes.append(f"Video: re-encode {video.codec} -> {enc.label}" + (f" [{', '.join(filters)}]" if filters else "")) + else: + codec_args += ["-c:v", "copy"] + out_codec = video.codec + notes.append(f"Video: copy {video.codec}") + if out_codec == "hevc" and opts.container in ("mp4", "mov", "m4v"): + # Apple devices (and some smart TVs) only play HEVC tagged as hvc1. + codec_args += ["-tag:v", "hvc1"] + + # ---- audio ----------------------------------------------------------- + if opts.audio_mode == "strip": + codec_args.append("-an") + notes.append("Audio: stripped") + else: + selected = [s for s in (info.stream(i) for i in audio_indices) if s is not None and s.kind == "audio"] + if not selected and info.audio: + notes.append("Audio: no tracks selected - output will have no audio") + enc = AUDIO_ENCODERS_BY_NAME[opts.audio_encoder] + for i, s in enumerate(selected): + maps += ["-map", f"0:{s.index}"] + codec_args += _audio_track_args(i, s, opts, container, enc, notes) + codec_args += [f"-disposition:a:{i}", "default" if i == 0 else "0"] + + # ---- subtitles ------------------------------------------------------- + if opts.subtitles: + j = 0 + for s in info.subtitles: + plan = subtitle_plan(container, s.codec) + if plan is None: + notes.append(f"Subtitles: dropped track #{s.index} ({s.codec}) - not supported in {container.ext}") + continue + maps += ["-map", f"0:{s.index}"] + codec_args += [f"-c:s:{j}", plan] + j += 1 + if opts.container == "mkv" and any(s.kind == "attachment" for s in info.streams): + # Fonts used by ASS subtitles. + maps += ["-map", "0:t?"] + codec_args += ["-c:t", "copy"] + + args += maps + codec_args + args += ["-max_muxing_queue_size", "4096"] + args += list(container.extra_args) + args.append(str(output)) + return BuiltCommand(args=args, notes=notes) + + +def _audio_track_args(i, s: Stream, opts: Options, container, enc, notes: list[str]) -> list[str]: + want_ch, want_sr = opts.audio_channels, opts.audio_sample_rate + label = f"Audio #{s.index}" + (f" [{s.language}]" if s.language else "") + + if opts.audio_mode == "copy": + if not container_accepts_audio(container, s.codec): + notes.append(f"Warning: {label}: {s.codec} may not be supported in {container.ext} when copying") + notes.append(f"{label}: copy {s.codec}") + return [f"-c:a:{i}", "copy"] + + if opts.audio_mode == "auto": + changes = (want_ch and want_ch != s.channels) or (want_sr and want_sr != s.sample_rate) + if container_accepts_audio(container, s.codec) and not changes: + notes.append(f"{label}: copy {s.codec}") + return [f"-c:a:{i}", "copy"] + + out = [f"-c:a:{i}", enc.name] + channels = want_ch or s.channels or 2 + if channels > enc.max_channels: + notes.append(f"{label}: {enc.label} supports at most {enc.max_channels} channels - downmixing") + channels = enc.max_channels + out += [f"-ac:a:{i}", str(channels)] + elif want_ch: + out += [f"-ac:a:{i}", str(channels)] + + if want_sr: + rate = enc.nearest_sample_rate(want_sr) + if rate != want_sr: + notes.append(f"{label}: {enc.label} does not support {want_sr} Hz - using {rate} Hz") + out += [f"-ar:a:{i}", str(rate)] + + if not enc.lossless: + kbps = min(opts.audio_bitrate or enc.auto_bitrate(channels), enc.max_kbps) + out += [f"-b:a:{i}", f"{kbps}k"] + if enc.name == "libopus" and channels > 2: + out += [f"-mapping_family:a:{i}", "1"] + notes.append(f"{label}: re-encode {s.codec} -> {enc.label}") + return out + + +# -------------------------------------------------------------------------- +# Output paths +# -------------------------------------------------------------------------- + +def _norm(p: Path) -> str: + return os.path.normcase(os.path.abspath(p)) + + +def plan_output_path(src: Path, opts: Options, taken: set[str], exists=os.path.exists) -> Path | None: + """Pick the output file for ``src``. Returns None if the file should be skipped. + + ``taken`` holds normalised paths already produced in this batch; they are + never overwritten even with the "overwrite" policy. + """ + ext = CONTAINERS[opts.container].ext + out_dir = Path(opts.output_dir) if opts.output_dir else src.parent + stem = src.stem + opts.suffix + cand = out_dir / f"{stem}{ext}" + if _norm(cand) == _norm(src): + # Never overwrite the input (e.g. mp4 -> mp4 in the same folder). + stem += "_converted" + cand = out_dir / f"{stem}{ext}" + + in_batch = _norm(cand) in taken + if not in_batch and not exists(cand): + return cand + if not in_batch and opts.if_exists == "skip": + return None + if not in_batch and opts.if_exists == "overwrite" and _norm(cand) != _norm(src): + return cand + n = 1 + while True: + p = out_dir / f"{stem} ({n}){ext}" + if _norm(p) not in taken and not exists(p) and _norm(p) != _norm(src): + return p + n += 1 + + +def temp_output_path(final: Path) -> Path: + """ffmpeg writes here first; renamed to ``final`` only on success.""" + return final.with_name(f"{final.stem}.part{final.suffix}") + + +# -------------------------------------------------------------------------- +# Audio track selection helpers +# -------------------------------------------------------------------------- + +def default_audio_selection(info: MediaInfo) -> list[int]: + return [s.index for s in info.audio] + + +def match_audio_selection(ref_all: list[Stream], ref_selected: list[int], target_all: list[Stream]) -> list[int]: + """Map a track selection made on one file onto another file's audio tracks. + + Identical track layouts map by position; otherwise each selected track is + matched by language, then codec / channels / title / position. Falls back to + the target's default (or first) track so audio is never silently lost. + """ + if not target_all: + return [] + if not ref_selected: + return [] + sig = lambda s: (s.language, s.codec, s.channels) # noqa: E731 + ref_pos = {s.index: pos for pos, s in enumerate(ref_all)} + by_index = {s.index: s for s in ref_all} + + if [sig(s) for s in ref_all] == [sig(s) for s in target_all]: + return [target_all[ref_pos[i]].index for i in ref_selected if i in ref_pos] + + result: list[int] = [] + for i in ref_selected: + r = by_index.get(i) + if r is None: + continue + best, best_score = None, 0 + for pos, t in enumerate(target_all): + if t.index in result: + continue + if r.language and t.language and r.language != t.language: + continue + score = 1 + score += 8 if r.language and r.language == t.language else 0 + score += 2 if r.codec == t.codec else 0 + score += 2 if r.channels == t.channels else 0 + score += 2 if r.title and r.title == t.title else 0 + score += 1 if pos == ref_pos[i] else 0 + if score > best_score: + best, best_score = t, score + if best is not None: + result.append(best.index) + + if not result: + default = next((t for t in target_all if t.default), target_all[0]) + result = [default.index] + return result + + +def format_command(args: list[str]) -> str: + return subprocess.list2cmdline(args) if IS_WINDOWS else shlex.join(args) diff --git a/ffwrap/formats.py b/ffwrap/formats.py new file mode 100644 index 0000000..198ab0c --- /dev/null +++ b/ffwrap/formats.py @@ -0,0 +1,248 @@ +"""Static knowledge about containers and encoders. + +Everything in here is plain data so it can be used (and tested) without Qt. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +# -------------------------------------------------------------------------- +# Containers +# -------------------------------------------------------------------------- + +TEXT_SUBTITLE_CODECS = { + "subrip", "srt", "ass", "ssa", "mov_text", "webvtt", "text", + "subviewer", "subviewer1", "microdvd", "sami", "realtext", "jacosub", +} + + +@dataclass(frozen=True) +class Container: + key: str + label: str + ext: str + # Codec names (as reported by ffprobe) that can be stream-copied into this + # container. ``None`` means "anything goes" (Matroska). + video_codecs: frozenset[str] | None + audio_codecs: frozenset[str] | None + # How text subtitles are stored: a target codec name, "copy", or None + # (container cannot hold text subtitles). + text_subtitles: str | None + # Image subtitles (PGS, VobSub, DVB) can only be copied, never converted + # to text, so a container either accepts them or not. + image_subtitles: frozenset[str] | None = frozenset() + extra_args: tuple[str, ...] = () + + +_MP4_VIDEO = frozenset({"h264", "hevc", "av1", "vp9", "mpeg4", "mpeg2video", "mpeg1video", "mjpeg"}) +_MP4_AUDIO = frozenset({"aac", "mp3", "ac3", "eac3", "opus", "flac", "alac", "mp2"}) +_FASTSTART = ("-movflags", "+faststart") + +CONTAINERS: dict[str, Container] = { + c.key: c + for c in ( + Container("mp4", "MP4 (.mp4)", ".mp4", _MP4_VIDEO, _MP4_AUDIO, "mov_text", extra_args=_FASTSTART), + Container("mkv", "Matroska (.mkv)", ".mkv", None, None, "copy", image_subtitles=None), + Container( + "mov", "QuickTime (.mov)", ".mov", + frozenset({"h264", "hevc", "mpeg4", "prores", "mjpeg", "dnxhd", "mpeg2video"}), + frozenset({"aac", "mp3", "ac3", "eac3", "alac", "pcm_s16le", "pcm_s24le", "pcm_s16be", "pcm_s24be"}), + "mov_text", extra_args=_FASTSTART, + ), + Container( + "m4v", "MPEG-4 Video (.m4v)", ".m4v", + frozenset({"h264", "hevc"}), frozenset({"aac", "ac3", "eac3", "alac"}), + "mov_text", extra_args=_FASTSTART, + ), + Container( + "webm", "WebM (.webm)", ".webm", + frozenset({"vp8", "vp9", "av1"}), frozenset({"opus", "vorbis"}), "webvtt", + ), + Container( + "avi", "AVI (.avi)", ".avi", + frozenset({"mpeg4", "h264", "mjpeg", "msmpeg4v3", "mpeg2video", "mpeg1video"}), + frozenset({"mp3", "ac3", "pcm_s16le", "mp2", "aac"}), None, + ), + Container( + "ts", "MPEG-TS (.ts)", ".ts", + frozenset({"h264", "hevc", "mpeg2video", "mpeg1video", "av1"}), + frozenset({"aac", "ac3", "eac3", "mp3", "mp2", "opus"}), + None, image_subtitles=frozenset({"dvb_subtitle"}), + ), + ) +} + + +def container_accepts_video(container: Container, codec: str | None) -> bool: + return container.video_codecs is None or codec in container.video_codecs + + +def container_accepts_audio(container: Container, codec: str | None) -> bool: + return container.audio_codecs is None or codec in container.audio_codecs + + +def subtitle_plan(container: Container, codec: str | None) -> str | None: + """How a subtitle stream with ``codec`` goes into ``container``. + + Returns "copy", a target encoder name, or None if it must be dropped. + """ + if codec in TEXT_SUBTITLE_CODECS: + target = container.text_subtitles + if target == "copy": + # Matroska can hold every text format except mov_text. + return "srt" if codec == "mov_text" else "copy" + if target is not None and target == codec: + return "copy" + return target + # Image based (or unknown) subtitles. + if container.image_subtitles is None or codec in container.image_subtitles: + return "copy" + return None + + +# -------------------------------------------------------------------------- +# Video encoders +# -------------------------------------------------------------------------- + +SPEED_LEVELS = ("Fastest", "Fast", "Balanced", "Slow", "Slowest") + + +@dataclass(frozen=True) +class VideoEncoder: + name: str # ffmpeg encoder name + label: str + codec: str # resulting codec name (as ffprobe reports it) + quality_range: tuple[int, int] + quality_default: int + quality_args: tuple[str, ...] # "{q}" is replaced by the quality value + speed_values: tuple[str, ...] # one entry per SPEED_LEVELS + speed_option: str | None + extra_args: tuple[str, ...] = () + # Force 8-bit 4:2:0 output; most H.264 players cannot decode High10. + pix_fmt: str | None = None + + def args(self, quality: int, speed: int) -> list[str]: + q = max(self.quality_range[0], min(self.quality_range[1], quality)) + out = ["-c:v", self.name] + out += [a.replace("{q}", str(q)) for a in self.quality_args] + if self.speed_option: + out += [self.speed_option, self.speed_values[max(0, min(4, speed))]] + out += list(self.extra_args) + if self.pix_fmt: + out += ["-pix_fmt", self.pix_fmt] + return out + + +_X26X_SPEEDS = ("ultrafast", "veryfast", "medium", "slow", "veryslow") +_NVENC_SPEEDS = ("p1", "p3", "p4", "p6", "p7") +_QSV_SPEEDS = ("veryfast", "faster", "medium", "slow", "veryslow") +_AMF_SPEEDS = ("speed", "speed", "balanced", "quality", "quality") +_NVENC_Q = ("-rc", "vbr", "-cq", "{q}", "-b:v", "0") +_AMF_Q = ("-rc", "cqp", "-qp_i", "{q}", "-qp_p", "{q}") + +VIDEO_ENCODERS: list[VideoEncoder] = [ + VideoEncoder("libx264", "H.264 / AVC (x264)", "h264", (0, 51), 20, ("-crf", "{q}"), _X26X_SPEEDS, "-preset", pix_fmt="yuv420p"), + VideoEncoder("libx265", "H.265 / HEVC (x265)", "hevc", (0, 51), 22, ("-crf", "{q}"), _X26X_SPEEDS, "-preset"), + VideoEncoder("libsvtav1", "AV1 (SVT-AV1)", "av1", (0, 63), 30, ("-crf", "{q}"), ("12", "10", "8", "6", "4"), "-preset"), + VideoEncoder("libvpx-vp9", "VP9 (libvpx)", "vp9", (0, 63), 31, ("-crf", "{q}", "-b:v", "0"), ("5", "4", "3", "1", "0"), "-cpu-used", ("-row-mt", "1")), + VideoEncoder("mpeg4", "MPEG-4 Part 2 (for AVI)", "mpeg4", (2, 31), 4, ("-q:v", "{q}"), ("",) * 5, None), + VideoEncoder("h264_nvenc", "H.264 - NVIDIA NVENC", "h264", (0, 51), 23, _NVENC_Q, _NVENC_SPEEDS, "-preset", pix_fmt="yuv420p"), + VideoEncoder("hevc_nvenc", "H.265 - NVIDIA NVENC", "hevc", (0, 51), 25, _NVENC_Q, _NVENC_SPEEDS, "-preset"), + VideoEncoder("av1_nvenc", "AV1 - NVIDIA NVENC", "av1", (0, 51), 30, _NVENC_Q, _NVENC_SPEEDS, "-preset"), + VideoEncoder("h264_qsv", "H.264 - Intel QuickSync", "h264", (1, 51), 23, ("-global_quality", "{q}"), _QSV_SPEEDS, "-preset", pix_fmt="nv12"), + VideoEncoder("hevc_qsv", "H.265 - Intel QuickSync", "hevc", (1, 51), 25, ("-global_quality", "{q}"), _QSV_SPEEDS, "-preset"), + VideoEncoder("av1_qsv", "AV1 - Intel QuickSync", "av1", (1, 51), 30, ("-global_quality", "{q}"), _QSV_SPEEDS, "-preset"), + VideoEncoder("h264_amf", "H.264 - AMD AMF", "h264", (0, 51), 23, _AMF_Q, _AMF_SPEEDS, "-quality", pix_fmt="nv12"), + VideoEncoder("hevc_amf", "H.265 - AMD AMF", "hevc", (0, 51), 25, _AMF_Q, _AMF_SPEEDS, "-quality"), +] +VIDEO_ENCODERS_BY_NAME = {e.name: e for e in VIDEO_ENCODERS} + +# (label, height) - height 0 means "keep original" +RESOLUTIONS: list[tuple[str, int]] = [ + ("Original", 0), + ("4320p (8K)", 4320), + ("2160p (4K)", 2160), + ("1440p", 1440), + ("1080p", 1080), + ("720p", 720), + ("576p", 576), + ("480p", 480), + ("360p", 360), +] +CUSTOM_RESOLUTION = -1 + +FRAME_RATES = ["Original", "23.976", "24", "25", "29.97", "30", "50", "59.94", "60"] +_FPS_EXPR = {"23.976": "24000/1001", "29.97": "30000/1001", "59.94": "60000/1001"} + + +def fps_expr(label: str) -> str | None: + if not label or label == "Original": + return None + return _FPS_EXPR.get(label, label) + + +# -------------------------------------------------------------------------- +# Audio encoders +# -------------------------------------------------------------------------- + +@dataclass(frozen=True) +class AudioEncoder: + name: str + label: str + codec: str + max_channels: int = 8 + sample_rates: frozenset[int] | None = None # None = anything + lossless: bool = False + # Default bitrate (kbps) for stereo and for 5.1; scaled for other layouts. + default_kbps: tuple[int, int] = (192, 384) + max_kbps: int = 1536 + extra_args: tuple[str, ...] = field(default=()) + + def auto_bitrate(self, channels: int) -> int: + stereo, surround = self.default_kbps + if channels <= 1: + kbps = stereo // 2 + elif channels <= 2: + kbps = stereo + else: + kbps = surround * channels // 6 + return min(kbps, self.max_kbps) + + def nearest_sample_rate(self, rate: int) -> int: + if self.sample_rates is None or rate in self.sample_rates: + return rate + higher = sorted(r for r in self.sample_rates if r >= rate) + return higher[0] if higher else max(self.sample_rates) + + +_MP3_RATES = frozenset({8000, 11025, 12000, 16000, 22050, 24000, 32000, 44100, 48000}) + +AUDIO_ENCODERS: list[AudioEncoder] = [ + AudioEncoder("aac", "AAC", "aac", default_kbps=(192, 384), max_kbps=768), + AudioEncoder("ac3", "Dolby Digital (AC-3)", "ac3", max_channels=6, + sample_rates=frozenset({32000, 44100, 48000}), default_kbps=(224, 448), max_kbps=640), + AudioEncoder("eac3", "Dolby Digital Plus (E-AC-3)", "eac3", max_channels=6, + sample_rates=frozenset({32000, 44100, 48000}), default_kbps=(224, 640), max_kbps=6144), + AudioEncoder("libopus", "Opus", "opus", sample_rates=frozenset({8000, 12000, 16000, 24000, 48000}), + default_kbps=(128, 320), max_kbps=512), + AudioEncoder("libmp3lame", "MP3", "mp3", max_channels=2, sample_rates=_MP3_RATES, + default_kbps=(192, 192), max_kbps=320), + AudioEncoder("libvorbis", "Vorbis", "vorbis", default_kbps=(192, 448), max_kbps=500), + AudioEncoder("flac", "FLAC (lossless)", "flac", lossless=True), + AudioEncoder("alac", "ALAC (lossless)", "alac", lossless=True), + AudioEncoder("pcm_s16le", "PCM 16-bit (uncompressed)", "pcm_s16le", lossless=True), +] +AUDIO_ENCODERS_BY_NAME = {e.name: e for e in AUDIO_ENCODERS} + +AUDIO_BITRATES = [0, 64, 96, 128, 160, 192, 224, 256, 320, 384, 448, 512, 640] # 0 = auto +AUDIO_CHANNELS: list[tuple[str, int]] = [ + ("Original", 0), ("Mono (1.0)", 1), ("Stereo (2.0)", 2), ("5.1", 6), ("7.1", 8), +] +AUDIO_SAMPLE_RATES = [0, 22050, 32000, 44100, 48000, 88200, 96000] # 0 = original + + +def describe_channels(n: int | None, layout: str | None = None) -> str: + if layout and layout not in ("unknown", ""): + return {"mono": "1.0", "stereo": "2.0"}.get(layout, layout.replace("(side)", "")) + return {1: "1.0", 2: "2.0", 6: "5.1", 8: "7.1"}.get(n or 0, f"{n}ch" if n else "?ch") diff --git a/ffwrap/gui/__init__.py b/ffwrap/gui/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ffwrap/gui/main_window.py b/ffwrap/gui/main_window.py new file mode 100644 index 0000000..dc42e0e --- /dev/null +++ b/ffwrap/gui/main_window.py @@ -0,0 +1,881 @@ +"""Main application window.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +from PySide6.QtCore import QObject, QRunnable, QSettings, Qt, QThreadPool, QUrl, Signal +from PySide6.QtGui import QAction, QDesktopServices, QFontDatabase, QKeySequence +from PySide6.QtWidgets import ( + QAbstractItemView, QCheckBox, QComboBox, QFileDialog, QFormLayout, QGroupBox, + QHBoxLayout, QHeaderView, QLabel, QLineEdit, QListWidget, QListWidgetItem, + QMainWindow, QMenu, QMessageBox, QPlainTextEdit, QProgressBar, QPushButton, + QRadioButton, QSpinBox, QSplitter, QTableView, QToolBar, QVBoxLayout, QWidget, +) + +from .. import APP_NAME, __version__ +from ..command import ( + AUDIO_MODES, EXISTS_POLICIES, VIDEO_MODES, Options, build_command, format_command, + match_audio_selection, plan_output_path, +) +from ..formats import ( + AUDIO_BITRATES, AUDIO_CHANNELS, AUDIO_ENCODERS, AUDIO_ENCODERS_BY_NAME, AUDIO_SAMPLE_RATES, CONTAINERS, + CUSTOM_RESOLUTION, FRAME_RATES, RESOLUTIONS, SPEED_LEVELS, TEXT_SUBTITLE_CODECS, + VIDEO_ENCODERS, VIDEO_ENCODERS_BY_NAME, +) +from ..probe import MediaInfo, ProbeError, probe +from ..tools import FFmpegTools, locate_tools +from .queue_model import COL_PROGRESS, RUNNABLE, Job, ProgressDelegate, QueueModel, Status +from .runner import BatchRunner + +MEDIA_EXTENSIONS = { + ".mkv", ".mp4", ".m4v", ".mov", ".avi", ".wmv", ".flv", ".webm", ".ts", ".m2ts", ".mts", + ".mpg", ".mpeg", ".vob", ".3gp", ".ogv", ".divx", ".rmvb", ".asf", ".mxf", ".dv", + ".mka", ".mp3", ".flac", ".m4a", ".wav", ".ogg", ".opus", ".ac3", ".dts", ".wma", +} + + +# -------------------------------------------------------------------------- +# Background probing +# -------------------------------------------------------------------------- + +class _ProbeSignals(QObject): + done = Signal(object, object, str) # job, MediaInfo | None, error + + +class _ProbeTask(QRunnable): + def __init__(self, ffprobe: str, job: Job, signals: _ProbeSignals): + super().__init__() + self.ffprobe, self.job, self.signals = ffprobe, job, signals + + def run(self): + try: + info = probe(self.ffprobe, self.job.path) + self.signals.done.emit(self.job, info, "") + except ProbeError as e: + self.signals.done.emit(self.job, None, str(e)) + except Exception as e: # never let a worker thread die silently + self.signals.done.emit(self.job, None, f"{type(e).__name__}: {e}") + + +class TrackList(QListWidget): + """Checkable, drag-reorderable list of audio tracks.""" + order_changed = Signal() + + def __init__(self, parent=None): + super().__init__(parent) + self.setDragDropMode(QAbstractItemView.InternalMove) + self.setDefaultDropAction(Qt.MoveAction) + self.setSelectionMode(QAbstractItemView.SingleSelection) + + def dropEvent(self, event): + super().dropEvent(event) + self.order_changed.emit() + + +def _combo(items: list[tuple[str, object]]) -> QComboBox: + cb = QComboBox() + for label, data in items: + cb.addItem(label, data) + return cb + + +def _select_data(cb: QComboBox, value) -> bool: + i = cb.findData(value) + if i >= 0: + cb.setCurrentIndex(i) + return True + return False + + +# -------------------------------------------------------------------------- +# Main window +# -------------------------------------------------------------------------- + +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle(APP_NAME) + self.resize(1280, 880) + self.setAcceptDrops(True) + + self.settings = QSettings("ffwrap", "FFmpegWrapper") + self.tools: FFmpegTools | None = None + self.model = QueueModel(self) + self.runner = BatchRunner(self) + self.pool = QThreadPool(self) + self.pool.setMaxThreadCount(max(2, min(6, os.cpu_count() or 2))) + self.probe_signals = _ProbeSignals() + self.probe_signals.done.connect(self._on_probed) + self._track_job: Job | None = None + self._loading = False + + self._build_ui() + self._connect() + self._load_settings() + self._init_tools() + + # ------------------------------------------------------------------ UI + def _build_ui(self) -> None: + # Toolbar / menus + tb = QToolBar("Main") + tb.setMovable(False) + tb.setToolButtonStyle(Qt.ToolButtonTextOnly) + self.addToolBar(tb) + self.act_add = QAction("Add Files...", self, shortcut=QKeySequence.Open) + self.act_add_dir = QAction("Add Folder...", self) + self.act_remove = QAction("Remove Selected", self, shortcut=QKeySequence.Delete) + self.act_clear = QAction("Clear Queue", self) + for a in (self.act_add, self.act_add_dir, self.act_remove, self.act_clear): + tb.addAction(a) + + def action(text, slot, shortcut=None): + a = QAction(text, self) + a.triggered.connect(slot) + if shortcut is not None: + a.setShortcut(shortcut) + return a + + m_file = self.menuBar().addMenu("&File") + m_file.addAction(self.act_add) + m_file.addAction(self.act_add_dir) + m_file.addSeparator() + m_file.addAction(action("Quit", self.close, QKeySequence.Quit)) + m_settings = self.menuBar().addMenu("&Settings") + m_settings.addAction(action("Set FFmpeg Location...", self._choose_ffmpeg_dir)) + m_settings.addAction(action("Use Automatic FFmpeg Detection", self._reset_ffmpeg_dir)) + m_settings.addSeparator() + m_settings.addAction(action("Reset Conversion Options", self._reset_options)) + self.menuBar().addMenu("&Help").addAction(action("About", self._about)) + + # Queue table + self.table = QTableView() + self.table.setModel(self.model) + self.table.setSelectionBehavior(QAbstractItemView.SelectRows) + self.table.setSelectionMode(QAbstractItemView.ExtendedSelection) + self.table.setAlternatingRowColors(True) + self.table.verticalHeader().setVisible(False) + self.table.setContextMenuPolicy(Qt.CustomContextMenu) + self.table.setItemDelegateForColumn(COL_PROGRESS, ProgressDelegate(self.table)) + hh = self.table.horizontalHeader() + hh.setSectionResizeMode(QHeaderView.Interactive) + hh.setSectionResizeMode(0, QHeaderView.Stretch) + for col, w in ((1, 70), (2, 190), (3, 70), (4, 45), (5, 190), (6, 110)): + self.table.setColumnWidth(col, w) + self.drop_hint = QLabel("Drop video files or folders here, or use Add Files...") + self.drop_hint.setAlignment(Qt.AlignCenter) + self.drop_hint.setStyleSheet("color: gray;") + + queue_box = QWidget() + ql = QVBoxLayout(queue_box) + ql.setContentsMargins(0, 0, 0, 0) + ql.addWidget(self.table) + ql.addWidget(self.drop_hint) + + # Audio track panel + tracks_box = QGroupBox("Audio tracks") + tl = QVBoxLayout(tracks_box) + self.tracks_title = QLabel("Select a file in the queue") + self.tracks_title.setWordWrap(True) + self.track_list = TrackList() + hint = QLabel("Tick the tracks to keep. Drag to reorder - the first ticked track becomes the default.") + hint.setWordWrap(True) + hint.setStyleSheet("color: gray;") + self.stream_info = QLabel("") + self.stream_info.setWordWrap(True) + row = QHBoxLayout() + self.btn_all = QPushButton("All") + self.btn_none = QPushButton("None") + self.btn_apply_all = QPushButton("Apply to All Files") + self.btn_apply_all.setToolTip( + "Copy this track selection to every other file in the queue.\n" + "Tracks are matched by language, codec and channel layout.") + row.addWidget(self.btn_all) + row.addWidget(self.btn_none) + row.addStretch(1) + row.addWidget(self.btn_apply_all) + tl.addWidget(self.tracks_title) + tl.addWidget(self.track_list, 1) + tl.addWidget(hint) + tl.addLayout(row) + tl.addWidget(self.stream_info) + + top = QSplitter(Qt.Horizontal) + top.addWidget(queue_box) + top.addWidget(tracks_box) + top.setStretchFactor(0, 3) + top.setStretchFactor(1, 2) + top.setSizes([800, 460]) + + # Settings + self.settings_box = QWidget() + sl = QHBoxLayout(self.settings_box) + sl.setContentsMargins(0, 0, 0, 0) + sl.addWidget(self._build_output_group(), 1) + sl.addWidget(self._build_video_group(), 1) + sl.addWidget(self._build_audio_group(), 1) + + # Log + self.log_view = QPlainTextEdit() + self.log_view.setReadOnly(True) + self.log_view.setMaximumBlockCount(10000) + self.log_view.setFont(QFontDatabase.systemFont(QFontDatabase.FixedFont)) + + splitter = QSplitter(Qt.Vertical) + splitter.addWidget(top) + splitter.addWidget(self.settings_box) + splitter.addWidget(self.log_view) + splitter.setStretchFactor(0, 3) + splitter.setStretchFactor(2, 1) + splitter.setCollapsible(1, False) + splitter.setSizes([380, 300, 160]) + + # Bottom bar + self.overall = QProgressBar() + self.overall.setRange(0, 1000) + self.overall.setFormat("%p%") + self.btn_preview = QPushButton("Show Command") + self.btn_preview.setToolTip("Print the ffmpeg command for the selected file to the log") + self.btn_start = QPushButton("Start") + self.btn_start.setDefault(True) + self.btn_start.setMinimumWidth(110) + self.btn_stop = QPushButton("Stop") + self.btn_stop.setEnabled(False) + bottom = QHBoxLayout() + bottom.addWidget(self.overall, 1) + bottom.addWidget(self.btn_preview) + bottom.addWidget(self.btn_start) + bottom.addWidget(self.btn_stop) + + central = QWidget() + cl = QVBoxLayout(central) + cl.addWidget(splitter, 1) + cl.addLayout(bottom) + self.setCentralWidget(central) + self.ffmpeg_label = QLabel() + self.statusBar().addPermanentWidget(self.ffmpeg_label) + + def _build_output_group(self) -> QGroupBox: + box = QGroupBox("Output") + f = QFormLayout(box) + self.cb_container = _combo([(c.label, c.key) for c in CONTAINERS.values()]) + f.addRow("Container:", self.cb_container) + + self.rb_same_dir = QRadioButton("Same folder as source") + self.rb_custom_dir = QRadioButton("Custom folder:") + self.ed_out_dir = QLineEdit() + self.ed_out_dir.setPlaceholderText("Choose an output folder") + self.btn_out_dir = QPushButton("Browse...") + dir_row = QHBoxLayout() + dir_row.addWidget(self.ed_out_dir, 1) + dir_row.addWidget(self.btn_out_dir) + f.addRow("Save to:", self.rb_same_dir) + f.addRow("", self.rb_custom_dir) + f.addRow("", dir_row) + + self.ed_suffix = QLineEdit() + self.ed_suffix.setPlaceholderText("e.g. _jellyfin (optional)") + f.addRow("Name suffix:", self.ed_suffix) + self.cb_exists = _combo([(v, k) for k, v in EXISTS_POLICIES.items()]) + f.addRow("If file exists:", self.cb_exists) + self.chk_subs = QCheckBox("Keep subtitles (converted where needed)") + self.chk_subs.setToolTip( + "MP4/MOV: text subtitles become mov_text; image subtitles (PGS/VobSub) are dropped.\n" + "MKV: all subtitles are kept as-is. WebM: text becomes WebVTT. AVI: none.") + f.addRow("", self.chk_subs) + return box + + def _build_video_group(self) -> QGroupBox: + self.video_box = box = QGroupBox("Video") + f = QFormLayout(box) + self.cb_video_mode = _combo([(v, k) for k, v in VIDEO_MODES.items()]) + f.addRow("Mode:", self.cb_video_mode) + self.cb_vcodec = QComboBox() + f.addRow("Encoder:", self.cb_vcodec) + + self.cb_resolution = _combo([(label, h) for label, h in RESOLUTIONS] + [("Custom...", CUSTOM_RESOLUTION)]) + self.cb_resolution.setToolTip("Target size of the shorter side; aspect ratio is kept.") + f.addRow("Resolution:", self.cb_resolution) + self.sp_width = QSpinBox() + self.sp_height = QSpinBox() + for sp in (self.sp_width, self.sp_height): + sp.setRange(0, 16384) + sp.setSingleStep(2) + sp.setSpecialValueText("auto") + self.custom_row = QWidget() + cr = QHBoxLayout(self.custom_row) + cr.setContentsMargins(0, 0, 0, 0) + cr.addWidget(self.sp_width, 1) + cr.addWidget(QLabel("x")) + cr.addWidget(self.sp_height, 1) + f.addRow("Width x Height:", self.custom_row) + self.custom_row_label = f.labelForField(self.custom_row) + self.chk_no_upscale = QCheckBox("Never upscale smaller videos") + f.addRow("", self.chk_no_upscale) + + self.cb_fps = _combo([(v, v) for v in FRAME_RATES]) + f.addRow("Frame rate:", self.cb_fps) + self.sp_quality = QSpinBox() + self.sp_quality.setToolTip("CRF / CQ value. Lower = better quality and bigger files.") + f.addRow("Quality (lower=better):", self.sp_quality) + self.cb_speed = _combo([(s, i) for i, s in enumerate(SPEED_LEVELS)]) + self.cb_speed.setToolTip("Encoder preset. Slower presets give smaller files at the same quality.") + f.addRow("Speed:", self.cb_speed) + return box + + def _build_audio_group(self) -> QGroupBox: + self.audio_box = box = QGroupBox("Audio") + f = QFormLayout(box) + self.cb_audio_mode = _combo([(v, k) for k, v in AUDIO_MODES.items()]) + f.addRow("Mode:", self.cb_audio_mode) + self.cb_acodec = QComboBox() + f.addRow("Encoder:", self.cb_acodec) + self.cb_abitrate = _combo([("Auto", 0)] + [(f"{b} kb/s", b) for b in AUDIO_BITRATES if b]) + f.addRow("Bitrate:", self.cb_abitrate) + self.cb_channels = _combo(AUDIO_CHANNELS) + self.cb_channels.setToolTip("Up/down-mix the channel layout (e.g. 5.1 to stereo).") + f.addRow("Channels:", self.cb_channels) + self.cb_samplerate = _combo([("Original", 0)] + [(f"{r / 1000:g} kHz", r) for r in AUDIO_SAMPLE_RATES if r]) + self.cb_samplerate.setToolTip("Resample audio to a different sample rate.") + f.addRow("Sample rate:", self.cb_samplerate) + note = QLabel("Auto copies tracks the container supports and re-encodes the rest " + "(e.g. DTS/TrueHD -> AAC for MP4).") + note.setWordWrap(True) + note.setStyleSheet("color: gray;") + f.addRow(note) + return box + + def _connect(self) -> None: + self.act_add.triggered.connect(self._add_files_dialog) + self.act_add_dir.triggered.connect(self._add_folder_dialog) + self.act_remove.triggered.connect(self._remove_selected) + self.act_clear.triggered.connect(self._clear_queue) + self.table.selectionModel().currentRowChanged.connect(lambda *_: self._show_tracks()) + self.table.customContextMenuRequested.connect(self._table_menu) + self.model.rowsInserted.connect(self._update_hint) + self.model.rowsRemoved.connect(self._update_hint) + + self.track_list.itemChanged.connect(self._track_item_changed) + self.track_list.order_changed.connect(self._track_order_changed) + self.btn_all.clicked.connect(lambda: self._check_all_tracks(True)) + self.btn_none.clicked.connect(lambda: self._check_all_tracks(False)) + self.btn_apply_all.clicked.connect(self._apply_tracks_to_all) + + self.cb_container.currentIndexChanged.connect(self._container_changed) + self.rb_same_dir.toggled.connect(self._update_enabled) + self.btn_out_dir.clicked.connect(self._browse_out_dir) + self.cb_video_mode.currentIndexChanged.connect(self._update_enabled) + self.cb_audio_mode.currentIndexChanged.connect(self._update_enabled) + self.cb_resolution.currentIndexChanged.connect(self._update_enabled) + self.cb_vcodec.currentIndexChanged.connect(self._video_encoder_changed) + self.cb_acodec.currentIndexChanged.connect(self._update_enabled) + + self.btn_start.clicked.connect(self._start) + self.btn_stop.clicked.connect(self.runner.cancel) + self.btn_preview.clicked.connect(self._preview_command) + self.runner.job_changed.connect(self.model.refresh) + self.runner.log.connect(self._log) + self.runner.progress.connect(lambda v: self.overall.setValue(int(v * 1000))) + self.runner.finished.connect(self._batch_finished) + + # ------------------------------------------------------------ settings + def _load_settings(self) -> None: + geo = self.settings.value("geometry") + if geo is not None: + self.restoreGeometry(geo) + try: + opts = Options.from_dict(json.loads(self.settings.value("options", "{}"))) + except (TypeError, ValueError): + opts = Options() + self._apply_options(opts) + + def _save_settings(self) -> None: + self.settings.setValue("geometry", self.saveGeometry()) + self.settings.setValue("options", json.dumps(self.current_options().to_dict())) + + def _apply_options(self, o: Options) -> None: + self._loading = True + try: + _select_data(self.cb_container, o.container) + self._populate_encoders() + self.rb_custom_dir.setChecked(bool(o.output_dir)) + self.rb_same_dir.setChecked(not o.output_dir) + self.ed_out_dir.setText(o.output_dir) + self.ed_suffix.setText(o.suffix) + _select_data(self.cb_exists, o.if_exists) + self.chk_subs.setChecked(o.subtitles) + + _select_data(self.cb_video_mode, o.video_mode) + _select_data(self.cb_vcodec, o.video_encoder) + _select_data(self.cb_resolution, o.resolution) + self.sp_width.setValue(o.custom_width) + self.sp_height.setValue(o.custom_height) + self.chk_no_upscale.setChecked(o.no_upscale) + _select_data(self.cb_fps, o.fps) + self._update_quality_range(reset=False) + self.sp_quality.setValue(o.quality) + _select_data(self.cb_speed, o.speed) + + _select_data(self.cb_audio_mode, o.audio_mode) + _select_data(self.cb_acodec, o.audio_encoder) + _select_data(self.cb_abitrate, o.audio_bitrate) + _select_data(self.cb_channels, o.audio_channels) + _select_data(self.cb_samplerate, o.audio_sample_rate) + finally: + self._loading = False + self._update_enabled() + + def current_options(self) -> Options: + return Options( + container=self.cb_container.currentData(), + output_dir=self.ed_out_dir.text().strip() if self.rb_custom_dir.isChecked() else "", + suffix=self.ed_suffix.text(), + if_exists=self.cb_exists.currentData(), + video_mode=self.cb_video_mode.currentData(), + video_encoder=self.cb_vcodec.currentData() or "libx264", + resolution=self.cb_resolution.currentData(), + custom_width=self.sp_width.value(), + custom_height=self.sp_height.value(), + no_upscale=self.chk_no_upscale.isChecked(), + fps=self.cb_fps.currentData(), + quality=self.sp_quality.value(), + speed=self.cb_speed.currentData(), + audio_mode=self.cb_audio_mode.currentData(), + audio_encoder=self.cb_acodec.currentData() or "aac", + audio_bitrate=self.cb_abitrate.currentData(), + audio_channels=self.cb_channels.currentData(), + audio_sample_rate=self.cb_samplerate.currentData(), + subtitles=self.chk_subs.isChecked(), + ) + + def _reset_options(self) -> None: + self._apply_options(Options()) + + # ------------------------------------------------------------- ffmpeg + def _init_tools(self) -> None: + self.tools = locate_tools(self.settings.value("ffmpeg_dir", "") or None) + if self.tools: + self.ffmpeg_label.setText(f"FFmpeg {self.tools.version}") + self.ffmpeg_label.setToolTip(f"{self.tools.ffmpeg}\n{self.tools.ffprobe}") + self._log(f"Using ffmpeg {self.tools.version}: {self.tools.ffmpeg}") + else: + self.ffmpeg_label.setText("FFmpeg not found - Settings > Set FFmpeg Location") + self.ffmpeg_label.setToolTip("") + self._log("FFmpeg was not found. Put ffmpeg and ffprobe next to this program, install them " + "on your PATH, or use Settings > Set FFmpeg Location.") + self._populate_encoders() + # Re-probe anything that was waiting for ffprobe. + for job in self.model.jobs: + if job.status in (Status.PROBING, Status.INVALID) and job.info is None: + self._probe(job) + + def _choose_ffmpeg_dir(self) -> None: + d = QFileDialog.getExistingDirectory(self, "Folder containing ffmpeg and ffprobe", + self.settings.value("ffmpeg_dir", "")) + if not d: + return + found = locate_tools(d) + if found is None or Path(found.ffmpeg).parent != Path(d): + QMessageBox.warning(self, APP_NAME, "ffmpeg and ffprobe were not found in that folder.") + return + self.settings.setValue("ffmpeg_dir", d) + self._init_tools() + + def _reset_ffmpeg_dir(self) -> None: + self.settings.remove("ffmpeg_dir") + self._init_tools() + + # ---------------------------------------------------- dynamic controls + def _populate_encoders(self) -> None: + """Fill encoder combos with encoders that are available and fit the container.""" + container = CONTAINERS[self.cb_container.currentData()] + available = self.tools.encoders if self.tools else None + vcodec_changed = False + + def fits(enc, accepted): + return (accepted is None or enc.codec in accepted) and (available is None or enc.name in available) + + for cb, encoders, accepted in ( + (self.cb_vcodec, VIDEO_ENCODERS, container.video_codecs), + (self.cb_acodec, AUDIO_ENCODERS, container.audio_codecs), + ): + current = cb.currentData() + cb.blockSignals(True) + cb.clear() + for enc in encoders: + if fits(enc, accepted): + cb.addItem(enc.label, enc.name) + if not _select_data(cb, current): + cb.setCurrentIndex(0) + cb.blockSignals(False) + if cb is self.cb_vcodec: + vcodec_changed = cb.currentData() != current + self._update_quality_range(reset=vcodec_changed and not self._loading) + + def _container_changed(self) -> None: + if self._loading: + return + self._populate_encoders() + self._update_enabled() + + def _video_encoder_changed(self) -> None: + if not self._loading: + self._update_quality_range(reset=True) + self._update_enabled() + + def _update_quality_range(self, reset: bool) -> None: + enc = VIDEO_ENCODERS_BY_NAME.get(self.cb_vcodec.currentData()) + if enc is None: + return + self.sp_quality.setRange(*enc.quality_range) + if reset: + self.sp_quality.setValue(enc.quality_default) + + def _update_enabled(self) -> None: + if self._loading: + return + running = self.runner.running + self.settings_box.setEnabled(not running) + + self.ed_out_dir.setEnabled(self.rb_custom_dir.isChecked()) + self.btn_out_dir.setEnabled(self.rb_custom_dir.isChecked()) + + container = CONTAINERS[self.cb_container.currentData()] + subs_possible = container.text_subtitles is not None or container.image_subtitles != frozenset() + self.chk_subs.setEnabled(subs_possible) + + vencode = self.cb_video_mode.currentData() != "copy" + for w in (self.cb_vcodec, self.cb_resolution, self.chk_no_upscale, self.cb_fps, self.sp_quality): + w.setEnabled(vencode) + enc = VIDEO_ENCODERS_BY_NAME.get(self.cb_vcodec.currentData()) + self.cb_speed.setEnabled(vencode and enc is not None and enc.speed_option is not None) + custom = self.cb_resolution.currentData() == CUSTOM_RESOLUTION + self.custom_row.setVisible(custom) + if self.custom_row_label: + self.custom_row_label.setVisible(custom) + self.custom_row.setEnabled(vencode) + + amode = self.cb_audio_mode.currentData() + aencode = amode in ("auto", "encode") + for w in (self.cb_acodec, self.cb_channels, self.cb_samplerate): + w.setEnabled(aencode) + aenc = AUDIO_ENCODERS_BY_NAME.get(self.cb_acodec.currentData()) + self.cb_abitrate.setEnabled(aencode and aenc is not None and not aenc.lossless) + self.track_list.setEnabled(amode != "strip") + for b in (self.btn_all, self.btn_none, self.btn_apply_all): + b.setEnabled(amode != "strip") + + def _browse_out_dir(self) -> None: + d = QFileDialog.getExistingDirectory(self, "Output folder", self.ed_out_dir.text() or str(Path.home())) + if d: + self.ed_out_dir.setText(d) + + # --------------------------------------------------------- adding files + def _add_files_dialog(self) -> None: + exts = " ".join(f"*{e}" for e in sorted(MEDIA_EXTENSIONS)) + files, _ = QFileDialog.getOpenFileNames( + self, "Add media files", self.settings.value("last_dir", str(Path.home())), + f"Media files ({exts});;All files (*)") + if files: + self.settings.setValue("last_dir", str(Path(files[0]).parent)) + self.add_paths([Path(f) for f in files]) + + def _add_folder_dialog(self) -> None: + d = QFileDialog.getExistingDirectory(self, "Add all media files in folder (including sub-folders)", + self.settings.value("last_dir", str(Path.home()))) + if d: + self.settings.setValue("last_dir", d) + self.add_paths([Path(d)]) + + def add_paths(self, paths: list[Path]) -> None: + files: list[Path] = [] + for p in paths: + if p.is_dir(): + files += sorted((f for f in p.rglob("*") if f.is_file() and f.suffix.lower() in MEDIA_EXTENSIONS), + key=lambda f: str(f).lower()) + elif p.is_file(): + files.append(p) + existing = {os.path.normcase(str(j.path.resolve())) for j in self.model.jobs} + new_jobs = [] + for f in files: + key = os.path.normcase(str(f.resolve())) + if key in existing or ".part." in f.name: + continue + existing.add(key) + new_jobs.append(Job(path=f.resolve())) + self.model.add(new_jobs) + for job in new_jobs: + self._probe(job) + if new_jobs and not self.table.currentIndex().isValid(): + self.table.selectRow(0) + if paths and not new_jobs: + self.statusBar().showMessage("No new media files found", 4000) + + def _probe(self, job: Job) -> None: + if not self.tools: + job.status, job.message = Status.INVALID, "ffprobe not found" + self.model.refresh(job) + return + job.status, job.message = Status.PROBING, "" + self.model.refresh(job) + self.pool.start(_ProbeTask(self.tools.ffprobe, job, self.probe_signals)) + + def _on_probed(self, job: Job, info: MediaInfo | None, error: str) -> None: + if self.model.row_of(job) < 0: + return # removed while probing + if info is None: + job.status, job.message = Status.INVALID, error + self._log(f"Cannot read {job.path.name}: {error}") + else: + job.info = info + job.audio_order = [s.index for s in info.audio] + job.audio_checked = set(job.audio_order) + job.status, job.message = Status.READY, "" + self.model.refresh(job) + if job is self._current_job(): + self._show_tracks() + + def dragEnterEvent(self, event): + if event.mimeData().hasUrls(): + event.acceptProposedAction() + + def dropEvent(self, event): + paths = [Path(u.toLocalFile()) for u in event.mimeData().urls() if u.isLocalFile()] + self.add_paths(paths) + + def _update_hint(self, *_): + self.drop_hint.setVisible(not self.model.jobs) + + # ------------------------------------------------------- queue actions + def _selected_rows(self) -> list[int]: + return sorted({i.row() for i in self.table.selectionModel().selectedRows()}) + + def _current_job(self) -> Job | None: + idx = self.table.currentIndex() + return self.model.job_at(idx.row()) if idx.isValid() else None + + def _remove_selected(self) -> None: + rows = [r for r in self._selected_rows() if self.model.jobs[r].status != Status.RUNNING + and self.model.jobs[r].status != Status.QUEUED] + self.model.remove_rows(rows) + self._show_tracks() + + def _clear_queue(self) -> None: + if self.runner.running: + QMessageBox.information(self, APP_NAME, "Stop the current batch before clearing the queue.") + return + self.model.remove_rows(list(range(len(self.model.jobs)))) + self.overall.setValue(0) + self._show_tracks() + + def _table_menu(self, pos) -> None: + job = self.model.job_at(self.table.indexAt(pos).row()) + if job is None: + return + menu = QMenu(self) + target = job.output if job.output and job.output.exists() else job.path + menu.addAction("Open Containing Folder").triggered.connect( + lambda: QDesktopServices.openUrl(QUrl.fromLocalFile(str(target.parent)))) + if job.status == Status.DONE: + menu.addAction("Convert Again").triggered.connect(lambda: self._requeue(job)) + if job.status == Status.INVALID: + menu.addAction("Retry Reading File").triggered.connect(lambda: self._probe(job)) + menu.addSeparator() + menu.addAction(self.act_remove) + menu.exec(self.table.viewport().mapToGlobal(pos)) + + def _requeue(self, job: Job) -> None: + job.status, job.progress, job.message = Status.READY, 0.0, "" + self.model.refresh(job) + + # ---------------------------------------------------------- track panel + def _show_tracks(self) -> None: + job = self._current_job() + self._track_job = job + self.track_list.blockSignals(True) + self.track_list.clear() + if job is None or job.info is None: + self.tracks_title.setText("Select a file in the queue" if job is None + else f"{job.path.name}\n({job.status.value})") + self.stream_info.setText("") + self.track_list.blockSignals(False) + return + info = job.info + self.tracks_title.setText(f"{job.path.name}") + by_index = {s.index: s for s in info.audio} + for n, idx in enumerate(job.audio_order): + s = by_index[idx] + pos = info.audio.index(s) + 1 + text = f"Track {pos}: {(s.language or 'und').upper():4} {s.describe()}" + if s.title: + text += f' "{s.title}"' + if s.default: + text += " (default)" + item = QListWidgetItem(text) + item.setData(Qt.UserRole, idx) + item.setFlags(item.flags() | Qt.ItemIsUserCheckable | Qt.ItemIsDragEnabled) + item.setCheckState(Qt.Checked if idx in job.audio_checked else Qt.Unchecked) + self.track_list.addItem(item) + if not info.audio: + self.track_list.addItem(QListWidgetItem("(no audio tracks)")) + + v = info.video + text_subs = sum(1 for s in info.subtitles if s.codec in TEXT_SUBTITLE_CODECS) + image_subs = len(info.subtitles) - text_subs + parts = [f"Video: {v.describe()}" if v else "Video: none"] + if info.subtitles: + parts.append(f"Subtitles: {text_subs} text, {image_subs} image") + self.stream_info.setText(" | ".join(parts)) + self.track_list.blockSignals(False) + + def _sync_tracks_from_list(self) -> None: + job = self._track_job + if job is None or job.info is None: + return + order, checked = [], set() + for i in range(self.track_list.count()): + item = self.track_list.item(i) + idx = item.data(Qt.UserRole) + if idx is None: + continue + order.append(idx) + if item.checkState() == Qt.Checked: + checked.add(idx) + job.audio_order, job.audio_checked = order, checked + self.model.refresh(job) + + def _track_item_changed(self, _item) -> None: + self._sync_tracks_from_list() + + def _track_order_changed(self) -> None: + self._sync_tracks_from_list() + + def _check_all_tracks(self, checked: bool) -> None: + self.track_list.blockSignals(True) + for i in range(self.track_list.count()): + item = self.track_list.item(i) + if item.data(Qt.UserRole) is not None: + item.setCheckState(Qt.Checked if checked else Qt.Unchecked) + self.track_list.blockSignals(False) + self._sync_tracks_from_list() + + def _apply_tracks_to_all(self) -> None: + ref = self._track_job + if ref is None or ref.info is None: + return + changed = 0 + for job in self.model.jobs: + if job is ref or job.info is None or job.status in (Status.RUNNING, Status.QUEUED): + continue + if not ref.selected_audio: + job.audio_checked = set() + else: + job.set_audio_selection(match_audio_selection(ref.info.audio, ref.selected_audio, job.info.audio)) + self.model.refresh(job) + changed += 1 + self.statusBar().showMessage(f"Track selection applied to {changed} other file(s)", 5000) + + # ------------------------------------------------------------- running + def _check_ready(self, opts: Options) -> bool: + if not self.tools: + QMessageBox.warning(self, APP_NAME, "FFmpeg was not found. Use Settings > Set FFmpeg Location.") + return False + if self.rb_custom_dir.isChecked(): + if not opts.output_dir: + QMessageBox.warning(self, APP_NAME, "Choose an output folder, or select 'Same folder as source'.") + return False + try: + Path(opts.output_dir).mkdir(parents=True, exist_ok=True) + except OSError as e: + QMessageBox.warning(self, APP_NAME, f"Cannot use output folder:\n{e}") + return False + if opts.video_mode != "copy" and opts.resolution == CUSTOM_RESOLUTION \ + and not (opts.custom_width or opts.custom_height): + QMessageBox.warning(self, APP_NAME, "Enter a custom width and/or height.") + return False + return True + + def _start(self) -> None: + if self.runner.running: + return + opts = self.current_options() + if not self._check_ready(opts): + return + jobs = [j for j in self.model.jobs if j.status in RUNNABLE and j.info is not None] + if not jobs: + busy = any(j.status == Status.PROBING for j in self.model.jobs) + QMessageBox.information(self, APP_NAME, "Files are still being read - try again in a moment." + if busy else "Nothing to convert. Add some files first.") + return + no_audio = [j for j in jobs if j.info.audio and not j.selected_audio and opts.audio_mode != "strip"] + if no_audio: + names = "\n".join(j.path.name for j in no_audio[:10]) + if QMessageBox.question(self, APP_NAME, f"These files have no audio track ticked and will be " + f"converted without audio:\n\n{names}\n\nContinue?") != QMessageBox.Yes: + return + self._save_settings() + self.overall.setValue(0) + self._set_running(True) + self.runner.start(jobs, opts, self.tools) + + def _set_running(self, running: bool) -> None: + self.btn_start.setEnabled(not running) + self.btn_stop.setEnabled(running) + self.act_clear.setEnabled(not running) + self.settings_box.setEnabled(not running) + if not running: + self._update_enabled() + + def _batch_finished(self, counts: dict) -> None: + self._set_running(False) + summary = ", ".join(f"{n} {k.lower()}" for k, n in counts.items()) + self._log(f"Batch finished: {summary or 'nothing to do'}") + self.statusBar().showMessage(f"Finished: {summary}", 10000) + if counts.get(Status.FAILED.value) or counts.get(Status.CANCELLED.value): + return + self.overall.setValue(1000) + + def _preview_command(self) -> None: + job = self._current_job() + if job is None or job.info is None: + self._log("Select a file that has been read successfully to preview its command.") + return + if not self.tools: + self._log("FFmpeg not found.") + return + opts = self.current_options() + out = plan_output_path(job.path, opts, set()) + if out is None: + self._log(f"{job.path.name}: output already exists and would be skipped.") + return + built = build_command(self.tools.ffmpeg, job.info, opts, job.selected_audio, out) + # Drop the progress plumbing so the command can be pasted into a terminal. + args = list(built.args) + for flag in ("-progress", "-nostats", "-nostdin"): + if flag in args: + i = args.index(flag) + del args[i:i + (2 if flag == "-progress" else 1)] + self._log("") + self._log(f"--- Command preview for {job.path.name}") + for note in built.notes: + self._log(f" {note}") + self._log(format_command(args)) + + def _log(self, text: str) -> None: + self.log_view.appendPlainText(text) + + def _about(self) -> None: + ff = f"FFmpeg {self.tools.version}\n{self.tools.ffmpeg}" if self.tools else "FFmpeg not found" + QMessageBox.about(self, APP_NAME, f"{APP_NAME} {__version__}\n\nA simple front-end for ffmpeg.\n\n{ff}") + + def closeEvent(self, event): + if self.runner.running: + if QMessageBox.question(self, APP_NAME, "A conversion is running. Stop it and quit?") != QMessageBox.Yes: + event.ignore() + return + self.runner.kill_now() + self._save_settings() + self.pool.clear() + event.accept() diff --git a/ffwrap/gui/queue_model.py b/ffwrap/gui/queue_model.py new file mode 100644 index 0000000..c887e64 --- /dev/null +++ b/ffwrap/gui/queue_model.py @@ -0,0 +1,160 @@ +"""The file queue: Job objects plus a Qt table model / progress delegate.""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from pathlib import Path + +from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt +from PySide6.QtWidgets import QApplication, QStyle, QStyledItemDelegate, QStyleOptionProgressBar + +from ..probe import MediaInfo + + +class Status(enum.Enum): + PROBING = "Reading..." + READY = "Ready" + INVALID = "Unreadable" + QUEUED = "Queued" + RUNNING = "Converting" + DONE = "Done" + FAILED = "Failed" + SKIPPED = "Skipped" + CANCELLED = "Cancelled" + + +# Jobs in these states are (re)processed when Start is pressed. +RUNNABLE = {Status.READY, Status.FAILED, Status.CANCELLED, Status.SKIPPED} + + +@dataclass(eq=False) +class Job: + path: Path + info: MediaInfo | None = None + status: Status = Status.PROBING + progress: float = 0.0 + message: str = "" + audio_order: list[int] = field(default_factory=list) + audio_checked: set[int] = field(default_factory=set) + output: Path | None = None + + @property + def selected_audio(self) -> list[int]: + return [i for i in self.audio_order if i in self.audio_checked] + + def set_audio_selection(self, indices: list[int]) -> None: + """Check ``indices`` and move them (in order) to the top of the list.""" + rest = [i for i in self.audio_order if i not in indices] + self.audio_order = list(indices) + rest + self.audio_checked = set(indices) + + +def fmt_duration(seconds: float | None) -> str: + if not seconds: + return "" + s = int(round(seconds)) + return f"{s // 3600}:{s % 3600 // 60:02d}:{s % 60:02d}" + + +COLUMNS = ["File", "Duration", "Video", "Audio", "Subs", "Status", "Progress"] +COL_STATUS, COL_PROGRESS = 5, 6 + + +class QueueModel(QAbstractTableModel): + def __init__(self, parent=None): + super().__init__(parent) + self.jobs: list[Job] = [] + + # -- Qt model API ------------------------------------------------------- + def rowCount(self, parent=QModelIndex()): + return 0 if parent.isValid() else len(self.jobs) + + def columnCount(self, parent=QModelIndex()): + return 0 if parent.isValid() else len(COLUMNS) + + def headerData(self, section, orientation, role=Qt.DisplayRole): + if orientation == Qt.Horizontal and role == Qt.DisplayRole: + return COLUMNS[section] + return None + + def data(self, index, role=Qt.DisplayRole): + if not index.isValid(): + return None + job = self.jobs[index.row()] + col = index.column() + info = job.info + if role == Qt.DisplayRole: + if col == 0: + return job.path.name + if col == 1: + return fmt_duration(info.duration) if info else "" + if col == 2: + return info.video.describe() if info and info.video else ("none" if info else "") + if col == 3: + if not info: + return "" + return f"{len(job.audio_checked)} of {len(info.audio)}" if info.audio else "none" + if col == 4: + return str(len(info.subtitles)) if info else "" + if col == COL_STATUS: + return job.status.value + (f": {job.message}" if job.message else "") + if col == COL_PROGRESS: + return None if job.status in (Status.PROBING, Status.INVALID) else job.progress + elif role == Qt.ToolTipRole: + if col == 0: + return str(job.path) + if col == COL_STATUS: + tip = job.status.value + (f"\n{job.message}" if job.message else "") + if job.output: + tip += f"\nOutput: {job.output}" + return tip + elif role == Qt.UserRole: + return job + return None + + # -- helpers ------------------------------------------------------------ + def add(self, jobs: list[Job]) -> None: + if not jobs: + return + start = len(self.jobs) + self.beginInsertRows(QModelIndex(), start, start + len(jobs) - 1) + self.jobs.extend(jobs) + self.endInsertRows() + + def remove_rows(self, rows: list[int]) -> None: + for row in sorted(set(rows), reverse=True): + self.beginRemoveRows(QModelIndex(), row, row) + del self.jobs[row] + self.endRemoveRows() + + def row_of(self, job: Job) -> int: + for i, j in enumerate(self.jobs): + if j is job: + return i + return -1 + + def refresh(self, job: Job) -> None: + row = self.row_of(job) + if row >= 0: + self.dataChanged.emit(self.index(row, 0), self.index(row, len(COLUMNS) - 1)) + + def job_at(self, row: int) -> Job | None: + return self.jobs[row] if 0 <= row < len(self.jobs) else None + + +class ProgressDelegate(QStyledItemDelegate): + def paint(self, painter, option, index): + value = index.data(Qt.DisplayRole) + if value is None: + super().paint(painter, option, index) + return + bar = QStyleOptionProgressBar() + bar.rect = option.rect.adjusted(2, 3, -2, -3) + bar.minimum, bar.maximum = 0, 1000 + bar.progress = int(value * 1000) + bar.text = f"{value * 100:.0f}%" + bar.textVisible = True + bar.state = option.state + style = option.widget.style() if option.widget else QApplication.style() + style.drawControl(QStyle.CE_ProgressBar, bar, painter, option.widget) diff --git a/ffwrap/gui/runner.py b/ffwrap/gui/runner.py new file mode 100644 index 0000000..552eb6d --- /dev/null +++ b/ffwrap/gui/runner.py @@ -0,0 +1,219 @@ +"""Runs ffmpeg for each queued job, one at a time, via QProcess.""" + +from __future__ import annotations + +import os +from pathlib import Path + +from PySide6.QtCore import QObject, QProcess, Signal + +from ..command import Options, build_command, format_command, plan_output_path, temp_output_path +from ..tools import FFmpegTools +from .queue_model import Job, Status + + +def _remove_quietly(path: Path | None) -> None: + if path is None: + return + try: + path.unlink() + except OSError: + pass + + +class BatchRunner(QObject): + job_changed = Signal(object) # Job + log = Signal(str) + progress = Signal(float) # overall 0..1 + finished = Signal(dict) # counts per status + + def __init__(self, parent=None): + super().__init__(parent) + self._proc: QProcess | None = None + self._jobs: list[Job] = [] + self._pos = 0 + self._opts = Options() + self._tools: FFmpegTools | None = None + self._taken: set[str] = set() + self._current: Job | None = None + self._temp: Path | None = None + self._stdout_buf = b"" + self._stderr_buf = b"" + self._stderr_tail: list[str] = [] + self._cancelled = False + + @property + def running(self) -> bool: + return self._current is not None + + # -- public API --------------------------------------------------------- + def start(self, jobs: list[Job], opts: Options, tools: FFmpegTools) -> None: + self._jobs, self._opts, self._tools = jobs, opts, tools + self._pos, self._taken, self._cancelled = 0, set(), False + for job in jobs: + job.status, job.progress, job.message, job.output = Status.QUEUED, 0.0, "", None + self.job_changed.emit(job) + self._next() + + def cancel(self) -> None: + self._cancelled = True + if self._proc is not None and self._proc.state() != QProcess.NotRunning: + self.log.emit("Stopping...") + self._proc.kill() + elif self._current is None: + self._finish_batch() + + def kill_now(self) -> None: + """Used on application exit: stop ffmpeg and remove the partial file.""" + self._cancelled = True + if self._proc is not None and self._proc.state() != QProcess.NotRunning: + self._proc.kill() + self._proc.waitForFinished(3000) + _remove_quietly(self._temp) + + # -- internals ---------------------------------------------------------- + def _emit_overall(self, current_fraction: float = 0.0) -> None: + if self._jobs: + self.progress.emit(min(1.0, (self._pos + current_fraction) / len(self._jobs))) + + def _next(self) -> None: + while self._pos < len(self._jobs): + if self._cancelled: + break + job = self._jobs[self._pos] + if self._launch(job): + return + self._pos += 1 + self._emit_overall() + self._finish_batch() + + def _launch(self, job: Job) -> bool: + """Start ffmpeg for ``job``. Returns False if the job finished immediately.""" + opts = self._opts + out = plan_output_path(job.path, opts, self._taken) + if out is None: + job.status, job.message = Status.SKIPPED, "output already exists" + self.log.emit(f"Skipping {job.path.name}: output already exists") + self.job_changed.emit(job) + return False + try: + out.parent.mkdir(parents=True, exist_ok=True) + except OSError as e: + job.status, job.message = Status.FAILED, f"cannot create output folder: {e}" + self.log.emit(f"ERROR {job.path.name}: {job.message}") + self.job_changed.emit(job) + return False + + self._temp = temp_output_path(out) + built = build_command(self._tools.ffmpeg, job.info, opts, job.selected_audio, self._temp) + job.output = out + self._taken.add(os.path.normcase(os.path.abspath(out))) + + self.log.emit("") + self.log.emit(f"=== [{self._pos + 1}/{len(self._jobs)}] {job.path}") + self.log.emit(f" -> {out}") + for note in built.notes: + self.log.emit(f" {note}") + self.log.emit(format_command(built.args)) + + self._current = job + self._stdout_buf = self._stderr_buf = b"" + self._stderr_tail = [] + job.status, job.progress, job.message = Status.RUNNING, 0.0, "" + self.job_changed.emit(job) + + proc = QProcess(self) + proc.readyReadStandardOutput.connect(self._on_stdout) + proc.readyReadStandardError.connect(self._on_stderr) + proc.finished.connect(self._on_finished) + proc.errorOccurred.connect(self._on_error) + self._proc = proc + proc.start(built.args[0], built.args[1:]) + return True + + def _on_stdout(self) -> None: + self._stdout_buf += bytes(self._proc.readAllStandardOutput()) + *lines, self._stdout_buf = self._stdout_buf.split(b"\n") + job = self._current + if job is None: + return + duration = job.info.duration if job.info else None + speed = None + updated = False + for raw in lines: + key, _, value = raw.decode(errors="replace").strip().partition("=") + if key == "out_time_us" and duration: + try: + job.progress = max(0.0, min(1.0, int(value) / 1e6 / duration)) + updated = True + except ValueError: + pass + elif key == "speed" and value not in ("N/A", ""): + speed = value + if updated: + job.message = f"{job.progress * 100:.0f}%" + (f" @ {speed}" if speed else "") + self.job_changed.emit(job) + self._emit_overall(job.progress) + + def _on_stderr(self) -> None: + self._stderr_buf += bytes(self._proc.readAllStandardError()) + *lines, self._stderr_buf = self._stderr_buf.split(b"\n") + for raw in lines: + line = raw.decode(errors="replace").rstrip() + if line: + self._stderr_tail = (self._stderr_tail + [line])[-5:] + self.log.emit(f" ffmpeg: {line}") + + def _on_error(self, error) -> None: + if error == QProcess.FailedToStart: + # finished() is not emitted in this case. + self._complete(ok=False, reason=f"could not start ffmpeg ({self._proc.errorString()})") + + def _on_finished(self, exit_code: int, exit_status) -> None: + self._on_stderr() + ok = exit_status == QProcess.NormalExit and exit_code == 0 + reason = "" if ok else (self._stderr_tail[-1] if self._stderr_tail else f"ffmpeg exited with code {exit_code}") + self._complete(ok, reason) + + def _complete(self, ok: bool, reason: str) -> None: + job = self._current + if job is None: + return + self._current = None + if self._proc is not None: + self._proc.deleteLater() + self._proc = None + + if self._cancelled: + job.status, job.message, job.progress = Status.CANCELLED, "", 0.0 + _remove_quietly(self._temp) + self.log.emit(f"Cancelled {job.path.name}") + elif ok: + try: + os.replace(self._temp, job.output) + job.status, job.progress, job.message = Status.DONE, 1.0, "" + self.log.emit(f"Done: {job.output}") + except OSError as e: + job.status, job.message = Status.FAILED, f"could not rename output: {e}" + _remove_quietly(self._temp) + self.log.emit(f"ERROR {job.path.name}: {job.message}") + else: + job.status, job.message = Status.FAILED, reason + _remove_quietly(self._temp) + self.log.emit(f"FAILED {job.path.name}: {reason}") + self._temp = None + self.job_changed.emit(job) + + self._pos += 1 + self._emit_overall() + self._next() + + def _finish_batch(self) -> None: + counts: dict[str, int] = {} + for job in self._jobs: + if job.status == Status.QUEUED: + job.status = Status.READY + self.job_changed.emit(job) + counts[job.status.value] = counts.get(job.status.value, 0) + 1 + self._jobs = [] + self.finished.emit(counts) diff --git a/ffwrap/probe.py b/ffwrap/probe.py new file mode 100644 index 0000000..75ce255 --- /dev/null +++ b/ffwrap/probe.py @@ -0,0 +1,184 @@ +"""ffprobe wrapper producing simple dataclasses.""" + +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + +from .formats import describe_channels +from .tools import run_quiet + + +class ProbeError(Exception): + pass + + +@dataclass +class Stream: + index: int # absolute stream index inside the file + kind: str # "video" | "audio" | "subtitle" | "data" | "attachment" + codec: str | None + profile: str | None = None + language: str | None = None + title: str | None = None + default: bool = False + forced: bool = False + attached_pic: bool = False + # video + width: int | None = None + height: int | None = None + fps: float | None = None + rotation: int = 0 + pix_fmt: str | None = None + # audio + channels: int | None = None + channel_layout: str | None = None + sample_rate: int | None = None + bit_rate: int | None = None + + @property + def display_size(self) -> tuple[int, int] | None: + """Width/height after applying rotation metadata (what players show).""" + if not self.width or not self.height: + return None + if self.rotation % 180 == 90: + return self.height, self.width + return self.width, self.height + + def describe(self) -> str: + if self.kind == "audio": + parts = [(self.codec or "?").upper(), describe_channels(self.channels, self.channel_layout)] + if self.sample_rate: + parts.append(f"{self.sample_rate / 1000:g} kHz") + if self.bit_rate: + parts.append(f"{self.bit_rate // 1000} kb/s") + return " ".join(parts) + if self.kind == "video": + size = self.display_size + s = (self.codec or "?").upper() + if size: + s += f" {size[0]}x{size[1]}" + if self.fps: + s += f" {self.fps:.3g}fps" + return s + return self.codec or "?" + + +@dataclass +class MediaInfo: + path: Path + duration: float | None + format_name: str + streams: list[Stream] = field(default_factory=list) + + @property + def video(self) -> Stream | None: + """The main video stream (cover art / thumbnails are ignored).""" + for s in self.streams: + if s.kind == "video" and not s.attached_pic: + return s + return None + + @property + def audio(self) -> list[Stream]: + return [s for s in self.streams if s.kind == "audio"] + + @property + def subtitles(self) -> list[Stream]: + return [s for s in self.streams if s.kind == "subtitle"] + + def stream(self, index: int) -> Stream | None: + for s in self.streams: + if s.index == index: + return s + return None + + +def _to_float(v) -> float | None: + try: + f = float(v) + except (TypeError, ValueError): + return None + return f if f == f else None # filter NaN + + +def _to_int(v) -> int | None: + f = _to_float(v) + return int(f) if f is not None else None + + +def _rate(v: str | None) -> float | None: + if not v or v in ("0/0", "0"): + return None + if "/" in v: + num, den = v.split("/", 1) + n, d = _to_float(num), _to_float(den) + return n / d if n and d else None + return _to_float(v) + + +def _rotation(raw: dict) -> int: + for sd in raw.get("side_data_list") or []: + if "rotation" in sd: + return int(_to_float(sd["rotation"]) or 0) % 360 + rot = (raw.get("tags") or {}).get("rotate") + return int(_to_float(rot) or 0) % 360 + + +def parse_probe(path: Path, data: dict) -> MediaInfo: + fmt = data.get("format") or {} + streams: list[Stream] = [] + for raw in data.get("streams") or []: + tags = {k.lower(): v for k, v in (raw.get("tags") or {}).items()} + disp = raw.get("disposition") or {} + lang = tags.get("language") + s = Stream( + index=int(raw.get("index", len(streams))), + kind=raw.get("codec_type") or "data", + codec=raw.get("codec_name"), + profile=raw.get("profile"), + language=lang if lang and lang != "und" else None, + title=tags.get("title"), + default=bool(disp.get("default")), + forced=bool(disp.get("forced")), + attached_pic=bool(disp.get("attached_pic")), + width=_to_int(raw.get("width")), + height=_to_int(raw.get("height")), + fps=_rate(raw.get("avg_frame_rate")) or _rate(raw.get("r_frame_rate")), + rotation=_rotation(raw), + pix_fmt=raw.get("pix_fmt"), + channels=_to_int(raw.get("channels")), + channel_layout=raw.get("channel_layout"), + sample_rate=_to_int(raw.get("sample_rate")), + bit_rate=_to_int(raw.get("bit_rate")) or _to_int(tags.get("bps")), + ) + streams.append(s) + + duration = _to_float(fmt.get("duration")) + if not duration: + durations = [_to_float(r.get("duration")) for r in data.get("streams") or []] + duration = max((d for d in durations if d), default=None) + return MediaInfo(path=path, duration=duration, format_name=fmt.get("format_name", ""), streams=streams) + + +def probe(ffprobe: str, path: Path) -> MediaInfo: + args = [ffprobe, "-v", "error", "-print_format", "json", "-show_format", "-show_streams", str(path)] + try: + res = run_quiet(args, timeout=120) + except (OSError, subprocess.SubprocessError) as e: + raise ProbeError(str(e)) from e + if res.returncode != 0: + lines = res.stderr.decode(errors="replace").strip().splitlines() + msg = lines[-1] if lines else f"ffprobe exited with code {res.returncode}" + # ffprobe prefixes errors with the file name, which the queue already shows. + raise ProbeError(msg.removeprefix(f"{path}: ")) + try: + data = json.loads(res.stdout.decode("utf-8", errors="replace")) + except json.JSONDecodeError as e: + raise ProbeError(f"Could not parse ffprobe output: {e}") from e + info = parse_probe(path, data) + if not info.video and not info.audio: + raise ProbeError("No video or audio streams found") + return info diff --git a/ffwrap/tools.py b/ffwrap/tools.py new file mode 100644 index 0000000..712d098 --- /dev/null +++ b/ffwrap/tools.py @@ -0,0 +1,102 @@ +"""Locating ffmpeg / ffprobe and running them without flashing console windows.""" + +from __future__ import annotations + +import os +import re +import shutil +import subprocess +import sys +from dataclasses import dataclass, field +from pathlib import Path + +IS_WINDOWS = sys.platform.startswith("win") +EXE_SUFFIX = ".exe" if IS_WINDOWS else "" + +# Keep ffprobe/ffmpeg helper calls from popping up a console on Windows. +_NO_WINDOW = {"creationflags": subprocess.CREATE_NO_WINDOW} if IS_WINDOWS else {} + + +def app_dir() -> Path: + """Directory holding the executable (frozen) or the project root (source).""" + if getattr(sys, "frozen", False): + return Path(sys.executable).resolve().parent + return Path(__file__).resolve().parent.parent + + +def run_quiet(args: list[str], timeout: float | None = 60) -> subprocess.CompletedProcess: + return subprocess.run( + args, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + **_NO_WINDOW, + ) + + +@dataclass +class FFmpegTools: + ffmpeg: str + ffprobe: str + version: str = "" + encoders: set[str] = field(default_factory=set) + + @property + def ok(self) -> bool: + return bool(self.ffmpeg and self.ffprobe) + + +def _candidate_dirs(configured_dir: str | None) -> list[Path]: + dirs: list[Path] = [] + if configured_dir: + dirs.append(Path(configured_dir)) + base = app_dir() + dirs += [base, base / "ffmpeg", base / "ffmpeg" / "bin", base / "bin"] + return dirs + + +def _find_in(dirs: list[Path], name: str) -> str | None: + for d in dirs: + p = d / f"{name}{EXE_SUFFIX}" + if p.is_file() and (IS_WINDOWS or os.access(p, os.X_OK)): + return str(p) + return None + + +def locate_tools(configured_dir: str | None = None) -> FFmpegTools | None: + """Find ffmpeg + ffprobe: configured dir, next to the app, then PATH.""" + dirs = _candidate_dirs(configured_dir) + ffmpeg = _find_in(dirs, "ffmpeg") + if ffmpeg: + # Prefer the ffprobe that ships alongside the chosen ffmpeg. + ffprobe = _find_in([Path(ffmpeg).parent], "ffprobe") or _find_in(dirs, "ffprobe") + else: + ffmpeg = shutil.which("ffmpeg") + ffprobe = None + if ffmpeg and not ffprobe: + ffprobe = _find_in([Path(ffmpeg).parent], "ffprobe") or shutil.which("ffprobe") + if not ffmpeg or not ffprobe: + return None + + tools = FFmpegTools(ffmpeg=ffmpeg, ffprobe=ffprobe) + try: + out = run_quiet([ffmpeg, "-hide_banner", "-version"], timeout=15).stdout.decode(errors="replace") + first = out.splitlines()[0] if out else "" + m = re.match(r"ffmpeg version (\S+)", first) + tools.version = m.group(1) if m else first + enc = run_quiet([ffmpeg, "-hide_banner", "-encoders"], timeout=15).stdout.decode(errors="replace") + tools.encoders = parse_encoders(enc) + except (OSError, subprocess.SubprocessError): + return None + return tools + + +def parse_encoders(text: str) -> set[str]: + names = set() + for line in text.splitlines(): + # " V....D libx264 libx264 H.264 ..." + m = re.match(r"^\s*([VAS])[\w.]{5}\s+(\S+)\s", line) + if m and m.group(2) != "=": + names.add(m.group(2)) + return names diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..5ee6477 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +testpaths = tests diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..6b04e09 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,3 @@ +-r requirements.txt +pyinstaller>=6.10 +pytest>=8 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..0fff071 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +# PySide6 6.12.0 over-releases Python's None (QComboBox.clear/currentData), which +# crashes the app on Python < 3.12. Stay below it until a fixed release ships. +PySide6-Essentials>=6.9,<6.12 diff --git a/run.py b/run.py new file mode 100644 index 0000000..b870a2d --- /dev/null +++ b/run.py @@ -0,0 +1,8 @@ +"""Launcher used for development (`python run.py`) and as the PyInstaller entry point.""" + +import sys + +from ffwrap.app import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/fetch_ffmpeg.py b/scripts/fetch_ffmpeg.py new file mode 100644 index 0000000..94afb54 --- /dev/null +++ b/scripts/fetch_ffmpeg.py @@ -0,0 +1,79 @@ +"""Download a static ffmpeg + ffprobe build for this OS into ./ffmpeg/bin. + +Uses the GPL builds from https://github.com/BtbN/FFmpeg-Builds (newest stable +release branch). The app finds ./ffmpeg/bin automatically when run from source, +and build.py --ffmpeg-dir ffmpeg/bin bundles them with the executable. +""" + +from __future__ import annotations + +import json +import re +import shutil +import sys +import tarfile +import tempfile +import urllib.request +import zipfile +from pathlib import Path + +DEST = Path(__file__).resolve().parent.parent / "ffmpeg" / "bin" +API = "https://api.github.com/repos/BtbN/FFmpeg-Builds/releases/latest" + + +def pick_asset(assets: list[dict], plat: str) -> dict: + ext = ".zip" if plat == "win64" else ".tar.xz" + stable = [] + for a in assets: + m = re.fullmatch(rf"ffmpeg-n(\d+)\.(\d+)-latest-{plat}-gpl-\d+\.\d+{re.escape(ext)}", a["name"]) + if m: + stable.append(((int(m.group(1)), int(m.group(2))), a)) + if stable: + return max(stable, key=lambda x: x[0])[1] + for a in assets: + if a["name"] == f"ffmpeg-master-latest-{plat}-gpl{ext}": + return a + raise SystemExit(f"No ffmpeg build found for {plat}") + + +def main() -> int: + if sys.platform.startswith("win"): + plat, exe = "win64", ".exe" + elif sys.platform.startswith("linux"): + plat, exe = "linux64", "" + else: + raise SystemExit("Only Windows and Linux are supported") + + with urllib.request.urlopen(API, timeout=30) as r: + asset = pick_asset(json.load(r)["assets"], plat) + print(f"Downloading {asset['name']} ({asset['size'] // 1_000_000} MB)...") + + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) / asset["name"] + with urllib.request.urlopen(asset["browser_download_url"], timeout=60) as r, open(archive, "wb") as f: + shutil.copyfileobj(r, f) + DEST.mkdir(parents=True, exist_ok=True) + wanted = {f"ffmpeg{exe}", f"ffprobe{exe}"} + if archive.suffix == ".zip": + with zipfile.ZipFile(archive) as z: + for info in z.infolist(): + name = Path(info.filename).name + if name in wanted and "/bin/" in info.filename: + (DEST / name).write_bytes(z.read(info)) + else: + with tarfile.open(archive) as t: + for m in t.getmembers(): + name = Path(m.name).name + if name in wanted and "/bin/" in m.name: + with t.extractfile(m) as src, open(DEST / name, "wb") as out: + shutil.copyfileobj(src, out) + (DEST / name).chmod(0o755) + missing = [n for n in wanted if not (DEST / n).is_file()] + if missing: + raise SystemExit(f"Archive did not contain {missing}") + print(f"Installed ffmpeg and ffprobe into {DEST}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/make_icon.py b/scripts/make_icon.py new file mode 100644 index 0000000..e06ffa5 --- /dev/null +++ b/scripts/make_icon.py @@ -0,0 +1,69 @@ +"""Render assets/icon.png and a multi-size assets/icon.ico. Run once; outputs are committed.""" + +import os +import struct +import sys +from pathlib import Path + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +from PySide6.QtCore import QBuffer, QPointF, QRectF, Qt +from PySide6.QtGui import QColor, QGuiApplication, QImage, QLinearGradient, QPainter, QPainterPath, QPolygonF + +ASSETS = Path(__file__).resolve().parent.parent / "assets" + + +def render(size: int) -> QImage: + img = QImage(size, size, QImage.Format_ARGB32) + img.fill(Qt.transparent) + p = QPainter(img) + p.setRenderHint(QPainter.Antialiasing) + s = size / 256 + grad = QLinearGradient(0, 0, 0, size) + grad.setColorAt(0, QColor("#2f80ed")) + grad.setColorAt(1, QColor("#1b4f9c")) + path = QPainterPath() + path.addRoundedRect(QRectF(8 * s, 8 * s, 240 * s, 240 * s), 52 * s, 52 * s) + p.fillPath(path, grad) + # Film strip holes along the top and bottom edges. + p.setBrush(QColor(255, 255, 255, 70)) + p.setPen(Qt.NoPen) + for i in range(6): + x = (38 + i * 34) * s + p.drawRoundedRect(QRectF(x, 22 * s, 18 * s, 14 * s), 3 * s, 3 * s) + p.drawRoundedRect(QRectF(x, 220 * s, 18 * s, 14 * s), 3 * s, 3 * s) + # Two chevrons: "convert from -> to". + p.setBrush(QColor("white")) + for dx in (0, 62): + p.drawPolygon(QPolygonF([QPointF((70 + dx) * s, 72 * s), QPointF((136 + dx) * s, 128 * s), + QPointF((70 + dx) * s, 184 * s), QPointF((70 + dx) * s, 152 * s), + QPointF((100 + dx) * s, 128 * s), QPointF((70 + dx) * s, 104 * s)])) + p.end() + return img + + +def png_bytes(img: QImage) -> bytes: + buf = QBuffer() + buf.open(QBuffer.WriteOnly) + img.save(buf, "PNG") + return bytes(buf.data()) + + +def write_ico(path: Path, sizes=(16, 24, 32, 48, 64, 128, 256)) -> None: + images = [png_bytes(render(n)) for n in sizes] + header = struct.pack(" None: + exe = Path(sys.argv[1]).resolve() + shot = Path(sys.argv[2] if len(sys.argv) > 2 else "smoke-test.png").resolve() + shot.unlink(missing_ok=True) + env = dict(os.environ, QT_QPA_PLATFORM="offscreen", FFWRAP_SMOKE_TEST=str(shot)) + res = subprocess.run([str(exe)], env=env, timeout=180) + if res.returncode != 0 or not shot.is_file(): + sys.exit(f"Smoke test failed (exit code {res.returncode}, screenshot {'ok' if shot.is_file() else 'missing'})") + print(f"Smoke test passed; screenshot saved to {shot}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_command.py b/tests/test_command.py new file mode 100644 index 0000000..677bd94 --- /dev/null +++ b/tests/test_command.py @@ -0,0 +1,239 @@ +from pathlib import Path + +import pytest + +from ffwrap.command import ( + Options, build_command, match_audio_selection, plan_output_path, scale_filter, temp_output_path, +) +from ffwrap.probe import MediaInfo, Stream + + +def video(codec="h264", w=1920, h=1080, rotation=0, index=0): + return Stream(index=index, kind="video", codec=codec, width=w, height=h, fps=23.976, rotation=rotation) + + +def audio(index, codec="aac", lang="eng", ch=2, title=None, default=False, rate=48000): + return Stream(index=index, kind="audio", codec=codec, language=lang, channels=ch, + title=title, default=default, sample_rate=rate) + + +def sub(index, codec="subrip", lang="eng"): + return Stream(index=index, kind="subtitle", codec=codec, language=lang) + + +def media(*streams, path="/videos/movie.mkv"): + return MediaInfo(path=Path(path), duration=100.0, format_name="matroska", streams=list(streams)) + + +def build(info, audio_idx=None, **opts): + o = Options(**opts) + idx = [s.index for s in info.audio] if audio_idx is None else audio_idx + return build_command("ffmpeg", info, o, idx, Path("/out/movie.mp4")) + + +def pairs(args): + """Return the args as a set of 'flag value' strings for easy assertions.""" + return {f"{a} {b}" for a, b in zip(args, args[1:])} + + +# ---------------------------------------------------------------- remuxing + +def test_jellyfin_remux_mkv_to_mp4(): + info = media(video("hevc"), audio(1, "ac3", ch=6), audio(2, "dts", ch=6), sub(3), sub(4, "hdmv_pgs_subtitle")) + b = build(info, container="mp4", video_mode="copy", audio_mode="auto", subtitles=True) + p = pairs(b.args) + assert "-c:v copy" in p + assert "-tag:v hvc1" in p # Apple-friendly HEVC tag + assert "-c:a:0 copy" in p # AC3 is fine in MP4 + assert "-c:a:1 aac" in p # DTS gets converted + assert "-c:s:0 mov_text" in p # SRT -> mov_text + assert "-map 0:4" not in p # PGS dropped for MP4 + assert any("dropped" in n for n in b.notes) + assert "-movflags +faststart" in p + assert b.args[-1] == str(Path("/out/movie.mp4")) + + +def test_mkv_keeps_everything(): + info = media(video(), audio(1, "truehd", ch=8), sub(2, "hdmv_pgs_subtitle"), sub(3, "ass"), + Stream(index=4, kind="attachment", codec="ttf")) + b = build(info, container="mkv", video_mode="copy", audio_mode="auto") + p = pairs(b.args) + assert {"-c:v copy", "-c:a:0 copy", "-c:s:0 copy", "-c:s:1 copy", "-map 0:t?", "-c:t copy"} <= p + assert "-movflags +faststart" not in p + + +def test_copy_mode_warns_about_incompatible_codec(): + info = media(video("vp9"), audio(1, "opus")) + b = build(info, container="avi", video_mode="copy", audio_mode="copy") + assert any("vp9" in n and "Warning" in n for n in b.notes) + + +def test_auto_video_reencodes_only_when_needed(): + info = media(video("vp9"), audio(1)) + assert "-c:v copy" in pairs(build(info, container="mkv", video_mode="auto").args) + b = build(info, container="mp4", video_mode="auto") # vp9 is OK in MP4 too + assert "-c:v copy" in pairs(b.args) + b = build(info, container="mov", video_mode="auto", video_encoder="libx264") + assert "-c:v libx264" in pairs(b.args) + b = build(info, container="mkv", video_mode="auto", resolution=720) + assert "-c:v libx264" in pairs(b.args) + + +def test_cover_art_is_not_mapped_as_video(): + cover = Stream(index=0, kind="video", codec="mjpeg", width=500, height=500, attached_pic=True) + info = media(cover, video("h264", index=1), audio(2)) + b = build(info, container="mp4") + assert "-map 0:1" in pairs(b.args) + assert "-map 0:0" not in pairs(b.args) + + +# --------------------------------------------------------------- encoding + +def test_reencode_h264_with_scaling_and_fps(): + info = media(video("hevc", 3840, 2160), audio(1)) + b = build(info, container="mp4", video_mode="encode", video_encoder="libx264", resolution=1080, + fps="29.97", quality=21, speed=3) + p = pairs(b.args) + assert "-vf scale=-2:1080:flags=lanczos,fps=30000/1001" in p + assert {"-c:v libx264", "-crf 21", "-preset slow", "-pix_fmt yuv420p"} <= p + + +def test_quality_is_clamped_to_encoder_range(): + info = media(video(), audio(1)) + b = build(info, video_mode="encode", video_encoder="mpeg4", container="avi", quality=60) + assert "-q:v 31" in pairs(b.args) + + +def test_strip_audio(): + info = media(video(), audio(1), audio(2)) + b = build(info, audio_mode="strip") + assert "-an" in b.args + assert not any(a.startswith("-c:a") for a in b.args) + assert "-map 0:1" not in pairs(b.args) + + +def test_selected_audio_tracks_in_order_with_default_disposition(): + info = media(video(), audio(1, lang="jpn"), audio(2, lang="eng"), audio(3, lang="eng", title="Commentary")) + b = build(info, audio_idx=[2, 1], audio_mode="copy") + maps = [b.args[i + 1] for i, a in enumerate(b.args) if a == "-map"] + assert maps == ["0:0", "0:2", "0:1"] + p = pairs(b.args) + assert {"-disposition:a:0 default", "-disposition:a:1 0"} <= p + + +def test_audio_downmix_resample_and_bitrate(): + info = media(video(), audio(1, "dts", ch=6, rate=48000)) + b = build(info, audio_mode="encode", audio_encoder="aac", audio_channels=2, audio_sample_rate=44100, + audio_bitrate=160) + assert {"-c:a:0 aac", "-ac:a:0 2", "-ar:a:0 44100", "-b:a:0 160k"} <= pairs(b.args) + + +def test_auto_audio_reencodes_when_channels_change(): + info = media(video(), audio(1, "aac", ch=6)) + b = build(info, audio_mode="auto", audio_channels=2) + assert "-c:a:0 aac" in pairs(b.args) + b = build(info, audio_mode="auto") + assert "-c:a:0 copy" in pairs(b.args) + + +def test_audio_encoder_limits(): + info = media(video(), audio(1, "truehd", ch=8)) + b = build(info, container="mkv", audio_mode="encode", audio_encoder="libmp3lame", audio_sample_rate=96000) + p = pairs(b.args) + assert "-ac:a:0 2" in p # MP3 is stereo only + assert "-ar:a:0 48000" in p # nearest supported rate + b = build(info, container="mkv", audio_mode="encode", audio_encoder="libopus") + assert "-mapping_family:a:0 1" in pairs(b.args) + b = build(info, container="mkv", audio_mode="encode", audio_encoder="flac") + assert not any(a.startswith("-b:a") for a in b.args) + + +# ---------------------------------------------------------------- scaling + +@pytest.mark.parametrize("w,h,rot,res,expected", [ + (1920, 1080, 0, 720, "scale=-2:720:flags=lanczos"), + (1080, 1920, 0, 720, "scale=720:-2:flags=lanczos"), # portrait + (1920, 1080, 90, 720, "scale=720:-2:flags=lanczos"), # rotated phone video + (1920, 1080, 0, 1080, None), # already that size + (1280, 720, 0, 2160, "scale=-2:2160:flags=lanczos"), # upscale allowed + (1920, 1080, 0, 0, None), +]) +def test_scale_filter(w, h, rot, res, expected): + assert scale_filter(video(w=w, h=h, rotation=rot), Options(resolution=res)) == expected + + +def test_scale_no_upscale_and_custom(): + v = video(w=1280, h=720) + assert scale_filter(v, Options(resolution=1080, no_upscale=True)) is None + assert scale_filter(v, Options(resolution=480, no_upscale=True)) == "scale=-2:480:flags=lanczos" + custom = scale_filter(v, Options(resolution=-1, custom_width=640, custom_height=640)) + assert custom.startswith("scale=640:640:force_original_aspect_ratio=decrease") + assert scale_filter(v, Options(resolution=-1, custom_width=853)) == "scale=852:-2:flags=lanczos" + assert scale_filter(v, Options(resolution=-1)) is None + + +# ------------------------------------------------------------ output paths + +def test_output_next_to_source(tmp_path): + src = tmp_path / "a.mkv" + assert plan_output_path(src, Options(container="mp4"), set()) == tmp_path / "a.mp4" + + +def test_output_never_overwrites_source(tmp_path): + src = tmp_path / "a.mp4" + src.touch() + out = plan_output_path(src, Options(container="mp4", if_exists="overwrite"), set()) + assert out == tmp_path / "a_converted.mp4" + + +def test_output_exists_policies(tmp_path): + src = tmp_path / "a.mkv" + (tmp_path / "a.mp4").touch() + assert plan_output_path(src, Options(if_exists="skip"), set()) is None + assert plan_output_path(src, Options(if_exists="overwrite"), set()) == tmp_path / "a.mp4" + assert plan_output_path(src, Options(if_exists="rename"), set()) == tmp_path / "a (1).mp4" + + +def test_output_custom_dir_and_batch_collisions(tmp_path): + out_dir = tmp_path / "out" + opts = Options(output_dir=str(out_dir), suffix="_web", if_exists="overwrite") + first = plan_output_path(tmp_path / "s1" / "ep.mkv", opts, set()) + assert first == out_dir / "ep_web.mp4" + taken = {__import__("os").path.normcase(str(first))} + # Same name from another folder in the same batch must not clobber the first one. + assert plan_output_path(tmp_path / "s2" / "ep.mkv", opts, taken) == out_dir / "ep_web (1).mp4" + + +def test_temp_output_path(): + assert temp_output_path(Path("/x/movie.mp4")) == Path("/x/movie.part.mp4") + + +# ----------------------------------------------------------- track matching + +def test_match_identical_layout_by_position(): + ref = [audio(1, lang="jpn"), audio(2, lang="eng")] + tgt = [audio(3, lang="jpn"), audio(4, lang="eng")] + assert match_audio_selection(ref, [2], tgt) == [4] + + +def test_match_by_language_when_layout_differs(): + ref = [audio(1, "ac3", "jpn", 6), audio(2, "aac", "eng", 2)] + tgt = [audio(1, "aac", "eng", 2), audio(2, "ac3", "jpn", 6), audio(3, "aac", "eng", 2, title="Commentary")] + assert match_audio_selection(ref, [2, 1], tgt) == [1, 2] + + +def test_match_falls_back_to_default_track(): + ref = [audio(1, lang="eng")] + tgt = [audio(1, lang="fre"), audio(2, lang="ger", default=True)] + assert match_audio_selection(ref, [1], tgt) == [2] + + +def test_match_empty_selection(): + assert match_audio_selection([audio(1)], [], [audio(1)]) == [] + assert match_audio_selection([audio(1)], [1], []) == [] + + +def test_options_roundtrip(): + o = Options(container="mkv", quality=18, no_upscale=True, output_dir="/tmp/x") + assert Options.from_dict(o.to_dict()) == o + assert Options.from_dict({"quality": "bad", "unknown": 1}) == Options() diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..d47592d --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,75 @@ +"""End-to-end tests against a real ffmpeg. Skipped when ffmpeg is not available. + +Set FFWRAP_FFMPEG_DIR to point at a folder containing ffmpeg/ffprobe, or put them on PATH. +""" + +import os +import subprocess +from pathlib import Path + +import pytest + +from ffwrap.command import Options, build_command, match_audio_selection +from ffwrap.probe import probe +from ffwrap.tools import locate_tools + +TOOLS = locate_tools(os.environ.get("FFWRAP_FFMPEG_DIR")) +pytestmark = pytest.mark.skipif(TOOLS is None, reason="ffmpeg/ffprobe not found") + + +@pytest.fixture(scope="module") +def sample(tmp_path_factory) -> Path: + """3 s 640x360 H.264 + jpn 5.1 AC3 + eng stereo AAC + English SRT in an MKV.""" + d = tmp_path_factory.mktemp("media") + srt = d / "subs.srt" + srt.write_text("1\n00:00:00,500 --> 00:00:02,000\nHello\n", encoding="utf-8") + out = d / "sample.mkv" + subprocess.run([ + TOOLS.ffmpeg, "-v", "error", "-y", + "-f", "lavfi", "-i", "testsrc2=size=640x360:rate=25:duration=3", + "-f", "lavfi", "-i", "sine=frequency=440:duration=3", + "-f", "lavfi", "-i", "sine=frequency=880:duration=3", + "-i", str(srt), + "-map", "0", "-map", "1", "-map", "2", "-map", "3", + "-c:v", "libx264", "-preset", "ultrafast", + "-c:a:0", "ac3", "-ac:a:0", "6", "-c:a:1", "aac", "-ac:a:1", "2", "-c:s", "srt", + "-metadata:s:a:0", "language=jpn", "-metadata:s:a:1", "language=eng", + "-metadata:s:s:0", "language=eng", + str(out), + ], check=True) + return out + + +def run(info, opts, audio, out): + built = build_command(TOOLS.ffmpeg, info, opts, audio, out) + res = subprocess.run(built.args, capture_output=True, text=True) + assert res.returncode == 0, res.stderr + return probe(TOOLS.ffprobe, out) + + +def test_remux_to_mp4_keeps_streams(sample, tmp_path): + info = probe(TOOLS.ffprobe, sample) + eng = match_audio_selection(info.audio, [info.audio[1].index], info.audio) + out = run(info, Options(container="mp4"), eng, tmp_path / "out.mp4") + assert out.video.codec == "h264" + assert [(a.codec, a.language) for a in out.audio] == [("aac", "eng")] + assert [s.codec for s in out.subtitles] == ["mov_text"] + assert abs(out.duration - info.duration) < 0.5 + + +def test_reencode_scale_and_downmix(sample, tmp_path): + info = probe(TOOLS.ffprobe, sample) + opts = Options(container="mkv", video_mode="encode", video_encoder="libx265", resolution=240, + speed=0, audio_mode="encode", audio_encoder="libopus", audio_channels=2, subtitles=False) + out = run(info, opts, [a.index for a in info.audio], tmp_path / "out.mkv") + assert out.video.codec == "hevc" and (out.video.width, out.video.height) == (426, 240) + assert [(a.codec, a.channels) for a in out.audio] == [("opus", 2), ("opus", 2)] + assert out.subtitles == [] + + +def test_strip_audio_webm(sample, tmp_path): + info = probe(TOOLS.ffprobe, sample) + opts = Options(container="webm", video_mode="auto", video_encoder="libvpx-vp9", speed=0, audio_mode="strip") + out = run(info, opts, [], tmp_path / "out.webm") + assert out.video.codec == "vp9" and out.audio == [] + assert [s.codec for s in out.subtitles] == ["webvtt"] diff --git a/tests/test_probe.py b/tests/test_probe.py new file mode 100644 index 0000000..ac045f5 --- /dev/null +++ b/tests/test_probe.py @@ -0,0 +1,56 @@ +from pathlib import Path + +from ffwrap.probe import parse_probe +from ffwrap.tools import parse_encoders + +SAMPLE = { + "format": {"format_name": "matroska,webm", "duration": "1421.5"}, + "streams": [ + {"index": 0, "codec_type": "video", "codec_name": "hevc", "width": 1920, "height": 1080, + "avg_frame_rate": "24000/1001", "pix_fmt": "yuv420p10le", "disposition": {"default": 1}}, + {"index": 1, "codec_type": "audio", "codec_name": "eac3", "channels": 6, "channel_layout": "5.1(side)", + "sample_rate": "48000", "tags": {"language": "jpn", "title": "Surround"}, "disposition": {"default": 1}}, + {"index": 2, "codec_type": "audio", "codec_name": "aac", "channels": 2, "sample_rate": "44100", + "tags": {"LANGUAGE": "und", "BPS": "128000"}}, + {"index": 3, "codec_type": "subtitle", "codec_name": "subrip", "tags": {"language": "eng"}, + "disposition": {"forced": 1}}, + {"index": 4, "codec_type": "video", "codec_name": "mjpeg", "width": 600, "height": 900, + "disposition": {"attached_pic": 1}}, + ], +} + + +def test_parse_probe(): + info = parse_probe(Path("x.mkv"), SAMPLE) + assert info.duration == 1421.5 + assert info.video.index == 0 and info.video.codec == "hevc" + assert abs(info.video.fps - 23.976) < 0.001 + assert [s.index for s in info.audio] == [1, 2] + a1, a2 = info.audio + assert a1.language == "jpn" and a1.title == "Surround" and a1.default and a1.channels == 6 + assert a2.language is None and a2.bit_rate == 128000 # "und" -> None, BPS tag used + assert info.subtitles[0].forced + assert "EAC3" in a1.describe() and "5.1" in a1.describe() + + +def test_rotation_swaps_display_size(): + data = {"format": {}, "streams": [{"index": 0, "codec_type": "video", "codec_name": "h264", "width": 1920, + "height": 1080, "side_data_list": [{"rotation": -90}]}]} + info = parse_probe(Path("phone.mp4"), data) + assert info.video.display_size == (1080, 1920) + + +def test_duration_falls_back_to_streams(): + data = {"format": {}, "streams": [{"index": 0, "codec_type": "audio", "codec_name": "aac", "duration": "12.5"}]} + assert parse_probe(Path("a.m4a"), data).duration == 12.5 + + +def test_parse_encoders(): + text = """Encoders: + V..... = Video + ------ + V....D libx264 libx264 H.264 / AVC + A....D aac AAC (Advanced Audio Coding) + S..... mov_text 3GPP Timed Text subtitle +""" + assert parse_encoders(text) == {"libx264", "aac", "mov_text"}