如何修复录制未知长度音频数据时的错误？（自定义时长控制场景）

Hey there! Let's get your custom-duration audio recording script working smoothly. Your initial approach uses multiprocessing with global variables, which can lead to unexpected behavior since processes don't share memory by default. Let's break down two reliable solutions depending on exactly what you need:

Solution 1: Record for a User-Specified Fixed Duration #

If you want users to input a specific length of time (in seconds) before recording starts, this straightforward approach works perfectly—no threads or processes needed:

import pyaudio
import wave

def record_audio(duration, output_filename="recording.wav"):
    # Core audio configuration (adjust these as needed)
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 44100  # Standard sample rate
    CHUNK = 1024  # Buffer size

    p = pyaudio.PyAudio()

    # Open the input stream
    stream = p.open(format=FORMAT,
                    channels=CHANNELS,
                    rate=RATE,
                    input=True,
                    frames_per_buffer=CHUNK)

    print(f"Recording for {duration} seconds... Don't make noise!")
    frames = []

    # Calculate how many chunks we need to capture for the given duration
    total_chunks = int(RATE / CHUNK * duration)
    for _ in range(total_chunks):
        data = stream.read(CHUNK)
        frames.append(data)

    print("Recording done!")

    # Clean up resources properly
    stream.stop_stream()
    stream.close()
    p.terminate()

    # Save the recorded data to a WAV file
    with wave.open(output_filename, 'wb') as wf:
        wf.setnchannels(CHANNELS)
        wf.setsampwidth(p.get_sample_size(FORMAT))
        wf.setframerate(RATE)
        wf.writeframes(b''.join(frames))
    
    print(f"Audio saved to {output_filename}")

if __name__ == "__main__":
    # Get valid duration input from the user
    while True:
        try:
            user_duration = float(input("Enter recording duration (seconds): "))
            if user_duration > 0:
                break
            print("Please enter a positive number—can't record for zero or negative time!")
        except ValueError:
            print("Oops, that's not a number. Try again.")
    
    record_audio(user_duration)

What Makes This Better: #

• No messy globals: We calculate exactly how much audio to capture upfront, so there's no need for shared state between processes.
• Input validation: Ensures users can't enter invalid values (like non-numbers or negative durations).
• Clean resource handling: Properly closes the audio stream and terminates PyAudio to avoid leaving system resources hanging.
• Clear feedback: Users get status updates so they know what's happening at each step.

Solution 2: Stop Recording On-Demand (User Input Triggers Stop) #

If you want users to start recording first, then stop it whenever they want (by pressing Enter, for example), a threaded approach is ideal. Threads share memory with the main process, so we can safely control the recording loop:

import pyaudio
import wave
import threading

# Thread-safe flag to control recording
is_recording = True
frames = []

def recording_worker():
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 44100
    CHUNK = 1024

    p = pyaudio.PyAudio()
    stream = p.open(format=FORMAT,
                    channels=CHANNELS,
                    rate=RATE,
                    input=True,
                    frames_per_buffer=CHUNK)

    print("Recording now! Press Enter to stop.")
    while is_recording:
        data = stream.read(CHUNK)
        frames.append(data)

    # Clean up after recording stops
    stream.stop_stream()
    stream.close()
    p.terminate()

def save_recording(output_filename="recording.wav"):
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 44100

    with wave.open(output_filename, 'wb') as wf:
        wf.setnchannels(CHANNELS)
        wf.setsampwidth(pyaudio.PyAudio().get_sample_size(FORMAT))
        wf.setframerate(RATE)
        wf.writeframes(b''.join(frames))
    
    print(f"Recording saved to {output_filename}")

if __name__ == "__main__":
    # Start recording in a background thread
    record_thread = threading.Thread(target=recording_worker)
    record_thread.start()

    # Wait for user input to stop recording
    input()
    is_recording = False

    # Wait for the recording thread to finish, then save
    record_thread.join()
    save_recording()

How This Works: #

• The recording runs in a background thread, so the main thread can listen for user input without blocking the audio capture.
• When the user presses Enter, we toggle the is_recording flag to stop the loop, then wait for the thread to finish before saving the file.
• This avoids the complexity of multiprocessing while still letting users control when recording stops.

内容的提问来源于stack exchange，提问作者Raady