Skip to main content
 首页 » 编程设计

python : PortAudio + Opus encoding/decoding

2025年05月04日10落叶

我正在使用 Pyaudio 从我的麦克风中捕获音频,并尝试使用 opus 编解码器对其进行编码/解码。我正在使用 SvartalF ( https://github.com/svartalf/python-opus ) 制作的 libopus 绑定(bind)。

这是我的代码:

import pyaudio 
from opus import encoder, decoder 
 
def streaming(p): 
    chunk = 960 
    FORMAT = pyaudio.paInt16 
    CHANNELS = 1 
    RATE = 48000 
    streamin = p.open(format = FORMAT, 
            channels = CHANNELS,  
            rate = RATE,  
            input = True, 
            input_device_index = 7, 
            frames_per_buffer = chunk) 
    streamout = p.open(format = FORMAT, 
            channels = CHANNELS,  
            rate = 48000,  
            output = True, 
            output_device_index = p.get_default_input_device_info()["index"], 
            frames_per_buffer = chunk) 
    enc = encoder.Encoder(RATE,CHANNELS,'voip') 
    dec = decoder.Decoder(RATE,CHANNELS) 
    data = [] 
    for i in xrange(100): 
        data.append(streamin.read(chunk*2)) 
    streamout.write(''.join(data)) 
    encdata = [] 
    for x in data: 
        encdata.append(enc.encode(x,chunk)) 
    print "DATA LENGTH :", len(''.join(data)) 
    print "ENCDATA LENGTH :", len(''.join(encdata)) 
    decdata = '' 
    for x in encdata: 
        decdata += dec.decode(x,chunk) 
    print "DECDATA LENGTH :", len(decdata) 
    streamout.write(decdata) 
    streamin.close() 
    streamout.close() 
 
 
p = pyaudio.PyAudio() 
streaming(p) 
p.terminate() 

我必须在 data.append(streamin.read(chunk*2)) 或 中放置 chunk*2 而不是 chunk >DECDATA LENGTH == DATA LENGTH*2 我不知道为什么。

输出:

DATA LENGTH :    384000   
ENCDATA LENGTH : 12865   
DECDATA LENGTH : 384000 

如果没有编码/解码,第一个 streamout.write(''.join(data)) 可以完美运行。通过编码/解码,streamout.write(decdata) 有点工作,但混入了很多噼啪声。

我在这里做错了什么?

请您参考如下方法:

这似乎是由解码方法中的 python-opus 错误引起的。

根据Opus API , opus_decode 返回解码的样本数。 python 绑定(bind)假定它将完全填满它传入的结果缓冲区,因此每组解码样本都会附加一个静默。这种沉默会导致低帧尺寸时出现裂纹,而高帧尺寸时会出现断断续续的情况。虽然文档对此没有任何说明,但返回的数字似乎是每个 channel 的。

换行 150 of opus/api/decoder.py以下为我修复它:

    return array.array('h', pcm[:result*channels]).tostring() 

如果您需要使用 decode_float 方法,可能需要进行相同的更改。