ffmpeg-python中文文档（二）——案例

获取视频信息 (ffprobe)

probe = ffmpeg.probe(args.in_filename)
video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None)
width = int(video_stream['width'])
height = int(video_stream['height'])

为视频生成缩略图

获取视频缩略图

(
    ffmpeg
    .input(in_filename, ss=time)
    .filter('scale', width, -1)
    .output(out_filename, vframes=1)
    .run()
)

将视频转换为 numpy 数组

ffmpeg-numpy 图

out, _ = (
    ffmpeg
    .input('in.mp4')
    .output('pipe:', format='rawvideo', pix_fmt='rgb24')
    .run(capture_stdout=True)
)
video = (
    np
    .frombuffer(out, np.uint8)
    .reshape([-1, height, width, 3])
)

通过管道将单个视频帧读取为 jpeg

将帧读取为 jpeg 图

out, _ = (
    ffmpeg
    .input(in_filename)
    .filter('select', 'gte(n,{})'.format(frame_num))
    .output('pipe:', vframes=1, format='image2', vcodec='mjpeg')
    .run(capture_stdout=True)
)

将声音转换为原始 PCM 音频

转录图

out, _ = (ffmpeg
    .input(in_filename, **input_kwargs)
    .output('-', format='s16le', acodec='pcm_s16le', ac=1, ar='16k')
    .overwrite_output()
    .run(capture_stdout=True)
)

从帧序列组装视频

球体

(
    ffmpeg
    .input('/path/to/jpegs/*.jpg', pattern_type='glob', framerate=25)
    .output('movie.mp4')
    .run()
)

使用额外的过滤：

全局过滤器

(
    ffmpeg
    .input('/path/to/jpegs/*.jpg', pattern_type='glob', framerate=25)
    .filter('deflicker', mode='pm', size=10)
    .filter('scale', size='hd1080', force_original_aspect_ratio='increase')
    .output('movie.mp4', crf=20, preset='slower', movflags='faststart', pix_fmt='yuv420p')
    .view(filename='filter_graph')
    .run()
)

音频/视频管道

[图片上传失败...(image-53ab8c-1641985483494)]

in1 = ffmpeg.input('in1.mp4')
in2 = ffmpeg.input('in2.mp4')
v1 = in1.video.hflip()
a1 = in1.audio
v2 = in2.video.filter('reverse').filter('hue', s=0)
a2 = in2.audio.filter('areverse').filter('aphaser')
joined = ffmpeg.concat(v1, a1, v2, a2, v=1, a=1).node
v3 = joined[0]
a3 = joined[1].filter('volume', 0.8)
out = ffmpeg.output(v3, a3, 'out.mp4')
out.run()

带有偏移和视频的单声道到立体声

单立体图

audio_left = (
    ffmpeg
    .input('audio-left.wav')
    .filter('atrim', start=5)
    .filter('asetpts', 'PTS-STARTPTS')
)

audio_right = (
    ffmpeg
    .input('audio-right.wav')
    .filter('atrim', start=10)
    .filter('asetpts', 'PTS-STARTPTS')
)

input_video = ffmpeg.input('input-video.mp4')

(
    ffmpeg
    .filter((audio_left, audio_right), 'join', inputs=2, channel_layout='stereo')
    .output(input_video.video, 'output-video.mp4', shortest=None, vcodec='copy')
    .overwrite_output()
    .run()
)

Jupyter 帧查看器

[图片上传失败...(image-120abf-1641985483494)]

Jupyter 流编辑器

jupyter 演示

TensorFlow 流式传输

张量流； 挑战模式：将其与下面的网络摄像头示例相结合

• 使用 ffmpeg 解码输入视频
• 使用“深度梦想”示例使用 tensorflow 处理视频
• 使用 ffmpeg 编码输出视频

process1 = (
    ffmpeg
    .input(in_filename)
    .output('pipe:', format='rawvideo', pix_fmt='rgb24', vframes=8)
    .run_async(pipe_stdout=True)
)

process2 = (
    ffmpeg
    .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height))
    .output(out_filename, pix_fmt='yuv420p')
    .overwrite_output()
    .run_async(pipe_stdin=True)
)

while True:
    in_bytes = process1.stdout.read(width * height * 3)
    if not in_bytes:
        break
    in_frame = (
        np
        .frombuffer(in_bytes, np.uint8)
        .reshape([height, width, 3])
    )

    # 见例子/tensorflow_stream.py:
    out_frame = deep_dream.process_frame(in_frame)

    process2.stdin.write(
        out_frame
        .astype(np.uint8)
        .tobytes()
    )

process2.stdin.close()
process1.wait()
process2.wait()

深梦流

FaceTime 网络摄像头输入 (OS X)

(
    ffmpeg
    .input('FaceTime', format='avfoundation', pix_fmt='uyvy422', framerate=30)
    .output('out.mp4', pix_fmt='yuv420p', vframes=100)
    .run()
)

从本地视频流式传输到 HTTP 服务器

video_format = "flv"
server_url = "http://127.0.0.1:8080"

process = (
    ffmpeg
    .input("input.mp4")
    .output(
        server_url, 
        codec = "copy", # 使用与原始视频相同的编解码器
        listen=1, # 启用 HTTP 服务器
        f=video_format)
    .global_args("-re") # 参数充当直播流
    .run()
)

要接收视频，您可以在终端中使用 ffplay：

$ ffplay -f flv http://localhost:8080

从 RTSP 服务器流式传输到 TCP 套接字

packet_size = 4096

process = (
    ffmpeg
    .input('rtsp://%s:8554/default')
    .output('-', format='h264')
    .run_async(pipe_stdout=True)
)

while process.poll() is None:
    packet = process.stdout.read(packet_size)
    try:
        tcp_socket.send(packet)
    except socket.error:
        process.stdout.close()
        process.wait()
        break