From 97c98b53f51d6fdd9be91bb1d275e0d4d1c97aa1 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 6 Jul 2026 09:49:12 +0900 Subject: [PATCH] feat: print batch progress with ETA during --execute runs Long batch conversions previously emitted only per-file result lines, leaving no way to gauge overall completion. After each file finishes, the CLI now prints a tab-separated progress line: PROGRESS\t/\t%\tETA The ETA extrapolates from the average per-file wall time so far (measured with time.monotonic) times the remaining file count. Output stays consistent with the existing tab-separated CONVERTED/DRY-RUN style; no progress bars, threads, or new dependencies. Dry runs are unaffected. Tests cover the PROGRESS line formatting (counts, percent, mm:ss ETA, negative clamping), a deterministic workers=1 batch with a patched monotonic clock, and the absence of PROGRESS output in dry-run mode. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CJRVbDrp1vGYkJgNHMGPpG --- media_shrinker.py | 29 ++++++++++++- tests/test_media_shrinker.py | 80 ++++++++++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+), 1 deletion(-) diff --git a/media_shrinker.py b/media_shrinker.py index d72a8d5..184108a 100644 --- a/media_shrinker.py +++ b/media_shrinker.py @@ -18,6 +18,7 @@ import subprocess import sys import tempfile +import time from dataclasses import dataclass from datetime import datetime from pathlib import Path @@ -1250,13 +1251,32 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: return parser.parse_args(_normalize_argv(argv)) +def _format_progress(completed: int, total: int, eta_seconds: float) -> str: + """Format a tab-separated batch progress line with an mm:ss ETA. + + ``completed`` and ``total`` are the number of finished and queued + candidates respectively (``total`` must be positive). ``eta_seconds`` is + the estimated remaining wall time; it is clamped at zero and truncated to + whole seconds before formatting, and minutes may exceed 59 for long + batches. + """ + percent = completed * 100 // total + eta = max(0, int(eta_seconds)) + return f"PROGRESS\t{completed}/{total}\t{percent}%\tETA {eta // 60:02d}:{eta % 60:02d}" + + def _execute_conversions( candidates: list[tuple[Path, int]], args: argparse.Namespace, root: Path, output_dir: Path, ) -> list[ConversionResult]: - """Execute conversions in parallel.""" + """Execute conversions in parallel, printing per-file results and progress. + + After each candidate finishes, a ``PROGRESS`` line reports the completed + count, percentage, and an ETA extrapolated from the average per-file wall + time so far (measured with a monotonic clock). + """ results: list[ConversionResult] = [] workers = choose_worker_count(args.workers) ffmpeg_threads = args.ffmpeg_threads if args.ffmpeg_threads >= 0 else None @@ -1299,6 +1319,9 @@ def process_candidate(candidate_tuple: tuple[Path, int]) -> list[ConversionResul ] print(f"WORKERS\t{workers}\tFFMPEG_THREADS\t{ffmpeg_threads}") + total = len(candidates) + completed = 0 + start_time = time.monotonic() with ThreadPoolExecutor(max_workers=workers) as executor: future_to_candidate = { executor.submit(process_candidate, candidate): candidate @@ -1309,6 +1332,10 @@ def process_candidate(candidate_tuple: tuple[Path, int]) -> list[ConversionResul results.extend(candidate_results) for result in candidate_results: print(_format_result(root, result), flush=True) + completed += 1 + elapsed = time.monotonic() - start_time + eta_seconds = (elapsed / completed) * (total - completed) + print(_format_progress(completed, total, eta_seconds), flush=True) return results diff --git a/tests/test_media_shrinker.py b/tests/test_media_shrinker.py index 0ab2397..9c753d6 100644 --- a/tests/test_media_shrinker.py +++ b/tests/test_media_shrinker.py @@ -1336,6 +1336,86 @@ def fake_convert_file(source: Path, **_kwargs): [item.message for item in results if item.status == "failed"], ["boom"] ) + def test_format_progress_formats_counts_percent_and_eta(self) -> None: + self.assertEqual( + media_shrinker._format_progress(1, 4, 125.9), + "PROGRESS\t1/4\t25%\tETA 02:05", + ) + self.assertEqual( + media_shrinker._format_progress(3, 3, 0.0), + "PROGRESS\t3/3\t100%\tETA 00:00", + ) + + def test_format_progress_clamps_negative_eta_to_zero(self) -> None: + self.assertEqual( + media_shrinker._format_progress(2, 2, -5.0), + "PROGRESS\t2/2\t100%\tETA 00:00", + ) + + def test_execute_conversions_prints_progress_lines_with_eta(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + sources = [] + for name in ("a.wav", "b.wav", "c.wav"): + source = root / name + source.write_bytes(b"12") + sources.append(source) + args = media_shrinker.parse_args([str(root), "--workers", "1"]) + + def fake_convert_file(source: Path, **_kwargs): + return [ + media_shrinker.ConversionResult( + source_path=source, + output_path=root / "out" / (source.stem + ".flac"), + status="converted", + original_size_bytes=2, + output_size_bytes=1, + strategy="flac-lossless", + ) + ] + + # One call at batch start, then one after each completion: + # elapsed 10s per file -> average 10s -> ETA 20s, 10s, 0s. + clock = iter([100.0, 110.0, 120.0, 130.0]) + with patch("media_shrinker.convert_file", side_effect=fake_convert_file): + with patch( + "media_shrinker.time.monotonic", + side_effect=lambda: next(clock, 130.0), + ): + with patch("builtins.print") as mock_print: + results = media_shrinker._execute_conversions( + [(source, 2) for source in sources], + args, + root, + root / "out", + ) + + self.assertEqual(len(results), 3) + printed = ["\t".join(map(str, call.args)) for call in mock_print.call_args_list] + progress_lines = [line for line in printed if line.startswith("PROGRESS\t")] + self.assertEqual( + progress_lines, + [ + "PROGRESS\t1/3\t33%\tETA 00:20", + "PROGRESS\t2/3\t66%\tETA 00:10", + "PROGRESS\t3/3\t100%\tETA 00:00", + ], + ) + + def test_main_dry_run_prints_no_progress_lines(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + (root / "source.wav").write_bytes(b"1234") + + with patch("builtins.print") as mock_print: + rc = media_shrinker.main([str(root), "--size-limit-bytes", "1"]) + + self.assertEqual(rc, 0) + printed = ["\t".join(map(str, call.args)) for call in mock_print.call_args_list] + self.assertFalse( + [line for line in printed if line.startswith("PROGRESS")], printed + ) + def test_small_segment_returns_single_segment(self) -> None: segments = build_segments(duration_seconds=1.0, max_segment_duration_seconds=2.0) self.assertEqual(segments, [MediaSegment(1, 0.0, 1.0, 1)])