Skip to content

Instantly share code, notes, and snippets.

@slhck
Last active July 10, 2023 11:29
Show Gist options
  • Star 2 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save slhck/702c68069ff45f9f125ee9aaa0a54ceb to your computer and use it in GitHub Desktop.
Save slhck/702c68069ff45f9f125ee9aaa0a54ceb to your computer and use it in GitHub Desktop.
Run an ffmpeg command and get the progress in percent
#!/usr/bin/env python3
#
# Run an ffmpeg command with a progress iterator
# Author: Werner Robitza
# Based on: https://gist.github.com/Hellowlol/5f8545e999259b4371c91ac223409209
# For a library version, see https://pypi.org/project/ffmpeg-progress-yield/
# License: MIT
import subprocess
import re
from typing import Iterator
DUR_REGEX = re.compile(
r"Duration: (?P<hour>\d{2}):(?P<min>\d{2}):(?P<sec>\d{2})\.(?P<ms>\d{2})"
)
TIME_REGEX = re.compile(
r"out_time=(?P<hour>\d{2}):(?P<min>\d{2}):(?P<sec>\d{2})\.(?P<ms>\d{2})"
)
def to_ms(s=None, des=None, **kwargs) -> float:
if s:
hour = int(s[0:2])
minute = int(s[3:5])
sec = int(s[6:8])
ms = int(s[10:11])
else:
hour = int(kwargs.get("hour", 0))
minute = int(kwargs.get("min", 0))
sec = int(kwargs.get("sec", 0))
ms = int(kwargs.get("ms", 0))
result = (hour * 60 * 60 * 1000) + (minute * 60 * 1000) + (sec * 1000) + ms
if des and isinstance(des, int):
return round(result, des)
return result
def run_ffmpeg_command(cmd: "list[str]") -> Iterator[int]:
"""
Run an ffmpeg command, trying to capture the process output and calculate
the duration / progress.
Yields the progress in percent.
"""
total_dur = None
cmd_with_progress = [cmd[0]] + ["-progress", "-", "-nostats"] + cmd[1:]
stderr = []
p = subprocess.Popen(
cmd_with_progress,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=False,
)
while True:
line = p.stdout.readline().decode("utf8", errors="replace").strip()
if line == "" and p.poll() is not None:
break
stderr.append(line.strip())
if not total_dur and DUR_REGEX.search(line):
total_dur = DUR_REGEX.search(line).groupdict()
total_dur = to_ms(**total_dur)
continue
if total_dur:
result = TIME_REGEX.search(line)
if result:
elapsed_time = to_ms(**result.groupdict())
yield int(elapsed_time / total_dur * 100)
if p.returncode != 0:
raise RuntimeError(
"Error running command {}: {}".format(cmd, str("\n".join(stderr)))
)
yield 100
@slhck
Copy link
Author

slhck commented Aug 2, 2022

Note that this code has been superseded by this package: https://pypi.org/project/ffmpeg-progress-yield/

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment