Add files via upload

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()
-```
+```

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()