diff --git a/README.md b/README.md index 7d0f3b3..7ac933b 100644 --- a/README.md +++ b/README.md @@ -10,18 +10,15 @@ frame= 692 fps= 58 q=28.0 size= 5376KiB time=00:00:28.77 bitrate=1530.3kbits ``` Better FFmpeg Progress outputs something like: ``` -⠏ Processing abc.webm ━━━━━━━━━╺━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 23% 0:00:04 00:15 +Processing abc.webm +(5.50 MB / ~451.67 MB) ━━╺━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 3% 0:00:11 04:15 ``` Where: - `Processing abc.webm` is the description of the progresss bar. -- `23%` is the percentage progress. -- `0:00:04` is the time elapsed. -- `00:15` is the estimated time until the FFmpeg process completes. - -## Installation -```bash -pip install better-ffmpeg-progress --upgrade -``` +- `(5.50 MB / ~451.67 MB)` is the current output size / estimated final output size. +- `3%` is the percentage progress. +- `0:00:11` is the time elapsed. +- `04:15` is the estimated time until the FFmpeg process completes. ## Usage Create an instance of the `FfmpegProcess` class and supply a list of arguments like you would when using `subprocess.run()` or `subprocess.Popen()`. Example: @@ -101,4 +98,4 @@ By default, the [Rich](https://github.com/Textualize/rich) library is used to di process = FfmpegProcess(command) process.use_tqdm = True process.run() -``` \ No newline at end of file +``` diff --git a/better_ffmpeg_progress.py b/better_ffmpeg_progress.py new file mode 100644 index 0000000..68b0d6b --- /dev/null +++ b/better_ffmpeg_progress.py @@ -0,0 +1,232 @@ +import math +import os +from pathlib import Path +import subprocess +from sys import exit +from typing import List, Optional, Union + +from .enums import FfmpegLogLevel +from .exceptions import ( + FfmpegProcessError, + FfmpegCommandError, + FfmpegProcessUserCancelledError, + FfmpegProcessInterruptedError, +) +from .utils import ( + validate_ffmpeg_command, + get_media_duration, + check_shell_needed_for_command, +) +from .terminate_process import terminate_ffmpeg_process +from .progress_bars import use_rich, use_tqdm + +_FFMPEG_OVERWRITE_FLAG = "-y" + + +class FfmpegProcess: + def _handle_overwrite_prompt(self) -> None: + """ + Handles the logic for overwriting an existing output file. + Modifies self._ffmpeg_command to include -y if overwrite is confirmed. + Raises FfmpegProcessUserCancelledError or + FfmpegProcessInterruptedError if not proceeding. + """ + try: + answer = ( + input(f"Output file {self._output_filepath} already exists. Overwrite? [y/N]: ") + .strip() + .lower() + ) + if answer == "y": + self._ffmpeg_command.insert(1, _FFMPEG_OVERWRITE_FLAG) + else: + raise FfmpegProcessUserCancelledError( + "FFmpeg process cancelled. Output file exists and overwrite declined." + ) + except KeyboardInterrupt as e: + raise FfmpegProcessInterruptedError( + "[KeyboardInterrupt] FFmpeg process cancelled during overwrite prompt." + ) from e + except EOFError as e: + raise FfmpegProcessInterruptedError( + "Input error (EOF) during overwrite prompt. FFmpeg process cancelled." + ) from e + + def __init__( + self, + command: List[str], + ffmpeg_log_level: Optional[Union[FfmpegLogLevel, str]] = None, + ffmpeg_log_file: Optional[Union[str, os.PathLike]] = None, + print_detected_duration: bool = False, + duration_override: Optional[float] = None, + ): + """ + Create an FFmpeg process with progress reporting. + + Args: + command: + FFmpeg command as a list of arguments. + ffmpeg_log_level: + FFmpeg log level enum or case-insensitive string. + ffmpeg_log_file: + File path or file-like object used for FFmpeg stderr. + print_detected_duration: + Print the duration used for progress calculation. + duration_override: + Specify input file duration in seconds to use for progress calculation, instead of trying to detect this automatically with FFprobe. This is useful for partial, damaged, streamed, or + synthetic inputs whose duration cannot be detected correctly. + """ + validate_ffmpeg_command(command) + + ffmpeg_log_level_val: str + + if ffmpeg_log_level is None: + ffmpeg_log_level_val = FfmpegLogLevel.VERBOSE.value + elif isinstance(ffmpeg_log_level, FfmpegLogLevel): + ffmpeg_log_level_val = ffmpeg_log_level.value + elif isinstance(ffmpeg_log_level, str): + try: + ffmpeg_log_level_val = FfmpegLogLevel(ffmpeg_log_level.lower()).value + except ValueError: + valid_levels = [e.value for e in FfmpegLogLevel] + raise FfmpegCommandError( + f"Invalid ffmpeg_log_level string: " + f"'{ffmpeg_log_level}'. Must be one of " + f"{valid_levels} (case-insensitive)." + ) + else: + raise TypeError( + "ffmpeg_log_level must be an FfmpegLogLevel enum instance, " + f"a string, or None, not " + f"{type(ffmpeg_log_level).__name__}" + ) + + self._ffmpeg_log_level_val = ffmpeg_log_level_val + + input_file_index = command.index("-i") + input_file_path_str = command[input_file_index + 1] + self._input_filepath = Path(input_file_path_str) + + # Assumes last argument is output. + self._output_filepath = Path(command[-1]) + + if ffmpeg_log_file is None: + self._ffmpeg_log_file = Path(f"{self._input_filepath.name}_ffmpeg_log.txt") + elif isinstance(ffmpeg_log_file, (str, os.PathLike)): + self._ffmpeg_log_file = Path(ffmpeg_log_file) + else: + self._ffmpeg_log_file = ffmpeg_log_file + + self._print_detected_duration = print_detected_duration + + if duration_override is not None: + if isinstance(duration_override, bool): + raise TypeError("duration_override must be a number of seconds, not bool") + + try: + duration_value = float(duration_override) + except (TypeError, ValueError) as e: + raise TypeError("duration_override must be a number of seconds or None") from e + + if not math.isfinite(duration_value) or duration_value <= 0: + raise ValueError("duration_override must be a finite number greater than 0") + + self._duration_secs = duration_value + else: + self._duration_secs = get_media_duration(input_file_path_str) + + if duration_override is not None: + print(f"Using duration override: {self._duration_secs:.2f} seconds") + elif self._print_detected_duration: + if self._duration_secs is None: + print("Could not detect duration. Progress bar may not show time remaining.") + else: + print(f"Detected duration: {self._duration_secs:.2f} seconds") + + self._ffmpeg_command = [ + command[0], + "-hide_banner", + "-loglevel", + self._ffmpeg_log_level_val, + "-progress", + "pipe:1", + "-nostats", + ] + self._ffmpeg_command.extend(command[1:]) + + is_overwrite_in_user_command = any(arg == _FFMPEG_OVERWRITE_FLAG for arg in command[1:]) + + if self._output_filepath.exists() and not is_overwrite_in_user_command: + self._handle_overwrite_prompt() + + self._process: Optional[subprocess.Popen] = None + self.use_tqdm: bool = False + + def run( + self, + print_command: bool = False, + ) -> None: + if print_command: + cmd_str = ( + " ".join(self._ffmpeg_command) + if isinstance(self._ffmpeg_command, list) + else self._ffmpeg_command + ) + print(f"Executing: {cmd_str}") + + self._shell_needed = check_shell_needed_for_command( + self._ffmpeg_command + if isinstance(self._ffmpeg_command, list) + else [self._ffmpeg_command] + ) + + current_ffmpeg_command = self._ffmpeg_command + if self._shell_needed and isinstance(current_ffmpeg_command, list): + current_ffmpeg_command = " ".join(current_ffmpeg_command) + + try: + creationflags = 0 + # Windows + if os.name == "nt": + creationflags = subprocess.CREATE_NEW_PROCESS_GROUP + + if isinstance(self._ffmpeg_log_file, Path): + with open( + self._ffmpeg_log_file, + "w", + encoding="utf-8", + ) as f: + self._process = subprocess.Popen( + current_ffmpeg_command, + shell=self._shell_needed, + stdout=subprocess.PIPE, + stderr=f, + creationflags=creationflags, + ) + else: + self._process = subprocess.Popen( + current_ffmpeg_command, + shell=self._shell_needed, + stdout=subprocess.PIPE, + stderr=self._ffmpeg_log_file, + creationflags=creationflags, + ) + except Exception as e: + raise FfmpegProcessError(f"Error starting FFmpeg process: {e}") from e + + try: + if self.use_tqdm: + use_tqdm(self, self._process) + else: + use_rich(self, self._process) + except KeyboardInterrupt: + self._terminate() + finally: + if self._process and self._process.stdout: + self._process.stdout.close() + + def _terminate(self): + if self._process: + terminate_ffmpeg_process(self._process) + else: + exit() diff --git a/progress_bars.py b/progress_bars.py new file mode 100644 index 0000000..0b96d63 --- /dev/null +++ b/progress_bars.py @@ -0,0 +1,166 @@ +import os +import subprocess +from typing import TYPE_CHECKING +from rich.console import Console +from rich.theme import Theme +from rich.progress import ( + Progress, + TextColumn, + BarColumn, + TaskProgressColumn, + TimeRemainingColumn, + TimeElapsedColumn, +) +from tqdm import tqdm + +from .utils import parse_ffmpeg_progress_line +from .exceptions import FfmpegProcessError + +if TYPE_CHECKING: + from .better_ffmpeg_progress import FfmpegProcess + + +def truncate_filename(filename: str, max_length: int = 60) -> str: + """Truncates a filename if it exceeds max_length while preserving the file extension.""" + if len(filename) <= max_length: + return filename + + stem, ext = os.path.splitext(filename) + ellipsis = "..." + allowed_stem_len = max_length - len(ext) - len(ellipsis) + + if allowed_stem_len < 1: + return filename[: max_length - len(ellipsis)] + ellipsis + + return f"{stem[:allowed_stem_len]}{ellipsis}{ext}" + +custom_theme = Theme({ + "progress.elapsed": "bold white", + "progress.remaining": "white", + "progress.percentage": "sea_green3", +}) + +console = Console(theme=custom_theme) + +def use_rich(ffmpeg_process_instance: "FfmpegProcess", process: subprocess.Popen) -> None: + task_id = None + progress_bar_instance = Progress( + TextColumn("[progress.description]{task.description}"), + BarColumn(bar_width=40, style="grey15", complete_style="bold sea_green3", finished_style="bold bright_green"), + TaskProgressColumn(), + TimeElapsedColumn(), + TimeRemainingColumn(compact=True), + refresh_per_second=8, + console=console, + ) + + duration_secs = ffmpeg_process_instance._duration_secs + input_filename = truncate_filename(ffmpeg_process_instance._input_filepath.name, max_length=60) + log_file = ffmpeg_process_instance._ffmpeg_log_file + + with progress_bar_instance as progress_bar: + if duration_secs: + current_size_mb = 0.0 + current_time_secs = 0.0 + estimated_size_mb = 0.0 + + # Line 1: Header/Filename printed separately + console.print(f"Processing {input_filename}") + + # Line 2: Progress metrics and size stats + task_id = progress_bar.add_task( + " (0.00 MB / ~0.00 MB)", + total=duration_secs, + ) + update_progress = progress_bar.update + + for line_bytes in process.stdout: + stripped_line_bytes = line_bytes.strip() + progress_data = parse_ffmpeg_progress_line(stripped_line_bytes, duration_secs) + + if progress_data and task_id is not None: + if "size_mb" in progress_data: + current_size_mb = progress_data["size_mb"] + + if "time_secs" in progress_data: + current_time_secs = progress_data["time_secs"] + + # Calculate dynamic estimated final size + if current_time_secs > 0: + estimated_size_mb = (current_size_mb / current_time_secs) * duration_secs + + update_progress( + task_id, + completed=current_time_secs, + description=f" ({current_size_mb:.2f} MB / ~{estimated_size_mb:.2f} MB)", + ) + else: + print(f"Processing '{input_filename}'...") + + process.wait() + + if process.returncode == 0: + if task_id is not None: + update_progress(task_id, completed=duration_secs) + progress_bar.columns = ( + TextColumn("[progress.description]{task.description}"), + BarColumn(bar_width=40, style="red", complete_style="green", finished_style="bold bright_green"), + TaskProgressColumn(), + ) + update_progress( + task_id, + description=f"✓ Final Size: {current_size_mb:.2f} MB", + ) + +def use_tqdm(ffmpeg_process_instance: "FfmpegProcess", process: subprocess.Popen) -> None: + progress_bar = None + try: + width = os.get_terminal_size().columns + except OSError: + width = 80 + + duration_secs = ffmpeg_process_instance._duration_secs + # Truncate input filename to a max of 60 characters + input_filename = truncate_filename(ffmpeg_process_instance._input_filepath.name, max_length=60) + log_file = ffmpeg_process_instance._ffmpeg_log_file + + if duration_secs: + current_size_mb = 0.0 + progress_bar = tqdm( + mininterval=0.5, + total=duration_secs, + desc=f"Processing '{input_filename}'", + ncols=80, + dynamic_ncols=True if width < 80 else False, + bar_format="{desc} {bar} {percentage:.1f}% [{elapsed}<{remaining}, {rate_fmt}{postfix}]", + ) + + for line_bytes in process.stdout: + stripped_line_bytes = line_bytes.strip() + progress_data = parse_ffmpeg_progress_line(stripped_line_bytes, duration_secs) + + if progress_data and progress_bar is not None: + if "size_mb" in progress_data: + current_size_mb = progress_data["size_mb"] + progress_bar.set_postfix_str(f"{current_size_mb:.2f} MB") + + if "time_secs" in progress_data: + progress_bar.n = progress_data["time_secs"] + progress_bar.refresh() + else: + print(f"Processing '{input_filename}'...") + + process.wait() + + if process.returncode == 0: + if progress_bar: + progress_bar.n = duration_secs + progress_bar.set_description(f"✓ Processed '{input_filename}'") + progress_bar.close() + else: + print(f"✓ Processed '{input_filename}'") + else: + if progress_bar: + progress_bar.close() + + raise FfmpegProcessError(f"FFmpeg process failed. Check '{log_file}' for details.") \ No newline at end of file diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..710936e --- /dev/null +++ b/utils.py @@ -0,0 +1,148 @@ +import shutil +import subprocess +from pathlib import Path +from typing import List, Optional +import requests # Added for URL validation + +from .exceptions import FfmpegCommandError + +_FFMPEG_INPUT_FLAG = "-i" +_OUT_TIME_US_PREFIX = b"out_time_us=" +_TOTAL_SIZE_PREFIX = b"total_size=" +_MULTIPLIER = 1e-6 + + +def validate_ffmpeg_command(command: List[str]) -> None: + """ + Validates the FFmpeg command. + Raises FfmpegCommandError if validation fails. + """ + if not command: + raise FfmpegCommandError("FFmpeg command list cannot be empty.") + + ffmpeg_executable = command[0] + if not shutil.which(ffmpeg_executable): + err_msg = ( + f"'{ffmpeg_executable}' not found. " + "Ensure it's in your system's PATH or provide the full path of the FFmpeg executable." + ) + + raise FfmpegCommandError(err_msg) + + if _FFMPEG_INPUT_FLAG not in command: + raise FfmpegCommandError( + f"FFmpeg command must include the input flag '{_FFMPEG_INPUT_FLAG}'." + ) + + try: + input_flag_pos = command.index(_FFMPEG_INPUT_FLAG) + except ValueError: + raise FfmpegCommandError( + f"Internal error: Input flag '{_FFMPEG_INPUT_FLAG}' not found after check." + ) + + input_file_idx = input_flag_pos + 1 + if input_file_idx >= len(command): + raise FfmpegCommandError(f"No input file specified after '{_FFMPEG_INPUT_FLAG}'.") + + input_file = command[input_file_idx] + if input_file.startswith("-") and len(input_file) == 2 and not input_file[1:].isdigit(): + raise FfmpegCommandError( + f"Input file path '{input_file}' looks like an option flag. Please check your command." + ) + + # Validate input file/URL + if input_file.startswith("http://") or input_file.startswith("https://"): + try: + response = requests.head(input_file, timeout=5, allow_redirects=True) + response.raise_for_status() + except requests.exceptions.RequestException as e: + raise FfmpegCommandError(f"Input URL not accessible: {input_file}. Error: {e}") + elif not Path(input_file).exists(): + raise FfmpegCommandError(f"Input file not found: {input_file}") + + if len(command) <= input_file_idx + 1: + raise FfmpegCommandError( + "No output file specified. The command seems to end after the input file path." + ) + + output_file_path_str = command[-1] + if output_file_path_str == input_file: + raise FfmpegCommandError( + "Output file path cannot be the same as the input file path in this command structure." + ) + + +def get_media_duration(input_file: str) -> Optional[float]: + """ + Retrieves the duration of a media file or URL using ffprobe. + Returns duration in seconds, or None if it cannot be determined. + Accepts input_file as a string to correctly handle URLs. + """ + if not shutil.which("ffprobe"): + print("Warning: ffprobe not found in PATH. Cannot determine media duration.") + return None + try: + output = subprocess.check_output( + [ + "ffprobe", + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "default=noprint_wrappers=1:nokey=1", + "-i", + input_file, + ], + text=True, + stderr=subprocess.DEVNULL, + ) + return float(output) + except (subprocess.CalledProcessError, ValueError, FileNotFoundError): + print(f"Warning: Could not determine duration for '{input_file}'.") + return None + + +def check_shell_needed_for_command(command: List[str]) -> bool: + """ + Checks if any part of the command contains shell operators, + indicating that shell=True might be needed for subprocess.Popen. + This is a basic check and might not cover all edge cases. + """ + shell_operators = {"|", ">", "<", ">>", "&&", "||"} # Basic set + return any(op in item for item in command for op in shell_operators) + + +def parse_ffmpeg_progress_line( + line: bytes, total_duration_secs: Optional[float] +) -> Optional[dict]: + if line.startswith(_OUT_TIME_US_PREFIX): + try: + value_str = line[len(_OUT_TIME_US_PREFIX) :] + if not value_str: + return None + + current_time_us = int(value_str) + current_time_secs = current_time_us * _MULTIPLIER + + if total_duration_secs is not None: + current_time_secs = min(current_time_secs, total_duration_secs) + + return {"time_secs": current_time_secs} + except (ValueError, IndexError): + return None + + if line.startswith(_TOTAL_SIZE_PREFIX): + try: + value_str = line[len(_TOTAL_SIZE_PREFIX) :] + if not value_str: + return None + + size_bytes = int(value_str) + size_mb = size_bytes / (1024 * 1024) + return {"size_mb": size_mb} + except (ValueError, IndexError): + return None + + return None \ No newline at end of file