FFMpeg + Python
About me
About me
Trimming a video file
ffmpeg -hide_banner -i video.mp4 \
-ss 00:00:30 -t 10 \
-y video_trim.mp4
Edge detection
Original video
Edge detection using ffmpeg
ffprobe
ffprobe -hide_banner video2.mp4
Input #0, mov,mp4,m4a,3gp,3g2,mj2, from 'video2.mp4':
Metadata:
major_brand : isom
minor_version : 512
compatible_brands: isomiso2avc1mp41
encoder : Lavf58.76.100
Duration: 00:00:07.92, start: 0.000000, bitrate: 2839 kb/s
Stream #0:0(jpn):
Video: h264 (High 10) (avc1 / 0x31637661),
yuv420p10le(tv, bt709),
1920x1080 (SAR 1:1 DAR 16:9),
2513 kb/s, 23.98 fps,
23.98 tbr, 24k tbn, 47.95 tbc (default)
Metadata:
handler_name : Video
vendor_id : [0][0][0][0]
Stream #0:1(jpn):
Audio: aac (LC) (mp4a / 0x6134706D),
48000 Hz, 5.1, fltp, 342 kb/s (default)
Metadata:
handler_name : Audio
vendor_id : [0][0][0][0]
Complex filtering
- Generate an RGB color histogram from the source video
- Run edge detection on the source video
- Place the results of the above on top of the source video as an overlay
Complex filtering
ffmpeg -hide_banner -i viofo_ph.mp4 \
-filter_complex \
"[0]format=gbrp,histogram=display_mode=stack[hist];\
[hist]scale=iw*2:ih[hist_scaled];\
[0]edgedetect,scale=iw/4:ih/4[edges];\
[0][hist_scaled]overlay[hist_over];\
[hist_over][edges]overlay=x=800:y=0" \
-y outputs/histogram.mp4
graph TD
video.mp4 --> 0
0 --> |format=gbrp,histogram=display_mode=stack| hist
hist -->|scale=iw*2:ih| hist_scaled
0 -->|edgedetect,scale=iw/4:ih/4| edges
0 -->|overlay - arg1| hist_over
hist_scaled -->|overlay - arg2| hist_over
hist_over --> out[outputs/histogram.mp4]
edges -->|overlay=x=800:y=0| out[outputs/histogram.mp4]
ffmpeg-python
ffmpeg-python
- https://github.com/kkroening/ffmpeg-python
- A convinient binding for ffmpeg.
- Generates ffmpeg arguments from your Python code.
- Provides helper functions for running ffmpeg in a subprocess.
- Includes a
ffmpeg.probe
function which returnsffprobe
results as a Pythondict
Complex filtering in Python
import ffmpeg
input_video = ffmpeg.input("viofo_ph.mp4")
hist = (
input_video
.filter("format", "gbrp")
.filter("histogram", display_mode="stack")
.filter("scale", "iw*2", "ih")
)
edges = (
input_video
.filter("edgedetect")
.filter("scale", "iw/4", "ih/4")
)
hist_overlay = input_video.overlay(hist)
out_stream = hist_overlay.overlay(edges, x=800, y=0)
out_stream.output(
f"outputs/histogram_python.mp4"
).overwrite_output().run()
Generated ffmpeg call
ffmpeg -i viofo_ph.mp4 -filter_complex \
"[0]format=gbrp[s0];\
[s0]histogram=display_mode=stack[s1];\
[s1]scale=iw*2:ih[s2];\
[0][s2]overlay=eof_action=repeat[s3];\
[0]edgedetect[s4];[s4]scale=iw/4:ih/4[s5];\
[s3][s5]overlay=eof_action=repeat:x=800:y=0[s6]" \
-map [s6] outputs/histogram_python.mp4
FFMpeg + Python + OpenCV
Original video
FFMpeg + Python + OpenCV
graph TD
subgraph ffmpeg process #1
a[outputs/registration_plates.mp4] --> b[decompressed frames]
end
subgraph Python
b -->|stdout, deserialisation| c[numpy array]
c -->|cv2 + cv2.CascadeClassifier| d[find registration plates, mark them]
end
subgraph ffmpeg process #2
d -->|stdin, serialisation| e[raw frames]
e --> f[outputs/registration_plates_marked.mp4]
end
FFMpeg + Python + OpenCV
from typing import Sequence
import ffmpeg
import numpy as np
import cv2
INPUT_FILE = "viofo.mp4"
OUTPUT_FILE = "outputs/viofo_marked.mp4"
def mark_plates(frame: cv2.typing.MatLike,
found_plates: Sequence[cv2.typing.Rect]):
y_offset = video_height // 20
x_offset = video_width // 20
for plate in found_plates:
x, y, width, height = plate
registration_plate = frame[
y : y + height, x : x + width
]
new_height = height * 20
new_width = width * 20
big_plate = cv2.resize(registration_plate,
(new_width, new_height))
frame[
y_offset : y_offset + new_height,
x_offset : x_offset + new_width
] = big_plate
cv2.rectangle(
frame, (x, y),
(x + width, y + height), (0, 255, 0), 5
)
y_offset += new_height + 10
video_width, video_height, framerate = next(
(
(
s["width"],
s["height"],
int(s["r_frame_rate"].split("/")[0])
) for s in ffmpeg.probe(INPUT_FILE)["streams"]
if s["codec_type"] == "video"
)
)
process1 = (
ffmpeg.input(INPUT_FILE)
.video.filter("fps", framerate)
.output("pipe:", format="rawvideo", pix_fmt="bgr24")
.run_async(pipe_stdout=True)
)
process2 = (
ffmpeg.input(
"pipe:",
format="rawvideo",
pix_fmt="bgr24",
s=f"{video_width}x{video_height}",
framerate=framerate // 4,
)
.output(OUTPUT_FILE, pix_fmt="yuv420p")
.overwrite_output()
.run_async(pipe_stdin=True)
)
classifier = cv2.CascadeClassifier(
"haarcascade_plate_number.xml"
)
while in_bytes := process1.stdout.read(
video_width * video_height * 3):
in_frame = (
np.frombuffer(in_bytes, np.uint8)
.reshape([video_height, video_width, 3])
.astype(np.uint8)
)
frame = in_frame.copy()
found_plates = classifier.detectMultiScale(
frame, minNeighbors=4, scaleFactor=1.05, minSize=(50, 20), maxSize=(100, 40)
)
mark_plates(frame, found_plates)
process2.stdin.write(frame.astype(np.uint8).tobytes())
process2.stdin.close()
process1.wait()
process2.wait()
Testing
- For simple processing, just use
ffmpeg.probe()
- Use
ffmpeg.compile()
to retrieve the generated command. - The FFMpeg maintainers provide a diverse library of multimedia samples at fate-suite.ffmpeg.org
Conclusions
- FFmpeg is very powerful and complex -
ffmpeg -h full
- When using it for complex multimedia processing it's better to use a good binding, like
ffmpeg-python
. - Basic operations are (przycinanie materiałów, konwersja napisów, kompresja video, ...)
Thank you!
https://github.com/mRokita/pywaw-ffmpeg https://mrokita.github.io/pywaw-ffmpeg/index.htmlBonus - tracking progress
import socket
from threading import Thread
from contextlib import contextmanager
import tempfile
import ffmpeg
from pathlib import Path
from tqdm import tqdm
@contextmanager
def open_progress_socket():
with tempfile.TemporaryDirectory() as tempdir:
sock = socket.socket(
socket.AF_UNIX, socket.SOCK_STREAM)
socket_filename = Path(tempdir) / "progress.sock"
sock.bind(str(socket_filename))
sock.settimeout(15)
try:
sock.listen(1)
yield sock, socket_filename
finally:
sock.close()
def watch_progress(sock: socket.socket, total_duration: int):
conn, _ = sock.accept()
abs_progress = 0
with tqdm(total=total_duration) as pbar:
while rec := conn.recv(4096):
out_time_us = dict(
map(
lambda x: x.split("="),
rec.decode("utf-8").split("\n")[:-1],
)
)["out_time_us"]
processed_seconds = round(
int(out_time_us) / 1000000
)
delta = processed_seconds - abs_progress
pbar.update(delta)
abs_progress = processed_seconds
with open_progress_socket() as (sock, socket_filename):
total_duration = float(
ffmpeg.probe(
"outputs/viofo_marked.mp4"
)["streams"][0]["duration"]
)
print(f"Total duration: {total_duration}")
t = Thread(target=watch_progress, args=(sock, total_duration))
t.start()
(
ffmpeg.input("outputs/viofo_marked.mp4")
.output(
"outputs/viofo_marked_h265.mp4",
vcodec="hevc"
)
.overwrite_output()
.global_args(
"-progress",
f"unix://{socket_filename}"
)
).run(quiet=True)
t.join()