时间戳

首页 > 解决方案 > 如何在主线程在Python中继续执行时从主线程生成子线程

python - 如何在主线程在Python中继续执行时从主线程生成子线程

问题描述

我正在制作一个应用程序来将从麦克风录制的音频转换为文本。录音的长度可能很长，比如 3 小时，所以我猜最好将其转换为持续时间短的波形文件，比如一分钟左右，然后生成一个子线程，在其中执行音频到文本的操作，而主线程可以开始录制下一分钟。音频到文本的操作比录制部分快得多，因此时间不会成为问题。

流程图

这是我认为它应该如何工作的流程图。

我pyaudio用于录制音频。它的代码：

import pyaudio
import wave
import time

def read_audio(stream):
    chunk = 1024  # Record in chunks of 1024 samples
    sample_format = pyaudio.paInt16  # 16 bits per sample
    channels = 2
    fs = 44100  # Record at 44100 samples per second
    seconds = 10
    filename = 'record.wav'
    frames = []  # Initialize array to store frames
    # Store data in chunks for 3 seconds
    
    for i in range(0, int(fs / chunk * seconds)):
        data = stream.read(chunk)
        frames.append(data)
    
    # Save the recorded data as a WAV file
    wf = wave.open(filename, 'wb')
    wf.setnchannels(channels)
    wf.setsampwidth(p.get_sample_size(sample_format))
    wf.setframerate(fs)
    wf.writeframes(b''.join(frames))
    wf.close()
    
    # Stop and close the stream
    stream.stop_stream()
    stream.close()

    

p = pyaudio.PyAudio()  # Create an interface to PortAudio
chunk = 1024  # Record in chunks of 1024 samples
sample_format = pyaudio.paInt16  # 16 bits per sample
channels = 2
fs = 44100
stream = p.open(format=sample_format,channels=channels,rate=fs,
                frames_per_buffer=chunk,input=True)
read_audio(stream)
p.terminate() # Terminate the PortAudio interface

对于语音识别，speech_recognition使用 Google 的 API。它的代码：

import speech_recognition as sr

def convert():
    sound = "record.wav"
 
    r = sr.Recognizer()
 
 
    with sr.AudioFile(sound) as source:
        r.adjust_for_ambient_noise(source)
        print("Converting Audio To Text and saving to file..... ") 
        audio = r.listen(source)
    try:

        value = r.recognize_google(audio) ##### API call to google for speech recognition

        if str is bytes: 
            result = u"{}".format(value).encode("utf-8")

        else: 
            result = "{}".format(value)

        with open("test.txt","a") as f:
            f.write(result)
        print("Done !\n\n")

    except sr.UnknownValueError:
        print("")
    except sr.RequestError as e:
        print("{0}".format(e))
    except KeyboardInterrupt:
        pass
 
convert()

标签： pythonmultithreading

解决方案

由于 GIL，Python 从来都不是真正的多线程，但是在您的情况下这可能并不重要，因为您正在使用 api 调用为您进行语音识别。

所以你可以试试这个来启动一个线程来做转换

from threading import Thread
t = Thread(target=convert)
t.start()

在您尝试转换下一分钟之前，您可能会尝试加入最后一个线程以确保它已完成

t.join()

您可能还可以使用 asyncio 库

虽然这可能有点矫枉过正，但我​​可能会使用多处理库。在您的情况下，您可能有一个不断录制和保存新声音文件的侦听器工作进程，以及一个不断寻找新文件并转换它们的转换工作进程。

如果需要，这将允许您编写更强大的系统。例如，如果您失去了互联网连接并且在几分钟内无法通过 google api 转换您的声音文件，录音机工作人员将继续保存声音文件，而不用关心在互联网连接恢复时会得到处理的声音文件。

无论如何，这里有一个你可以使用的转换工作进程的小例子。

import multiprocessing as mp
import os
from pathlib import Path
from time import sleep

class ConversionWorker:

    def __init__(self, sound_file_directory_path: str, text_save_filepath: str):
        self.sound_directory_path = Path(sound_file_directory_path)
        self.text_filepath = Path(text_save_filepath)

    def run(self):
        while True:

            # find and convert all wav files in the target directory
            filepaths = self.sound_directory_path.glob('*.wav')
            for path in filepaths:
                # convert from path
                # save to self.text_filepath
                convert()

                # we can delete the sound file after converting it
                os.remove(path)

            # sleep for a bit since we are only saving files once a minute or so
            sleep(5)

def main():
    conversion_worker = ConversionWorker(sound_file_directory_path='path/to/sounds', text_save_filepath='path/to/text')
    p = mp.Process(target=conversion_worker.run)
    p.start()

    # do the recording and saving for as long as you want

    p.terminate()

推荐阅读