使用 ffmpeg-python+命名管道进行图片转视频或推流_python ffmpeg推流

命名管道（Named Pipe），也被称为FIFO，是一种在UNIX、Linux和类Unix系统中用于实现进程间通信（IPC）的机制。在Python中，我们可以使用os模块来创建和操作命名管道。

命名管道实际上是个特殊的文件，需要先创建

# 创建命名管道
os.mkfifo(pipe_name)

读写前后需要打开关闭

# 打开备写
pipeout = os.open(pipe_name, os.O_WRONLY)
 
# 写入数据
os.write(pipeout, framedata)
 
# 关闭管道
os.close(pipeout)

ffmpeg从命名管道输入源的方法与普通文件输入类似

# 从管道输入输入rawvideo
video_in = ffmpeg.input(PIPE_VIDEO_PATH, format='rawvideo', pix_fmt='bgr24', s="{}x{}".format(VIDEO_WIDTH, VIDEO_HEIGHT))

完整代码，推流或生成视频仅需输入目标rtmp：

# coding: utf-8
import os
import time
import cv2
import ffmpeg
 
VIEW_PATH = 'view'
VIDEO_WIDTH = 1280
VIDEO_HEIGHT = 720
 
PIPE_VIDEO_PATH = '/tmp/pipe'
try:
    # 创建命名管道
    os.mkfifo( PIPE_VIDEO_PATH )
except OSError:
    print("mkfifo error:", OSError)
 
def main(rtmp='test.flv'):
    global VIDEO_HEIGHT
    global VIDEO_WIDTH
    filepath = os.path.join(VIEW_PATH, 'p640x1.jpg')
    src_img=cv2.imread(filepath)
    height, width, channels = src_img.shape
    VIDEO_HEIGHT = height
    VIDEO_WIDTH = width
    print('INFO', "height, width, channels : {} {} {}".format(height, width, channels))
    process_stdin = ffmpeg_process(rtmp)
    pipeout = os.open(PIPE_VIDEO_PATH, os.O_WRONLY) #打开命名管道准备写入
    s = time.time()
    count = 0
    while True:
        if count >= 10*25:  #目标视频时长10秒
            break
        os.write(pipeout, src_img.tobytes())
        count+=1
    os.close(pipeout)
    process_stdin.stdin.close()
    process_stdin.wait()
    print(time.time()-s)
    return 0
 
 
def ffmpeg_process(rtmp):
    # 从管道输入输入rawvideo
    video_in = ffmpeg.input(PIPE_VIDEO_PATH, format='rawvideo', pix_fmt='bgr24', s="{}x{}".format(VIDEO_WIDTH, VIDEO_HEIGHT))
    process_stdin = (
        ffmpeg
        .output(
                video_in,
                rtmp,
                #loglevel='quiet',
                vcodec='libx264',
                #acodec='aac',
                threads='3',
                # ac='2',
                # ar='44100',
                # ab='320k',
                preset='ultrafast',
                tune='zerolatency',
                f='flv'
                )
        .overwrite_output()
        #.run_async(cmd=["ffmpeg", "-re"],pipe_stdin=True)  #推流
        .run_async(cmd=["ffmpeg"],pipe_stdin=True)      #生成视频不用-re
    )
    return process_stdin
 
if __name__ == '__main__':
    main()