Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 146 additions & 18 deletions ffpb.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
# OR OTHER DEALINGS IN THE SOFTWARE.

"""A progress bar for `ffmpeg` using `tqdm`.
"""A colored progress bar for `ffmpeg` using `tqdm`.
"""

from __future__ import unicode_literals
Expand All @@ -44,12 +44,49 @@
from tqdm import tqdm


class ProgressNotifier(object):

_DURATION_RX = re.compile(b"Duration: (\d{2}):(\d{2}):(\d{2})\.\d{2}")
_PROGRESS_RX = re.compile(b"time=(\d{2}):(\d{2}):(\d{2})\.\d{2}")
_SOURCE_RX = re.compile(b"from '(.*)':")
_FPS_RX = re.compile(b"(\d{2}\.\d{2}|\d{2}) fps")
class Colors:
"""ANSI color codes for terminal output"""
RESET = '\033[0m'
BOLD = '\033[1m'
DIM = '\033[2m'

# Standard colors
BLACK = '\033[30m'
RED = '\033[31m'
GREEN = '\033[32m'
YELLOW = '\033[33m'
BLUE = '\033[34m'
MAGENTA = '\033[35m'
CYAN = '\033[36m'
WHITE = '\033[37m'

# Bright colors
BRIGHT_BLACK = '\033[90m'
BRIGHT_RED = '\033[91m'
BRIGHT_GREEN = '\033[92m'
BRIGHT_YELLOW = '\033[93m'
BRIGHT_BLUE = '\033[94m'
BRIGHT_MAGENTA = '\033[95m'
BRIGHT_CYAN = '\033[96m'
BRIGHT_WHITE = '\033[97m'

# Background colors
BG_BLACK = '\033[40m'
BG_RED = '\033[41m'
BG_GREEN = '\033[42m'
BG_YELLOW = '\033[43m'
BG_BLUE = '\033[44m'
BG_MAGENTA = '\033[45m'
BG_CYAN = '\033[46m'
BG_WHITE = '\033[47m'


class ColoredProgressNotifier(object):

_DURATION_RX = re.compile(rb"Duration: (\d{2}):(\d{2}):(\d{2})\.\d{2}")
_PROGRESS_RX = re.compile(rb"time=(\d{2}):(\d{2}):(\d{2})\.\d{2}")
_SOURCE_RX = re.compile(rb"from '(.*)':")
_FPS_RX = re.compile(rb"(\d{2}\.\d{2}|\d{2}) fps")

@staticmethod
def _seconds(hours, minutes, seconds):
Expand All @@ -62,7 +99,7 @@ def __exit__(self, exc_type, exc_value, traceback):
if self.pbar is not None:
self.pbar.close()

def __init__(self, file=None, encoding=None, tqdm=tqdm):
def __init__(self, file=None, encoding=None, tqdm=tqdm, use_colors=True):
self.lines = []
self.line_acc = bytearray()
self.duration = None
Expand All @@ -73,8 +110,55 @@ def __init__(self, file=None, encoding=None, tqdm=tqdm):
self.file = file or sys.stderr
self.encoding = encoding or locale.getpreferredencoding() or 'UTF-8'
self.tqdm = tqdm
self.use_colors = use_colors and self._supports_color()
self.colors = Colors() if self.use_colors else None

def _supports_color(self):
"""Check if terminal supports color output"""
if os.name == 'nt':
# Enable color support on Windows 10+
try:
import colorama
colorama.init()
return True
except ImportError:
# Try to enable ANSI escape sequences on Windows
try:
import ctypes
kernel32 = ctypes.windll.kernel32
kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7)
return True
except:
return False
else:
# Unix-like systems
return hasattr(self.file, 'isatty') and self.file.isatty()

def _colorize_filename(self, filename):
"""Apply color to filename, trimming to 30 characters if needed"""
if len(filename) > 30:
filename = filename[:27] + "..." # Trim to 27 chars + add "..." (total 30)
if not self.use_colors:
return filename
return f"{self.colors.BRIGHT_CYAN}{self.colors.BOLD}{filename}{self.colors.RESET}"

def _get_progress_color(self, percentage):
"""Get color based on progress percentage"""
if not self.use_colors:
return ""

if percentage < 25:
return self.colors.BRIGHT_RED
elif percentage < 50:
return self.colors.BRIGHT_YELLOW
elif percentage < 75:
return self.colors.BRIGHT_BLUE
elif percentage < 95:
return self.colors.BRIGHT_GREEN
else:
return self.colors.GREEN + self.colors.BOLD

def __call__(self, char, stdin = None):
def __call__(self, char, stdin=None):
if isinstance(char, unicode):
char = char.encode('ascii')
if char in b"\r\n":
Expand All @@ -89,7 +173,13 @@ def __call__(self, char, stdin = None):
else:
self.line_acc.extend(char)
if self.line_acc[-6:] == bytearray(b"[y/N] "):
print(self.line_acc.decode(self.encoding), end="", file=self.file)
prompt_text = self.line_acc.decode(self.encoding)
if self.use_colors:
# Color the prompt
colored_prompt = f"{self.colors.BRIGHT_YELLOW}{self.colors.BOLD}{prompt_text}{self.colors.RESET}"
print(colored_prompt, end="", file=self.file)
else:
print(prompt_text, end="", file=self.file)
self.file.flush()
if stdin:
stdin.put(input() + "\n")
Expand Down Expand Up @@ -134,23 +224,51 @@ def progress(self, line):
total *= self.fps

if self.pbar is None:
# Create colored description
desc = self.source
if self.use_colors and desc:
desc = self._colorize_filename(desc)

# Create progress bar with custom format
bar_format = None
if self.use_colors:
bar_format = (
f'{self.colors.BRIGHT_WHITE}{{desc}}: '
f'{self.colors.BRIGHT_GREEN}{{percentage:3.0f}}%'
f'{self.colors.RESET}|{self.colors.BRIGHT_BLUE}{{bar}}{self.colors.RESET}| '
f'{self.colors.WHITE}{{n_fmt}}/{{total_fmt}}'
f'{self.colors.BRIGHT_YELLOW} [{{elapsed}}<{{remaining}}, {{rate_fmt}}]'
f'{self.colors.RESET}'
)

self.pbar = self.tqdm(
desc=self.source,
desc=desc,
file=self.file,
total=total,
dynamic_ncols=True,
unit=unit,
ncols=0,
ascii=os.name=="nt", # windows cmd has problems with unicode
#ascii=os.name == "nt" and not self.use_colors, # use unicode if colors are supported
ascii=' ▒',
bar_format=bar_format,
colour='green' if self.use_colors else None,
)

# Update progress bar with color changes based on percentage
if total and self.use_colors:
percentage = (current / total) * 100
color = self._get_progress_color(percentage)
# Update the bar color dynamically
self.pbar.colour = 'green'

self.pbar.update(current - self.pbar.n)

def main(argv=None, stream=sys.stderr, encoding=None, tqdm=tqdm):

def main(argv=None, stream=sys.stderr, encoding=None, tqdm=tqdm, use_colors=True):
argv = argv or sys.argv[1:]

try:
with ProgressNotifier(file=stream, encoding=encoding, tqdm=tqdm) as notifier:
with ColoredProgressNotifier(file=stream, encoding=encoding, tqdm=tqdm, use_colors=use_colors) as notifier:

cmd = ["ffmpeg"] + argv
p = subprocess.Popen(cmd, stderr=subprocess.PIPE)
Expand All @@ -163,18 +281,28 @@ def main(argv=None, stream=sys.stderr, encoding=None, tqdm=tqdm):
notifier(out)

except KeyboardInterrupt:
print("Exiting.", file=stream)
if use_colors and hasattr(stream, 'isatty') and stream.isatty():
print(f"{Colors.BRIGHT_RED}{Colors.BOLD}Exiting.{Colors.RESET}", file=stream)
else:
print("Exiting.", file=stream)
return signal.SIGINT + 128 # POSIX standard

except Exception as err:
print("Unexpected exception:", err, file=stream)
if use_colors and hasattr(stream, 'isatty') and stream.isatty():
print(f"{Colors.BRIGHT_RED}Unexpected exception: {Colors.BRIGHT_WHITE}{err}{Colors.RESET}", file=stream)
else:
print("Unexpected exception:", err, file=stream)
return 1

else:
if p.returncode != 0:
print(notifier.lines[-1].decode(notifier.encoding), file=stream)
error_msg = notifier.lines[-1].decode(notifier.encoding)
if use_colors and hasattr(stream, 'isatty') and stream.isatty():
print(f"{Colors.BRIGHT_RED}{error_msg}{Colors.RESET}", file=stream)
else:
print(error_msg, file=stream)
return p.returncode


if __name__ == "__main__":
sys.exit(main())
sys.exit(main())