audio recorder #python #IO #audio #STT

audio_recorder.py

class AudioRecorder:
    def __init__(self):

        self.is_recording = False
        self.recording_queue = queue.Queue()
        self.stop_recording = threading.Event()
        self.recording_thread = None

    def start_recording(self):
        if not self.is_recording:
            self.is_recording = True
            self.stop_recording.clear()
            logger.info(
                "Started recording... Press 'esc' to stop recording without transcribing."
            )
            self.recording_thread = threading.Thread(target=self._record_audio)
            self.recording_thread.start()

    def stop_recording_process(self):
        self.is_recording = False
        self.stop_recording.set()
        if self.recording_thread:
            self.recording_thread.join()

    def _record_audio(self, chunk_duration=0.5):
        recognizer = sr.Recognizer()
        with sr.Microphone() as source:
            logger.info("Recording audio. Please speak now...")
            recognizer.adjust_for_ambient_noise(source)
            audio_chunks = []
            while self.is_recording:
                if self.stop_recording.is_set():
                    logger.info("Recording stopped by 'esc' key.")
                    audio_chunks.clear()
                    break
                audio_chunk = recognizer.record(source, duration=chunk_duration)
                audio_chunks.append(audio_chunk)

        if audio_chunks:
            raw_data = b"".join(chunk.get_raw_data() for chunk in audio_chunks)
            combined_audio = sr.AudioData(
                raw_data, audio_chunks[0].sample_rate, audio_chunks[0].sample_width
            )
            self.recording_queue.put(combined_audio)

    def get_recorded_audio(self):
        try:
            return self.recording_queue.get_nowait()
        except queue.Empty:
            return None