python subprocess启动多个ffmpeg shell?

Question

我需要依次调用三次subprocess. 发现最后一个没有执行? 在流媒体服务器的控制台上只能看到: aac和mp4的rtmp信息, 看不到live的rtmp连接
参考代码:

if __name__ == "__main__":
    device: str = "cuda:0" if torch.cuda.is_available() else "cpu"
    model, tokenizer = loadModel(device=device)
    #
    # 音频(aac)
    #
    audioProcess = sp.Popen(["ffmpeg", 
                             "-f", "s16le", 
                             '-y', '-vn',
                             "-ac", "1",
                             "-ar", "16000",
                             "-channel_layout", "mono",
                             '-acodec','pcm_s16le',
                             "-i", "pipe:",
                             "-ar", "44100",
                             "-f", "flv",
                             rtmpAURL], stdin=sp.PIPE)
    
    #
    # 视频(mp4)
    #
    vedioProcess = sp.Popen(["ffmpeg",
                             "-re",
                             '-y', '-an',
                             '-rtbufsize', '1024M',
                             "-f", "dshow",
                             "-i", "video=Q8 HD Webcam",
                             "-pix_fmt", "yuvj420p",
                             "-framerate", "25",
                             '-vcodec', 'libx264', 
                             '-preset', 'fast', 
                             '-crf', '25',
                             "-vf", "scale=640:480",
                             "-f", "flv", rtmpVURL], shell=True)
    
    #
    #https://stackoverflow.com/questions/18618191/ffmpeg-merge-multiple-rtmp-stream-inputs-to-a-single-rtmp-output
    #ffmpeg -i rtmp://ip:1935/live/micMyStream7 -i rtmp://ip:1935/live/MyStream7 -filter_complex "[0:a][1:a]amix[a]" -map 0:v -map "[a]" -c:v copy -f flv rtmp://ip:1935/live/bcove7
    #
    # 合并(live)
    # 
    mergeVAP = sp.Popen(["ffmpeg", 
                                "-i", rtmpAURL,
                                "-i", rtmpVURL,
                                "-c:v", "copy",
                                "-c:a", "aac",
                                "-preset", "veryfast",
                                "-f", "flv",
                                "-flvflags", "no_duration_filesize",
                                rtmpURL], shell=True, stderr=True)
    

    writeAudioThead = Thread(target=soundRecorder, args=(model, device, tokenizer, audioProcess))
    writeAudioThead.start()
    writeAudioThead.join()

    audioProcess.wait()
    vedioProcess.wait()
    mergeVAP.wait()

问题1：这三个是串行执行的. 如何让他们异步执行
问题2: 上面的示例串行执行, 为什么 mergeVAP没有执行.在命令行上执行是可以执行的

Answer 1

捕获了一下合并流的subprocess发现它启动时音频流不可用,所以才无法正常通行. 在音频流可用的第一时间调用封装了合并流的函数都可以正常运行了. 示例代码:

if __name__ == "__main__":
    device: str = "cuda:0" if torch.cuda.is_available() else "cpu"
    model, tokenizer = loadModel(device=device)
    #
    # 音频
    #
    audioProcess = sp.Popen(["ffmpeg", 
                             "-f", "s16le", 
                             "-y", "-vn",
                             "-ac", "1",
                             "-ar", "16000",
                             "-channel_layout", "mono",
                             "-acodec","pcm_s16le",
                             "-i", "pipe:",
                             "-ar", "44100",
                             "-f", "flv",
                             rtmpAURL], stdin=sp.PIPE)
    writeAudioThead = Thread(target=soundRecorder, args=(model, device, tokenizer, audioProcess))
    writeAudioThead.start()
    #writeAudioThead.join()
    #
    # 视频
    #
    vedioProcess = sp.Popen(["ffmpeg",
                             "-re",
                             "-y", "-an",
                             "-f", "dshow",
                             "-i", "video=Q8 HD Webcam",
                             "-pix_fmt", "yuvj420p",
                             "-framerate", "25",
                             "-vcodec", "libx264", 
                             "-x264-params", "nal-hrd=cbr",
                             "-minrate", "1M",
                             "-maxrate", "1M",
                             "-bufsize", "2M",
                             "-preset", "fast", 
                             "-crf", "25",
                             "-vf", "scale=640:480",
                             "-f", "flv", rtmpVURL], shell=True)
    
    #
    #https://stackoverflow.com/questions/18618191/ffmpeg-merge-multiple-rtmp-stream-inputs-to-a-single-rtmp-output
    #ffmpeg -i rtmp://ip:1935/live/micMyStream7 -i rtmp://ip:1935/live/MyStream7 -filter_complex "[0:a][1:a]amix[a]" -map 0:v -map "[a]" -c:v copy -f flv rtmp://ip:1935/live/bcove7
    #
    # 合并
    # 
    isFirst = False
    while True:
        if isFirst:
            mergeVA()
            break
    # 
    #
    audioProcess.wait()
    vedioProcess.wait()
    #mergeVAP.wait(60)
    #

在soundRecorder方法中将全局的状态(isFirst)切换为true以触发合并流的调用. 本来想用signal发现windows中不支持，会报错:AttributeError: module 'signal' has no attribute 'alarm'

def soundRecorder(model, device, tokenizer, audioPipProcesss):
    global isFirst
    #
    filepath = 'oldmansea.txt'
    for content in read_file(filepath):
        print(content)
        audioPipWriter(generte(prompt=content, device=device, model=model, tokenizer=tokenizer), audioPipProcesss=audioPipProcesss)

        if not isFirst:
            isFirst = True

        time.sleep(2)

    audioPipProcesss.stdin.close() 

mergeVA的方法代码如下：

def mergeVA():
    print('------------->start vedio and autio')
    #time.sleep(30)
    try:
        mergeVAP = sp.Popen(["ffmpeg", 
                                "-rw_timeout", "5000000",
                                "-rtbufsize", "1024M",
                                "-i", rtmpAURL,
                                "-i", rtmpVURL,
                                "-c:v", "copy",
                                "-c:a", "aac",
                                "-preset", "veryfast",
                                "-f", "flv",
                                "-flvflags", "no_duration_filesize",
                                rtmpURL], stderr=sp.PIPE, shell=True)
        out, err = mergeVAP.communicate()
        print('merge vedio&audio has error:%s' % err)
        print(out)
    except sp.CalledProcessError as error:
        print('ffmpeg exited with return code ' + str(error.returncode))
        print(f'**ERROR** An error occurred while converting the file')

    mergeVAP.wait()

当前方案的采集和推流方法实时性欠缺. 希望还有提升的空间