python - 如何在 python 中录制不确定持续时间的音频并允许暂停和恢复功能？

Question

我正在编写一个 Python 应用程序，将音频录制为 WAV 文件，直到用户按下暂停或停止。暂停音频后，用户还应该能够恢复录制。另外:

• 应用程序无法预先知道录制时间
• 应用程序应避免内存不足(因为录音可能会很长)。例如，它可以实时写入 WAV 文件，以防止将不断增长的录音存储在内存中。

解决这个问题的好方法是什么？您能为您的解决方案提供一些代码片段吗？

与 python-sounddevice ，我可以 stop() 和 start() 流来模仿“暂停”功能。我可以指定一个 numpy 数组作为记录的输出。但是:

• 我不知道要制作多大的数组(因为我不知道录制持续时间)
• 当数组填满时我该怎么办？

<强> python-sounddevice and sound-file 可以支持在事先不知道大小的情况下进行录音。但是:

• 如何合并“暂停”和“恢复”功能？声音文件只有 read 和 write 方法。
• 是否有比使用 KeyBoardInterrupt 更好的方法来停止流？
• 我可以在每次“暂停”后创建不同的录音吗 combine the WAV files用户点击“停止”后？
• 我尝试使用 Threading.Event() 来阻止录制线程以模仿暂停功能，但录制内容不断写入文件

<小时/>

我对声音设备方法的尝试

paused = False

def record():
    self.recording = ? # create numpy.ndarray of the correct size 
                       # (not sure the best way to do this without 
                       # knowing the recording duration)
    with sd.InputStream(samplerate=44100, device=mic, channels=1, 
        callback=self.callback):

            while self.paused:
            sd.stop()
        sd.rec(out=recording) # but what happens if 
                              # recording is very long
                              # or numpy array fills up?

def stop_and_save():
    sd.stop()
    scipy.io.wavfile.write("recording.wav", 44100, self.recording)

<小时/>

声音设备和声音文件方法:

with sf.SoundFile(args.filename, mode='x', samplerate=args.samplerate,
                      channels=args.channels, subtype=args.subtype) as file:
        with sd.InputStream(samplerate=args.samplerate, device=args.device,
                            channels=args.channels, callback=callback):
            print('press Ctrl+C to stop the recording')
            while True:
                file.write(q.get())  # but how do you stop writing when 'paused'?

except KeyboardInterrupt:
    print('\nRecording finished: ' + repr(args.filename))
    parser.exit(0)
except Exception as e:
    parser.exit(type(e).__name__ + ': ' + str(e))

Answer 1

我提出了这个暂停/恢复功能的解决方案，它利用声音设备和声音文件方法，只要用户单击暂停，恢复后将开始新的录制。然后，在用户单击停止后，所有WAV文件将按顺序组合。

( Matthias' code 看起来也是一个很好的解决方案，可以更多地利用线程。)

<小时/>

要开始录制音频:

    def record(self):
        try:
            with sf.SoundFile(self.filepath,
                                       mode='x', samplerate=self.SAMPLE_RATE,
                                       channels=self.CHANNELS, subtype=None) as file:
                with sd.InputStream(samplerate=self.SAMPLE_RATE, device=self.mic_id,
                                           channels=self.CHANNELS, callback=self.callback):
                    logger.info(f"New recording started: {self.sound_file.name}")
                    try:
                        while True:
                            file.write(self.mic_queue.get())

                    except RuntimeError as re:
                        logger.debug(f"{re}. If recording was stopped by the user, then this can be ignored")

record()的回调:

    def callback(self, indata, frames, time, status):
        """This is called (from a separate thread) for each audio block."""
        if status:
            print(status, file=sys.stderr)
        self.mic_queue.put(indata.copy())

要暂停:

    def pause_recording(self):
        """Mimics a 'pause' functionality by writing the current sound file changes to disk.
        Upon 'resume' a new recording will be made. Note: close() is not called here, because
        that would kill the recording thread
        """
        self.sound_file.flush()
        logger.info(f"'Paused' (closed) recording: {self.sound_file.name}")

要恢复:

    def resume_recording(self):
        """
        Mimics 'resuming' by starting a new recording, which will be merged with the others
        when the user selects Stop & Save (or deleted upon Stop & Delete)
        Note: get_full_sound_file_name() outputs a new recording with the same base name as the first, but appends a `_part2` or `_part3` etc. to the suffix to distinguish it from the first and maintain order.
        """
        self.sound_file = self.get_full_sound_file_name()
        self.record()

要停止录制:

    def stop_mic_recording(self):
        try:
            self.sound_file.flush()
            self.sound_file.close()
            logger.info(f"Stopped and closed recording: {self.sound_file.name}")

        except RuntimeError as e:
            logger.info(f"Error stopping/saving {self.sound_file.name}. Make sure the file exists and can be modified")
            logger.info(f"RunTimeError: \n{e}")

合并音频(在 stop_recording() 之后调用):

   def combine_recordings_if_needed(self):
        """
        If recording was paused, combines all sections in alphabetical order into a new audio file
        """
        if self.section_count > 1:   # this is incremented when a recording is paused/resumed
            combined_audio = AudioSegment.empty()
            files_combined = []
            for rec in glob.glob(os.path.join(RECORDING_DIR, "*" + self.FILE_EXT)):
                combined_audio = combined_audio + AudioSegment.from_wav(rec) # this is why alphabetical order is important
                files_combined.append(rec)

            combined_file_name = os.path.join(RECORDING_DIR, self.base_filename + "_combined" + self.FILE_EXT)
            combined_audio.export(out_f=combined_file_name, format="wav")
            logger.info(f"Combined the following recordings into {combined_file_name}:"
                        f"\n {files_combined}")