如何解决ffmpeg 中的多个命名管道
这个问题是this question
的后续在我的应用程序中,我想修改各种 mp3,然后将它们混合在一起。我知道我可以在 FFmpeg 中使用单个命令行来完成,但最终可能会非常混乱,因为我需要对每个样本使用各种过滤器,而我有五个。我的想法是单独编辑每个样本,将它们保存到管道中,最后将它们混合。
import subprocess
import os
def create_pipes():
os.mkfifo("pipe1")
os.mkfifo("pipe2")
def create_samp():
sample= subprocess.run(["ffmpeg","-i","https://freesound.org/data/previews/186/186942_2594536-hq.mp3",\
"-af","adelay=15000|15000","-f","mp3","pipe:pipe1"],stdout=subprocess.PIPE).stdout
return(sample)
def create_samp_2():
sample= subprocess.run(["ffmpeg","https://freesound.org/data/previews/370/370934_6399962-lq.ogg","adelay=1000|1000","pipe:pipe2"],stdout=subprocess.PIPE).stdout
return(sample)
def record(samp,samp_2):
process = subprocess.Popen(["ffmpeg","-y",'-f','mp3',\
"-i","https://cdns-preview-b.dzcdn.net/stream/c-b0b684fe962f93dc43f1f7ea493683a1-3.mp3","pipe1","pipe2",\
"-filter_complex","amix=inputs=3:duration=longest","output.mp3"],stdin=subprocess.PIPE)
process.stdin.write(samp)
process.stdin.write(samp_2)
process.stdin.close()
process.wait()
create_pipes()
samp = create_samp()
samp_2 = create_samp_2()
record(samp,samp_2)
当我运行脚本时,create_samp()
和 create_samp2()
运行良好。但是我运行 record()
,程序卡住了,没有错误消息,所以我无法弄清楚问题是什么。
解决方法
使用命名管道(仅限 Linux):
当有两个或更多输入流(需要从内存缓冲区通过管道传输)时,需要命名管道。
使用命名管道一点也不简单...
从 FFmpeg 的角度来看,命名管道就像(不可搜索的)输入文件。
在 Python 中使用命名管道(在 Linux 中):
假设 pipe1
是“命名管道”的名称(例如 pipe1 = "audio_pipe1"
)。
-
创建一个“命名管道”:
os.mkfifo(pipe1)
-
将管道作为“只写”文件打开:
fd_pipe = os.open(pipe_name,os.O_WRONLY) # fd_pipe1 is a file descriptor (an integer).
-
将数据以小块的形式写入管道。
根据 this 帖子,大多数 Linux 系统中管道的默认缓冲区大小为 64 KB。
由于数据大于65536字节,我们需要将数据以小块的形式写入管道。
我决定使用 1024 字节的任意块大小。
管道写操作是一种“阻塞”操作。
我通过使用“作家”线程解决了这个问题:def writer(data,pipe_name,chunk_size): # Open the pipes as opening "low level IO" files (open for "open for writing only"). fd_pipe = os.open(pipe_name,os.O_WRONLY) # fd_pipe1 is a file descriptor (an integer) for i in range(0,len(data),chunk_size): # Write to named pipe as writing to a "low level IO" file (but write the data in small chunks). os.write(fd_pipe,data[i:chunk_size+i]) # Write 1024 bytes of data to fd_pipe
-
关闭管道:
os.close(fd_pipe)
-
移除(取消链接)命名管道:
os.unlink(pipe1)
这是来自 previous post 的示例,使用两个命名管道:
import subprocess
import os
from threading import Thread
def create_samp():
# Read audio stream from https://freesound.org/data/previews/186/186942_2594536-hq.mp3
# Apply adelay audio filter.
# Encode the audio in mp3 format.
# FFmpeg output is passed to stdout pipe,and stored in sample bytes array.
sample1 = subprocess.run(["ffmpeg","-i","https://freesound.org/data/previews/186/186942_2594536-hq.mp3","-af","adelay=15000|15000","-f","mp3","pipe:"],stdout=subprocess.PIPE).stdout
# Read second audio sample from https://cdns-preview-b.dzcdn.net/stream/c-b0b684fe962f93dc43f1f7ea493683a1-3.mp3
sample2 = subprocess.run(["ffmpeg","https://cdns-preview-b.dzcdn.net/stream/c-b0b684fe962f93dc43f1f7ea493683a1-3.mp3",stdout=subprocess.PIPE).stdout
return sample1,sample2
def writer(data,chunk_size):
# Open the pipes as opening files (open for "open for writing only").
fd_pipe = os.open(pipe_name,chunk_size):
# Write to named pipe as writing to a file (but write the data in small chunks).
os.write(fd_pipe,data[i:chunk_size+i]) # Write 1024 bytes of data to fd_pipe
# Closing the pipes as closing files.
os.close(fd_pipe)
def record(samp1,samp2):
# Names of the "Named pipes"
pipe1 = "audio_pipe1"
pipe2 = "audio_pipe2"
# Create "named pipes".
os.mkfifo(pipe1)
os.mkfifo(pipe2)
# Open FFmpeg as sub-process
# Use two audio input streams:
# 1. Named pipe: "audio_pipe1"
# 2. Named pipe: "audio_pipe2"
# Merge the two audio streams using amix audio filter.
# Store the result to output file: output.mp3
process = subprocess.Popen(["ffmpeg","-y",'-f','mp3',pipe1,pipe2,"-filter_complex","amix=inputs=2:duration=longest","output.mp3"],stdin=subprocess.PIPE)
# Initialize two "writer" threads (each writer writes data to named pipe in chunks of 1024 bytes).
thread1 = Thread(target=writer,args=(samp1,1024)) # thread1 writes samp1 to pipe1
thread2 = Thread(target=writer,args=(samp2,1024)) # thread2 writes samp2 to pipe2
# Start the two threads
thread1.start()
thread2.start()
# Wait for the two writer threads to finish
thread1.join()
thread2.join()
process.wait() # Wait for FFmpeg sub-process to finish
# Remove the "named pipes".
os.unlink(pipe1)
os.unlink(pipe2)
sampl1,sampl2 = create_samp()
record(sampl1,sampl2)
更新:
使用类的相同解决方案:
使用类(“NamedPipeWriter
”类)实现解决方案更优雅一些。
该类继承了Thread类,并覆盖了run
方法。
您可以创建一个包含多个对象的列表,并在循环中迭代它们(而不是为每个新的输入流复制代码)。
这是使用类的相同解决方案:
import subprocess
import os
import stat
from threading import Thread
def create_samp():
# Read audio stream from https://freesound.org/data/previews/186/186942_2594536-hq.mp3
# Apply adelay audio filter.
# Encode the audio in mp3 format.
# FFmpeg output is passed to stdout pipe,sample2
class NamedPipeWriter(Thread):
""" Write data (in small chunks) to a named pipe using a thread """
def __init__(self,data):
""" Initialization - get pipe name and data to be written """
super().__init__()
self._pipe_name = pipe_name
self._chunk_size = 1024
self._data = data
def run(self):
""" Open the pipe,write data in small chunks and close the pipe """
chunk_size = self._chunk_size
data = self._data
# Open the pipes as opening files (open for "open for writing only").
fd_pipe = os.open(self._pipe_name,os.O_WRONLY) # fd_pipe1 is a file descriptor (an integer)
for i in range(0,chunk_size):
# Write to named pipe as writing to a file (but write the data in small chunks).
os.write(fd_pipe,data[i:chunk_size+i]) # Write 1024 bytes of data to fd_pipe
# Closing the pipes as closing files.
os.close(fd_pipe)
def record(samp1,samp2):
# Names of the "Named pipes"
pipe1 = "audio_pipe1"
pipe2 = "audio_pipe2"
# Create "named pipes".
if not stat.S_ISFIFO(os.stat(pipe1).st_mode):
os.mkfifo(pipe1) # Create the pipe only if not exist.
if not stat.S_ISFIFO(os.stat(pipe2).st_mode):
os.mkfifo(pipe2)
# Open FFmpeg as sub-process
# Use two audio input streams:
# 1. Named pipe: "audio_pipe1"
# 2. Named pipe: "audio_pipe2"
# Merge the two audio streams using amix audio filter.
# Store the result to output file: output.mp3
process = subprocess.Popen(["ffmpeg",stdin=subprocess.PIPE)
# Initialize two "writer" threads (each writer writes data to named pipe in chunks of 1024 bytes).
named_pipe_writer1 = NamedPipeWriter(pipe1,samp1)
named_pipe_writer2 = NamedPipeWriter(pipe2,samp2)
# Start the two threads
named_pipe_writer1.start()
named_pipe_writer2.start()
# Wait for the two writer threads to finish
named_pipe_writer1.join()
named_pipe_writer1.join()
process.wait() # Wait for FFmpeg sub-process to finish
# Remove the "named pipes".
os.unlink(pipe1)
os.unlink(pipe2)
sampl1,sampl2)
注意事项:
- 代码已在 Ubuntu 18.04(在虚拟机中)测试。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。