From 3d082382623abf6af86a7eb326e8718b06902316 Mon Sep 17 00:00:00 2001 From: Zach Fleeman Date: Wed, 28 Oct 2026 12:19:48 -0600 Subject: [PATCH] Show a progress bar during CLI encodes (#135) --- ffmpeg4discord/__main__.py | 194 +++++++++++++++++++++++++++++-------- ffmpeg4discord/twopass.py | 5 +- tests/test_main.py | 128 +++++++++++++++++++++++- tests/test_twopass.py | 14 +++ 4 files changed, 300 insertions(+), 41 deletions(-) diff --git a/ffmpeg4discord/__main__.py b/ffmpeg4discord/__main__.py index cccaef6..6ef5069 100644 --- a/ffmpeg4discord/__main__.py +++ b/ffmpeg4discord/__main__.py @@ -5,6 +5,7 @@ import math import re import shutil +import sys import threading import time import webbrowser @@ -105,7 +106,7 @@ def success_message(result: EncodeResult) -> str: return f"Your compressed video file ({round(result.output_filesize, 2)}MB) is located at {output_path}" -# What the web UI's progress line calls each pass +# What the progress bars call each pass PASS_LABELS = {"first": "First pass", "second": "Second pass", "single": "Encoding"} @@ -124,31 +125,164 @@ def progress_seconds(progress_file: Optional[Path]) -> Optional[float]: @dataclass -class EncodeJob: +class EncodeProgress: """ - A web UI encode. It runs in a background thread while the page polls /status for its progress. + How far along an encode is. Its `start_pass` goes to `twopass_loop` as `on_pass`, and both progress bars read it. Attributes: - values (dict): The submitted form, to show again while and after the encode runs. clip_seconds (float): The trimmed clip's length, for working out the progress. codec (str): The encode's codec, e.g. "vp9". - state (str): "running", then "done" or "failed". attempt (int): Which attempt is running. ff4d tries again when an output is over the target. pass_name (str): The running pass: "first", "second", or "single" for hardware encoders. progress_file (Path): The file ffmpeg writes the running pass's progress to. pass_started (float): When the running pass started, from time.monotonic(). - result (EncodeResult): The finished encode, once the state is "done". - errors (list[str]): What went wrong, once the state is "failed". + seconds (float): How many seconds of video the running pass has encoded, or None before ffmpeg says. """ - values: dict clip_seconds: float codec: str = "" - state: str = "running" attempt: int = 0 pass_name: str = "" progress_file: Optional[Path] = None pass_started: float = 0.0 + seconds: Optional[float] = None + + def start_pass(self, pass_name: str, progress_file: Path) -> None: + """TwoPass.run() calls this before each pass. A first or single pass means a new attempt has started.""" + if pass_name in ("first", "single"): + self.attempt += 1 + self.progress_file = progress_file + self.pass_started = time.monotonic() + self.seconds = None + self.pass_name = pass_name + + def finish_pass(self) -> None: + """Marks the running pass as done. ffmpeg's last report says so, but it can be deleted before anyone reads it.""" + self.seconds = self.clip_seconds + + def report(self) -> tuple[float, str]: + """The percent of the whole encode that's done, and a label that describes it.""" + if not self.pass_name: + return 0, "Starting…" + + # Each pass of a two-pass encode gets half of the bar + start, share = {"first": (0, 50), "second": (50, 50), "single": (0, 100)}[self.pass_name] + # Keep the furthest time ffmpeg reported. It can dip near a pass's end, and TwoPass.run() deletes the file + # when an attempt ends. + seconds = progress_seconds(self.progress_file) + if seconds is not None: + self.seconds = max(seconds, self.seconds or 0) + if self.seconds is None: + # No time yet. Most passes report one within a second, but VP9's and AV1's first passes write no video, + # so they never do. + done = 0 + label = f"{PASS_LABELS[self.pass_name]}: running for {time.monotonic() - self.pass_started:.0f} s" + if self.codec in ("vp9", "av1") and self.pass_name == "first": + label += f". {self.codec.upper()} doesn't report progress until this pass ends." + else: + done = min(self.seconds / self.clip_seconds, 1) + label = f"{PASS_LABELS[self.pass_name]}: {done:.0%}" + if self.attempt > 1: + label += f". Attempt {self.attempt}, since the last one was over the target." + return round(start + share * done, 1), label + + +# How many characters wide the CLI's bar is, between its brackets +BAR_WIDTH = 30 + + +def bar_line(percent: float, label: str, width: int) -> str: + """The CLI's progress line, e.g. "[###---] 50% Second pass: 0%", cut or padded to `width` characters.""" + filled = round(BAR_WIDTH * percent / 100) + line = f"[{'#' * filled}{'-' * (BAR_WIDTH - filled)}] {percent:3.0f}% {label}" + # The padding covers the end of a longer line drawn before it + return line[:width].ljust(width) + + +class ProgressLine: + """ + Shows a CLI encode's progress on one line that a background thread redraws in place. + + It stands in for sys.stdout while the encode runs, so ff4d's own prints, like "Performing second pass", end the + progress line first instead of running into it. ff4d only prints between passes, so those prints also mean the + pass is done. + """ + + def __init__(self, progress: EncodeProgress): + self.progress = progress + self.stream = sys.stdout + # Reentrant, because write() ends the line, and ending it draws it one last time + self.lock = threading.RLock() + self.drawn = "" # the progress line on screen, or "" when the cursor isn't on one + self.at_line_start = True + self.stopped = threading.Event() + self.thread = threading.Thread(target=self.redraw_until_stopped, name="Progress", daemon=True) + + def __enter__(self) -> "ProgressLine": + sys.stdout = self + self.thread.start() + return self + + def __exit__(self, exc_type, *_) -> None: + self.stopped.set() + self.thread.join() + sys.stdout = self.stream + # A pass that stopped with an error or Ctrl+C keeps the percent it got to + self.end_line(pass_done=exc_type is None) + + def redraw_until_stopped(self) -> None: + # ffmpeg updates its -progress file about twice a second + while not self.stopped.wait(0.5): + self.draw() + + def draw(self) -> None: + """Redraws the line if the progress changed. It waits for a pass to start and for other prints to end their line.""" + with self.lock: + if not self.progress.pass_name or not (self.drawn or self.at_line_start): + return + line = "\r" + bar_line(*self.progress.report(), shutil.get_terminal_size().columns - 1) + if line != self.drawn: + self.stream.write(line) + self.stream.flush() + self.drawn = line + + def end_line(self, pass_done: bool = True) -> None: + """Draws the line one last time, so it keeps the pass's final percent, then moves below it.""" + with self.lock: + if self.drawn: + if pass_done: + self.progress.finish_pass() + self.draw() + self.stream.write("\n") + self.drawn = "" + self.at_line_start = True + + def write(self, text: str) -> int: + with self.lock: + if text: + self.end_line() + self.stream.write(text) + self.at_line_start = text.endswith("\n") + return len(text) + + def flush(self) -> None: + self.stream.flush() + + +@dataclass +class EncodeJob(EncodeProgress): + """ + A web UI encode. It runs in a background thread while the page polls /status for its progress. + + Attributes: + values (dict): The submitted form, to show again while and after the encode runs. + state (str): "running", then "done" or "failed". + result (EncodeResult): The finished encode, once the state is "done". + errors (list[str]): What went wrong, once the state is "failed". + """ + + values: dict = field(default_factory=dict) + state: str = "running" result: Optional[EncodeResult] = None errors: list[str] = field(default_factory=list) @@ -172,14 +306,6 @@ def run(self, twopass: TwoPass, settings: EncodeSettings, approx: bool) -> None: self.errors = ["Something went wrong while encoding. The details are in the terminal running ff4d."] self.state = "failed" - def start_pass(self, pass_name: str, progress_file: Path) -> None: - """TwoPass.run() calls this before each pass. A first or single pass means a new attempt has started.""" - if pass_name in ("first", "single"): - self.attempt += 1 - self.progress_file = progress_file - self.pass_started = time.monotonic() - self.pass_name = pass_name - def remove_temp_folder(self) -> None: """ Deletes a running encode's temp folder. The encode thread is a daemon, so when Ctrl+C stops the server, @@ -200,25 +326,8 @@ def status(self) -> dict: """What /status sends the page. While running, `percent` covers the whole encode and `label` describes it.""" if self.state != "running": return {"state": self.state} - if not self.pass_name: - return {"state": "running", "percent": 0, "label": "Starting…"} - - # Each pass of a two-pass encode gets half of the bar - start, share = {"first": (0, 50), "second": (50, 50), "single": (0, 100)}[self.pass_name] - seconds = progress_seconds(self.progress_file) - if seconds is None: - # No time yet. Most passes report one within a second, but VP9's and AV1's first passes write no video, - # so they never do. - done = 0 - label = f"{PASS_LABELS[self.pass_name]}: running for {time.monotonic() - self.pass_started:.0f} s" - if self.codec in ("vp9", "av1") and self.pass_name == "first": - label += f". {self.codec.upper()} doesn't report progress until this pass ends." - else: - done = min(seconds / self.clip_seconds, 1) - label = f"{PASS_LABELS[self.pass_name]}: {done:.0%}" - if self.attempt > 1: - label = f"Attempt {self.attempt}, since the last one was over the target. {label}" - return {"state": "running", "percent": round(start + share * done, 1), "label": label} + percent, label = self.report() + return {"state": "running", "percent": percent, "label": label} def source_facts(twopass: TwoPass) -> dict: @@ -527,7 +636,7 @@ def form_twopass(): start, end = twopass.clip_times(form_settings) values = form_values(twopass, form_settings, form_approx) - job = EncodeJob(values, clip_seconds=end - start, codec=form_settings.codec) + job = EncodeJob(values=values, clip_seconds=end - start, codec=form_settings.codec) worker = threading.Thread( target=job.run, args=[twopass, form_settings, form_approx], name="Encode", daemon=True ) @@ -540,7 +649,16 @@ def form_twopass(): app.run("127.0.0.1", port=port) else: with exit_on_setting_error(): - result = twopass.run(settings, dry_run=True) if dry_run else twopass_loop(twopass, settings, approx) + if dry_run: + result = twopass.run(settings, dry_run=True) + elif settings.verbose or not sys.stdout.isatty(): + # --verbose prints ffmpeg's log, which a redrawn line would garble. So would a pipe or a log file. + result = twopass_loop(twopass, settings, approx) + else: + start, end = twopass.clip_times(settings) + progress = EncodeProgress(clip_seconds=end - start, codec=settings.codec) + with ProgressLine(progress): + result = twopass_loop(twopass, settings, approx, on_pass=progress.start_pass) if dry_run: print(format_commands(result.commands)) diff --git a/ffmpeg4discord/twopass.py b/ffmpeg4discord/twopass.py index e1c2b76..438cfce 100644 --- a/ffmpeg4discord/twopass.py +++ b/ffmpeg4discord/twopass.py @@ -647,7 +647,8 @@ def run( :param settings: what to encode and how :param dry_run: build the ffmpeg commands without running them :param on_pass: called with the pass name ("first", "second", or "single") and a file that ffmpeg writes - its progress to, just before each pass starts. The web UI uses it for its progress bar. + its progress to, just before each pass starts. The web UI and the CLI use it for their progress bars, so + ffmpeg's own stats line is left out unless `settings.verbose` is set. :return: the output file, its size, and the ffmpeg commands """ @@ -706,7 +707,7 @@ def build_passes(pass1: dict, pass2: dict) -> tuple: def with_logging(ffoutput, pass_name: str): """Adds ff4d's logging flags, plus a -progress file when `on_pass` is watching.""" - args = ["-loglevel", loglevel, "-stats"] + args = ["-loglevel", loglevel, "-nostats" if on_pass and not settings.verbose else "-stats"] if on_pass: progress_file = Path(log_dir, f"{pass_name}-progress.txt") args += ["-progress", str(progress_file)] diff --git a/tests/test_main.py b/tests/test_main.py index 55f8f24..5b5da2f 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,6 +1,9 @@ +import io import json +import os import re import sys +from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace from unittest.mock import ANY, MagicMock @@ -562,7 +565,7 @@ def fake_loop(twopass, settings, approx, on_pass): (["first"], 6, 10, "First pass: 20%"), (["first", "second"], 30, 100, "Second pass: 100%"), (["single"], 15, 50, "Encoding: 50%"), - (["first", "second", "first"], 3, 5, "Attempt 2, since the last one was over the target. First pass: 10%"), + (["first", "second", "first"], 3, 5, "First pass: 10%. Attempt 2, since the last one was over the target."), (["first"], 45, 50, "First pass: 100%"), # out_time can run a little past the clip's end ], ) @@ -619,6 +622,129 @@ def test_job_status_before_ffmpeg_reports_a_time(tmp_path, monkeypatch, codec, p assert job.status() == {"state": "running", "percent": start, "label": label} +def test_progress_keeps_the_furthest_time(tmp_path): + progress_file = tmp_path / "second-progress.txt" + progress_file.write_text("out_time_us=27000000\n") + progress = mainmod.EncodeProgress(clip_seconds=30) + progress.start_pass("second", progress_file) + assert progress.report() == (95, "Second pass: 90%") + # ffmpeg's time can dip near the end of a pass, and the file is gone once the attempt ends + progress_file.write_text("out_time_us=24000000\n") + assert progress.report() == (95, "Second pass: 90%") + progress_file.unlink() + assert progress.report() == (95, "Second pass: 90%") + progress.finish_pass() + assert progress.report() == (100, "Second pass: 100%") + # a new pass starts from nothing + progress.start_pass("first", progress_file) + assert progress.report()[0] == 0 + + +# --- the CLI's progress line (issue #135) --- + + +def test_bar_line(): + line = mainmod.bar_line(50, "Second pass: 0%", 60) + assert line == "[###############---------------] 50% Second pass: 0%".ljust(60) + # cut to the terminal's width, so redrawing it doesn't wrap onto a second line + assert mainmod.bar_line(100, "x" * 100, 40) == "[" + "#" * 30 + "] 100% x" + + +@pytest.fixture +def screen(monkeypatch): + """A 41-column terminal for a ProgressLine to draw on.""" + monkeypatch.setattr(mainmod.shutil, "get_terminal_size", lambda: os.terminal_size((41, 24))) + return io.StringIO() + + +def first_pass_line(screen, tmp_path): + """A ProgressLine that draws on `screen`, for a 30-second clip 15 seconds into its first pass.""" + progress_file = tmp_path / "first-progress.txt" + progress_file.write_text("out_time_us=15000000\n") + progress = mainmod.EncodeProgress(clip_seconds=30) + progress.start_pass("first", progress_file) + # Set here, not in the screen fixture, because pytest swaps its own capture back in before each test runs + line = mainmod.ProgressLine(progress) + line.stream = screen + return line + + +def test_progress_line_waits_for_a_pass_to_start(screen): + line = mainmod.ProgressLine(mainmod.EncodeProgress(clip_seconds=30)) + line.stream = screen + line.draw() + assert screen.getvalue() == "" + + +def test_progress_line_redraws_only_when_it_changes(screen, tmp_path): + line = first_pass_line(screen, tmp_path) + line.draw() + line.draw() + assert screen.getvalue() == "\r" + mainmod.bar_line(25, "First pass: 50%", 40) + + +def test_a_print_ends_the_progress_line_at_the_passs_final_percent(screen, tmp_path): + line = first_pass_line(screen, tmp_path) + line.draw() + print("Performing second pass", file=line, flush=True) + first, final = mainmod.bar_line(25, "First pass: 50%", 40), mainmod.bar_line(50, "First pass: 100%", 40) + assert screen.getvalue() == f"\r{first}\r{final}\nPerforming second pass\n" + + +def test_progress_line_waits_for_a_print_to_end_its_line(screen, tmp_path): + line = first_pass_line(screen, tmp_path) + line.write("Performing") + line.draw() + assert screen.getvalue() == "Performing" + line.write(" second pass\n") + line.draw() + assert screen.getvalue() == "Performing second pass\n\r" + mainmod.bar_line(25, "First pass: 50%", 40) + + +@pytest.mark.parametrize("error, final", [(None, "First pass: 100%"), (KeyboardInterrupt, "First pass: 50%")]) +def test_progress_line_stands_in_for_stdout(screen, monkeypatch, tmp_path, error, final): + monkeypatch.setattr(mainmod.sys, "stdout", screen) + line = mainmod.ProgressLine(first_pass_line(screen, tmp_path).progress) + with pytest.raises(error) if error else nullcontext(): + with line: + assert mainmod.sys.stdout is line + line.draw() + if error: + raise error + assert mainmod.sys.stdout is screen + assert not line.thread.is_alive() + # an encode that stopped early keeps the percent it got to + assert screen.getvalue().endswith(mainmod.bar_line(*line.progress.report(), 40) + "\n") + assert line.progress.report()[1] == final + + +def test_progress_line_redraws_until_stopped(screen, monkeypatch, tmp_path): + line = first_pass_line(screen, tmp_path) + monkeypatch.setattr(line, "draw", MagicMock()) + monkeypatch.setattr(line.stopped, "wait", MagicMock(side_effect=[False, False, True])) + line.redraw_until_stopped() + assert line.draw.call_count == 2 + + +@pytest.mark.parametrize( + "tty, verbose, shows_bar", + [ + (True, False, True), + (False, False, False), # piped or saved to a file + (True, True, False), # ffmpeg's verbose log would garble the line + ], +) +def test_main_cli_shows_a_progress_line_in_a_terminal(run_main, twopass, monkeypatch, tty, verbose, shows_bar): + monkeypatch.setattr(mainmod.sys.stdout, "isatty", lambda: tty) + twopass.clip_times.return_value = (5, 15) + twopass_loop = run_main(verbose=verbose, codec="vp9") + on_pass = twopass_loop.call_args.kwargs.get("on_pass") + if shows_bar: + assert on_pass.__self__ == mainmod.EncodeProgress(clip_seconds=10, codec="vp9") + else: + assert on_pass is None + + def test_exit_removes_the_running_encodes_temp_folder(web, tmp_path): folder = tmp_path / "ff4d-test" folder.mkdir() diff --git a/tests/test_twopass.py b/tests/test_twopass.py index 34cffd4..7e7e303 100644 --- a/tests/test_twopass.py +++ b/tests/test_twopass.py @@ -541,6 +541,20 @@ def test_run_without_on_pass_writes_no_progress(make_twopass, passes): assert all("-progress" not in p.args for p in passes) +@pytest.mark.parametrize( + "watched, verbose, stats", + [ + (False, False, "-stats"), + (True, False, "-nostats"), # the progress bar replaces ffmpeg's stats line + (True, True, "-stats"), # --verbose keeps ffmpeg's own output + ], +) +def test_run_leaves_out_the_stats_line_when_on_pass_watches(make_twopass, passes, watched, verbose, stats): + on_pass = (lambda name, path: None) if watched else None + make_twopass().run(make_settings(verbose=verbose), on_pass=on_pass) + assert all(stats in p.args for p in passes) + + # --- showing the ffmpeg commands (issue #92) ---