实时说话人识别的Python多处理模块的实现

Question

我正在开发一种实时说话人识别算法。我的想法是使用writeAudio()模块并行运行三个任务，即detectionBlock()和identificationBlock()。

实际上，writeAudio()函数使用PyAudio捕获连续记录并将0.5秒音频文件保存到本地目录，detectionBlock()函数处理来自目录的两个最老的0.5秒文件，并使用语音活动检测(VAD)模型来确定音频是语音还是噪声，identificationBlock()函数处理单独的3秒音频文件(从0.5秒音频文件块保存到另一个目录)，然后使用语音识别(VR)模型来确定扬声器的身份。

我希望我能在这里应用multiprocessing，以避开Global解释器锁(GIL)，同时作为Process对象运行这三个函数。目前，程序将在detectionBlock()或identificationBlock()函数完成录制之后才开始运行。

下面是当前使用multiprocessing实现的代码

from multiprocessing import Process

# Perform Parallel Processing with the Multiprocessing Module
    def parallelProcessing(self):
        
        # Define Individual Functions as Process() Objects
        rec = Process(target=self.writeAudio()) # Cog 1
        vad = Process(target=self.detectionBlock()) # Cog 2
        si = Process(target=self.identificationBlock()) # Cog 3
        
        cogs = [rec, vad, si] # List of functions
        
        # Run All Three Cogs in Parallel
        rec.start() # Start Cog 1
        
        time.sleep(3) # Wait 3 sec to start speech detection & identification
 
        vad.start() # Start Cog 2
        si.start() # Start Cog 3
        
        for cog in cogs:
            cog.join() # Wait for processes to complete before continuing

我以前从未应用过multiprocessing，所以我想知道用不同的实现方法这是否可行。谢谢你的帮助。

编辑：

为了提高清晰度，我添加了下面函数的简化版本。

# Speech Detection Sequence
    def detectionBlock(self):
        
        # Create VoiceActivityDetectionModel() Class Object
        vad = VoiceActivityDetectionModel()
        
        # Run Speech Detection on Oldest Audio Segments in Directory
        files = self.getListDir() # List of audiofiles
        index = 0 # First file in list
        
        path_1 = os.path.join(self.VAD_audio_path, files[index])
        path_2 = os.path.join(self.VAD_audio_path, files[index+1])
        
        label_1, _, _ = vad.detectFromAudiofile(path_1) # VAD classifier for first segment
        label_2, _, _ = vad.detectFromAudiofile(path_2) # VAD classifier for second segment
        
        if (label_1 == 'speech') and (label_2 == 'speech'):
            self.combineAudio(index) # Generate 3-sec recording for SI if 
                                     # speech is detected in both audiofiles
        else:
            self.deleteAudio() # Remove oldest audio segment

    # Speaker Identification Sequence
    def identificationBlock(self):
        
        # Create EnsemblePredictions() Class Object
        ep = EnsemblePredictions()
        
        # Run Speaker Identification on Oldest Audio Segment in Directory
        files = self.getListDir(audio_type='SI')
        index = 0 # First file in list
        
        if files:
            filepath = os.path.join(self.SI_audio_path, files[index])
        
            speaker, prob_list = ep.predict(filepath, first_session=False) # SI classifier
            time_stamp = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime()) # Time of identification
        
            self.speakerDiarization(speaker=speaker, prob_list=prob_list, time_stamp=time_stamp) # Save results
        
            # Remove 3-Second Audio Segment from Directory
            self.deleteAudio(audio_type='SI')

# Audio Recording Sequence
    def writeAudio(self):
        
        # Instantiate Recording System Variables
        FORMAT = pyaudio.paFloat32 # 32 bits per sample
        CHANNELS = 1 # Mono
        RATE = 16000 # Sampling Rate
        CHUNK = int(self.VAD_audio_length*RATE) # Chunks of bytes to record from microphone
        
        # Initialize Recording
        p = pyaudio.PyAudio() # Create interface to PortAudio

        input('Press ENTER to Begin Recording') # Wait for keypress to record

        if keyboard.is_pressed('Enter'):
            stream = p.open(format=FORMAT,
                            channels=CHANNELS,
                            rate=RATE,
                            frames_per_buffer=CHUNK,
                            input=True)
            
            print()
            print('Hold SPACE to Finish Recording')
            
            while(True):
                # End Process with Manual User Interrupt
                if keyboard.is_pressed('Space'):
                    break 
                
                # Generate Audio Recording
                data = stream.read(CHUNK) # Read 0.5-second segment from audio stream
                data = np.frombuffer(data, dtype=np.float32) # Convert to NumPy array
                
                filename = 'VAD_segment_' + str(self.VAD_audio_count) + '.wav'
                
                sf.write(os.path.join(self.VAD_audio_path, filename), data, RATE)
        
                # Adjust Segment Count
                self.VAD_audio_count = self.VAD_audio_count + 1 # Increment
                
            # Stop & Close Stream
            stream.stop_stream()
            stream.close()
        
            # Terminate PortAudio Interface
            p.terminate()

Answer 1

下面是我在评论中提到的一个例子。我没有真正运行它的所有组件，所以把它当作伪代码来处理，但是我认为它应该是一个很好的起点。主要的改进是在pastream的帮助下进行了一些简化，它声称基本没有GIL的portaudio迭代。这里的好处是减少开销和更容易地将数据传输到流水线的至少第一阶段，即检测音频。在出现减速时，您可能需要一些额外的复杂性来删除框架，但是如果我正确理解pastream文档，这种结构通常应该能工作。

import pastream
import multiprocessing as mp
from Queue import Empty

class ExitFlag: pass

def voice_identification(rx_q: mp.Queue):
    while True:
        try:
            received = rx_q.get(1) 
            #if voice_identification is too slow you may want to `get` until 
            #  the queue is empty to drop all but most recent frame. This way
            #  you won't have an infinitely growing queue.
        except Empty:
            pass
        if isinstance(received, ExitFlag):
            break
        
        else:
            print(identify(received)) #identify audio
    print("identifier process exiting")

if __name__ == "__main__":
    tx_q = mp.Queue()
    identifier_p = mp.Process(target=voice_identification, args=(tx_q,))
    identifier_p.start()
    
    samplerate=44100
    stream = pastream.InputStream()
    #3 second chunks every half second
    for chunk in stream.chunks(chunksize=samplerate/2, overlap=(samplerate/2)*5):
        if detect_audio(chunk): #detect audio
            tx_q.put(chunk)
        if exit_key_down(): #however you want to detect this, it's good to ensure smooth shutdown of child
            tx_q.put(ExitFlag())
            identifier_p.join()
            break