RealTimeSTT-fork/RealtimeSTT/audio_recorder.py

"""

The AudioToTextRecorder class in the provided code facilitates fast speech-to-text transcription.

The class employs the faster_whisper library to transcribe the recorded audio 
into text using machine learning models, which can be run either on a GPU or CPU.
Voice activity detection (VAD) is built in, meaning the software can automatically 
start or stop recording based on the presence or absence of speech.
It integrates wake word detection through the pvporcupine library, allowing the 
software to initiate recording when a specific word or phrase is spoken.
The system provides real-time feedback and can be further customized.

Features:
- Voice Activity Detection: Automatically starts/stops recording when speech is detected or when speech ends.
- Wake Word Detection: Starts recording when a specified wake word (or words) is detected.
- Event Callbacks: Customizable callbacks for when recording starts or finishes.
- Fast Transcription: Returns the transcribed text from the audio as fast as possible.

Author: Kolja Beigel

"""

from multiprocessing import Process, Pipe, Queue, Event
import faster_whisper
import collections
import numpy as np
import pvporcupine
import collections
import traceback
import threading
import webrtcvad
import itertools
import pyaudio
import logging
import struct
import torch
import halo
import time
import os
import re

INIT_MODEL_TRANSCRIPTION = "tiny"
INIT_MODEL_TRANSCRIPTION_REALTIME = "tiny"
INIT_REALTIME_PROCESSING_PAUSE = 0.2
INIT_SILERO_SENSITIVITY = 0.4
INIT_WEBRTC_SENSITIVITY = 3
INIT_POST_SPEECH_SILENCE_DURATION = 0.6
INIT_MIN_LENGTH_OF_RECORDING = 0.5
INIT_MIN_GAP_BETWEEN_RECORDINGS = 0
INIT_WAKE_WORDS_SENSITIVITY = 0.6
INIT_PRE_RECORDING_BUFFER_DURATION = 1.0
INIT_WAKE_WORD_ACTIVATION_DELAY = 0.0
INIT_WAKE_WORD_TIMEOUT = 5.0
ALLOWED_LATENCY_LIMIT = 10

TIME_SLEEP = 0.02
SAMPLE_RATE = 16000
BUFFER_SIZE = 512
INT16_MAX_ABS_VALUE = 32768.0

class AudioToTextRecorder:
    """
    A class responsible for capturing audio from the microphone, detecting voice activity, and then transcribing the captured audio using the `faster_whisper` model.
    """
    
    def __init__(self,
                 model: str = INIT_MODEL_TRANSCRIPTION,
                 language: str = "",
                 on_recording_start = None,
                 on_recording_stop = None,
                 on_transcription_start = None,
                 ensure_sentence_starting_uppercase = True,
                 ensure_sentence_ends_with_period = True,
                 spinner = True,
                 level=logging.WARNING,

                 # Realtime transcription parameters
                 enable_realtime_transcription = False,
                 realtime_model_type = INIT_MODEL_TRANSCRIPTION_REALTIME,
                 realtime_processing_pause = INIT_REALTIME_PROCESSING_PAUSE,
                 on_realtime_transcription_update = None,
                 on_realtime_transcription_stabilized = None,

                 # Voice activation parameters
                 silero_sensitivity: float = INIT_SILERO_SENSITIVITY,
                 silero_use_onnx: bool = False,
                 webrtc_sensitivity: int = INIT_WEBRTC_SENSITIVITY,
                 post_speech_silence_duration: float = INIT_POST_SPEECH_SILENCE_DURATION,
                 min_length_of_recording: float = INIT_MIN_LENGTH_OF_RECORDING,
                 min_gap_between_recordings: float = INIT_MIN_GAP_BETWEEN_RECORDINGS,
                 pre_recording_buffer_duration: float = INIT_PRE_RECORDING_BUFFER_DURATION,
                 on_vad_detect_start = None,
                 on_vad_detect_stop = None,

                 # Wake word parameters
                 wake_words: str = "",
                 wake_words_sensitivity: float = INIT_WAKE_WORDS_SENSITIVITY,
                 wake_word_activation_delay: float = INIT_WAKE_WORD_ACTIVATION_DELAY,
                 wake_word_timeout: float = INIT_WAKE_WORD_TIMEOUT,
                 on_wakeword_detected = None,
                 on_wakeword_timeout = None,
                 on_wakeword_detection_start = None,
                 on_wakeword_detection_end = None,
                 ):
        """
        Initializes an audio recorder and  transcription and wake word detection.

        Args:
        - model (str, default="tiny"): Specifies the size of the transcription model to use or the path to a converted model directory. 
                Valid options are 'tiny', 'tiny.en', 'base', 'base.en', 'small', 'small.en', 'medium', 'medium.en', 'large-v1', 'large-v2'. 
                If a specific size is provided, the model is downloaded from the Hugging Face Hub.
        - language (str, default=""): Language code for speech-to-text engine. If not specified, the model will attempt to detect the language automatically.
        - on_recording_start (callable, default=None): Callback function to be called when recording of audio to be transcripted starts.
        - on_recording_stop (callable, default=None): Callback function to be called when recording of audio to be transcripted stops.
        - on_transcription_start (callable, default=None): Callback function to be called when transcription of audio to text starts.  
        - ensure_sentence_starting_uppercase (bool, default=True): Ensures that every sentence detected by the algorithm starts with an uppercase letter.
        - ensure_sentence_ends_with_period (bool, default=True): Ensures that every sentence that doesn't end with punctuation such as "?", "!" ends with a period
        - spinner (bool, default=True): Show spinner animation with current state.
        - level (int, default=logging.WARNING): Logging level.
        - enable_realtime_transcription (bool, default=False): Enables or disables real-time transcription of audio. When set to True, the audio will be transcribed continuously as it is being recorded.
        - realtime_model_type (str, default="tiny"): Specifies the machine learning model to be used for real-time transcription. Valid options include 'tiny', 'tiny.en', 'base', 'base.en', 'small', 'small.en', 'medium', 'medium.en', 'large-v1', 'large-v2'.
        - realtime_processing_pause (float, default=0.1): Specifies the time interval in seconds after a chunk of audio gets transcribed. Lower values will result in more "real-time" (frequent) transcription updates but may increase computational load.
        - on_realtime_transcription_update = A callback function that is triggered whenever there's an update in the real-time transcription. The function is called with the newly transcribed text as its argument.
        - on_realtime_transcription_stabilized = A callback function that is triggered when the transcribed text stabilizes in quality. The stabilized text is generally more accurate but may arrive with a slight delay compared to the regular real-time updates.
        - silero_sensitivity (float, default=SILERO_SENSITIVITY): Sensitivity for the Silero Voice Activity Detection model ranging from 0 (least sensitive) to 1 (most sensitive). Default is 0.5.
        - silero_use_onnx (bool, default=False): Enables usage of the pre-trained model from Silero in the ONNX (Open Neural Network Exchange) format instead of the PyTorch format. This is recommended for faster performance.
        - webrtc_sensitivity (int, default=WEBRTC_SENSITIVITY): Sensitivity for the WebRTC Voice Activity Detection engine ranging from 0 (least aggressive / most sensitive) to 3 (most aggressive, least sensitive). Default is 3.
        - post_speech_silence_duration (float, default=0.2): Duration in seconds of silence that must follow speech before the recording is considered to be completed. This ensures that any brief pauses during speech don't prematurely end the recording.
        - min_gap_between_recordings (float, default=1.0): Specifies the minimum time interval in seconds that should exist between the end of one recording session and the beginning of another to prevent rapid consecutive recordings.
        - min_length_of_recording (float, default=1.0): Specifies the minimum duration in seconds that a recording session should last to ensure meaningful audio capture, preventing excessively short or fragmented recordings.
        - pre_recording_buffer_duration (float, default=0.2): Duration in seconds for the audio buffer to maintain pre-roll audio (compensates speech activity detection latency)
        - on_vad_detect_start (callable, default=None): Callback function to be called when the system listens for voice activity.
        - on_vad_detect_stop (callable, default=None): Callback function to be called when the system stops listening for voice activity.
        - wake_words (str, default=""): Comma-separated string of wake words to initiate recording. Supported wake words include:
                'alexa', 'americano', 'blueberry', 'bumblebee', 'computer', 'grapefruits', 'grasshopper', 'hey google', 'hey siri', 'jarvis', 'ok google', 'picovoice', 'porcupine', 'terminator'.
        - wake_words_sensitivity (float, default=0.5): Sensitivity for wake word detection, ranging from 0 (least sensitive) to 1 (most sensitive). Default is 0.5.
        - wake_word_activation_delay (float, default=0): Duration in seconds after the start of monitoring before the system switches to wake word activation if no voice is initially detected. If set to zero, the system uses wake word activation immediately.
        - wake_word_timeout (float, default=5): Duration in seconds after a wake word is recognized. If no subsequent voice activity is detected within this window, the system transitions back to an inactive state, awaiting the next wake word or voice activation.
        - on_wakeword_detected (callable, default=None): Callback function to be called when a wake word is detected.
        - on_wakeword_timeout (callable, default=None): Callback function to be called when the system goes back to an inactive state after when no speech was detected after wake word activation
        - on_wakeword_detection_start (callable, default=None): Callback function to be called when the system starts to listen for wake words
        - on_wakeword_detection_end (callable, default=None): Callback function to be called when the system stops to listen for wake words (e.g. because of timeout or wake word detected)

        Raises:
            Exception: Errors related to initializing transcription model, wake word detection, or audio recording.
        """

        self.language = language
        self.wake_words = wake_words
        self.wake_word_activation_delay = wake_word_activation_delay
        self.wake_word_timeout = wake_word_timeout
        self.ensure_sentence_starting_uppercase = ensure_sentence_starting_uppercase
        self.ensure_sentence_ends_with_period = ensure_sentence_ends_with_period
        self.min_gap_between_recordings = min_gap_between_recordings
        self.min_length_of_recording = min_length_of_recording       
        self.pre_recording_buffer_duration = pre_recording_buffer_duration
        self.post_speech_silence_duration = post_speech_silence_duration
        self.on_recording_start = on_recording_start
        self.on_recording_stop = on_recording_stop        
        self.on_wakeword_detected = on_wakeword_detected
        self.on_wakeword_timeout = on_wakeword_timeout
        self.on_vad_detect_start = on_vad_detect_start
        self.on_vad_detect_stop = on_vad_detect_stop
        self.on_wakeword_detection_start = on_wakeword_detection_start
        self.on_wakeword_detection_end = on_wakeword_detection_end
        self.on_transcription_start = on_transcription_start
        self.enable_realtime_transcription = enable_realtime_transcription
        self.realtime_model_type = realtime_model_type
        self.realtime_processing_pause = realtime_processing_pause
        self.on_realtime_transcription_update = on_realtime_transcription_update
        self.on_realtime_transcription_stabilized = on_realtime_transcription_stabilized
        self.allowed_latency_limit = ALLOWED_LATENCY_LIMIT
    
        self.level = level
        self.audio_queue = Queue()
        self.buffer_size = BUFFER_SIZE
        self.sample_rate = SAMPLE_RATE
        self.recording_start_time = 0
        self.recording_stop_time = 0
        self.wake_word_detect_time = 0
        self.silero_check_time = 0 
        self.silero_working = False
        self.speech_end_silence_start = 0 
        self.silero_sensitivity = silero_sensitivity
        self.listen_start = 0
        self.spinner = spinner
        self.halo = None
        self.state = "inactive"
        self.wakeword_detected = False
        self.text_storage = []
        self.realtime_stabilized_text = ""
        self.realtime_stabilized_safetext = ""
        self.is_webrtc_speech_active = False
        self.is_silero_speech_active = False
        self.recording_thread = None
        self.realtime_thread = None
        self.audio_interface = None
        self.audio = None
        self.stream = None
        self.start_recording_event = threading.Event()
        self.stop_recording_event = threading.Event()


        # Initialize the logging configuration with the specified level
        log_format = 'RealTimeSTT: %(name)s - %(levelname)s - %(message)s'

        # Create a logger
        logger = logging.getLogger()
        logger.setLevel(level)  # Set the root logger's level

        # Create a file handler and set its level
        file_handler = logging.FileHandler('realtimesst.log')
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(logging.Formatter(log_format))

        # Create a console handler and set its level
        console_handler = logging.StreamHandler()
        console_handler.setLevel(level)
        console_handler.setFormatter(logging.Formatter(log_format))

        # Add the handlers to the logger
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)

        self.is_shut_down = False
        self.shutdown_event = Event()

        logging.info(f"Starting RealTimeSTT")

        # Start transcription process
        self.main_transcription_ready_event = Event()
        self.parent_transcription_pipe, child_transcription_pipe = Pipe()
        self.transcript_process = Process(target=AudioToTextRecorder._transcription_worker, args=(child_transcription_pipe, model, self.main_transcription_ready_event, self.shutdown_event))
        self.transcript_process.start()

        # Start audio data reading process
        self.reader_process = Process(target=AudioToTextRecorder._audio_data_worker, args=(self.audio_queue, self.sample_rate, self.buffer_size, self.shutdown_event))
        self.reader_process.start()

        # Initialize the realtime transcription model
        if self.enable_realtime_transcription:
            try:
                logging.info(f"Initializing faster_whisper realtime transcription model {self.realtime_model_type}")
                self.realtime_model_type = faster_whisper.WhisperModel(model_size_or_path=self.realtime_model_type, device='cuda' if torch.cuda.is_available() else 'cpu')

            except Exception as e:
                logging.exception(f"Error initializing faster_whisper realtime transcription model: {e}")
                raise

            logging.debug('Faster_whisper realtime speech to text transcription model initialized successfully')


        # Setup wake word detection
        if wake_words:

            self.wake_words_list = [word.strip() for word in wake_words.lower().split(',')]
            sensitivity_list = [float(wake_words_sensitivity) for _ in range(len(self.wake_words_list))]

            try:
                self.porcupine  = pvporcupine.create(keywords=self.wake_words_list, sensitivities=sensitivity_list)
                self.buffer_size = self.porcupine.frame_length
                self.sample_rate = self.porcupine.sample_rate

            except Exception as e:
                logging.exception(f"Error initializing porcupine wake word detection engine: {e}")
                raise

            logging.debug('Porcupine wake word detection engine initialized successfully')


        # Setup voice activity detection model WebRTC
        try:
            logging.info(f"Initializing WebRTC voice with Sensitivity {webrtc_sensitivity}")
            self.webrtc_vad_model = webrtcvad.Vad()
            self.webrtc_vad_model.set_mode(webrtc_sensitivity)

        except Exception as e:
            logging.exception(f"Error initializing WebRTC voice activity detection engine: {e}")
            raise       

        logging.debug('WebRTC VAD voice activity detection engine initialized successfully')


        # Setup voice activity detection model Silero VAD
        try:
            self.silero_vad_model, _ = torch.hub.load(
                repo_or_dir="snakers4/silero-vad",
                model="silero_vad",
                verbose=False,
                onnx=silero_use_onnx
            )

        except Exception as e:
            logging.exception(f"Error initializing Silero VAD voice activity detection engine: {e}")
            raise       

        logging.debug('Silero VAD voice activity detection engine initialized successfully')


        self.audio_buffer = collections.deque(maxlen=int((self.sample_rate // self.buffer_size) * self.pre_recording_buffer_duration))
        self.frames = []

        # Recording control flags
        self.is_recording = False
        self.is_running = True
        self.start_recording_on_voice_activity = False
        self.stop_recording_on_voice_deactivity = False

        # Start the recording worker thread
        self.recording_thread = threading.Thread(target=self._recording_worker)
        self.recording_thread.daemon = True
        self.recording_thread.start()

        # Start the realtime transcription worker thread
        self.realtime_thread = threading.Thread(target=self._realtime_worker)
        self.realtime_thread.daemon = True
        self.realtime_thread.start()

        # Wait for transcription models to start
        logging.debug('Waiting for main transcription model to start')
        self.main_transcription_ready_event.wait()
        logging.debug('Main transcription model ready')

        logging.debug('RealtimeSTT initialization completed successfully')


    @staticmethod
    def _transcription_worker(conn, model_path, ready_event, shutdown_event):
        """
        Worker method that handles the continuous process of transcribing audio data.

        This method runs in a separate process and is responsible for:
        - Initializing the `faster_whisper` model used for transcription.
        - Receiving audio data sent through a pipe and using the model to transcribe it.
        - Sending transcription results back through the pipe.
        - Continuously checking for a shutdown event to gracefully terminate the transcription process.

        Args:
            conn (multiprocessing.Connection): The connection endpoint used for receiving audio data and sending transcription results.
            model_path (str): The path to the pre-trained faster_whisper model for transcription.
            ready_event (threading.Event): An event that is set when the transcription model is successfully initialized and ready.
            shutdown_event (threading.Event): An event that, when set, signals this worker method to terminate.

        Raises:
            Exception: If there is an error while initializing the transcription model.
        """        

        logging.info(f"Initializing faster_whisper main transcription model {model_path}")

        try:
            model = faster_whisper.WhisperModel(
                model_size_or_path=model_path,
                device='cuda' if torch.cuda.is_available() else 'cpu'
            )

        except Exception as e:
            logging.exception(f"Error initializing main faster_whisper transcription model: {e}")
            raise

        ready_event.set()

        logging.debug('Faster_whisper main speech to text transcription model initialized successfully')

        while not shutdown_event.is_set():
            if conn.poll(0.5):
                audio, language = conn.recv()
                try:
                    segments = model.transcribe(audio, language=language if language else None)[0]
                    transcription = " ".join(seg.text for seg in segments).strip()
                    conn.send(('success', transcription))
                except faster_whisper.WhisperError as e:
                    logging.error(f"Whisper transcription error: {e}")
                    conn.send(('error', str(e)))      
                except Exception as e:
                    logging.error(f"General transcription error: {e}")
                    conn.send(('error', str(e)))
            else:
                # If there's no data, sleep for a short while to prevent busy waiting
                time.sleep(0.02)


    @staticmethod
    def _audio_data_worker(audio_queue, sample_rate, buffer_size, shutdown_event):
        """
        Worker method that handles the audio recording process.

        This method runs in a separate process and is responsible for:
        - Setting up the audio input stream for recording.
        - Continuously reading audio data from the input stream and placing it in a queue.
        - Handling errors during the recording process, including input overflow.
        - Gracefully terminating the recording process when a shutdown event is set.

        Args:
            audio_queue (queue.Queue): A queue where recorded audio data is placed.
            sample_rate (int): The sample rate of the audio input stream.
            buffer_size (int): The size of the buffer used in the audio input stream.
            shutdown_event (threading.Event): An event that, when set, signals this worker method to terminate.

        Raises:
            Exception: If there is an error while initializing the audio recording.
        """
        logging.info("Initializing audio recording (creating pyAudio input stream)")

        try:
            audio_interface = pyaudio.PyAudio()
            stream = audio_interface.open(rate=sample_rate, format=pyaudio.paInt16, channels=1, input=True, frames_per_buffer=buffer_size)

        except Exception as e:
            logging.exception(f"Error initializing pyaudio audio recording: {e}")
            raise       

        logging.debug('Audio recording (pyAudio input stream) initialized successfully')
   
        try:
            while not shutdown_event.is_set():
                try:
                    data = stream.read(buffer_size)

                except OSError as e:
                    if e.errno == pyaudio.paInputOverflowed:
                        logging.warning("Input overflowed. Frame dropped.")
                    else:
                        logging.error(f"Error during recording: {e}")
                    tb_str = traceback.format_exc()
                    print (f"Traceback: {tb_str}")
                    print (f"Error: {e}")
                    continue

                except Exception as e:
                    logging.error(f"Error during recording: {e}")
                    tb_str = traceback.format_exc()
                    print (f"Traceback: {tb_str}")
                    print (f"Error: {e}")
                    continue

                audio_queue.put(data)                
        finally:
            stream.stop_stream()
            stream.close()
            audio_interface.terminate()


    def wait_audio(self):
        """
        Waits for the start and completion of the audio recording process.

        This method is responsible for:
        - Waiting for voice activity to begin recording if not yet started.
        - Waiting for voice inactivity to complete the recording.
        - Setting the audio buffer from the recorded frames.
        - Resetting recording-related attributes.

        Side effects:
        - Updates the state of the instance.
        - Modifies the audio attribute to contain the processed audio data.
        """

        self.listen_start = time.time()

        # If not yet started recording, wait for voice activity to initiate.
        if not self.is_recording and not self.frames:
            self._set_state("listening")
            self.start_recording_on_voice_activity = True

            # Wait until recording starts
            self.start_recording_event.wait()

        # If recording is ongoing, wait for voice inactivity to finish recording.
        if self.is_recording:
            self.stop_recording_on_voice_deactivity = True

            # Wait until recording stops
            self.stop_recording_event.wait()

        # Convert recorded frames to the appropriate audio format.
        audio_array = np.frombuffer(b''.join(self.frames), dtype=np.int16)
        self.audio = audio_array.astype(np.float32) / INT16_MAX_ABS_VALUE
        self.frames.clear()

        # Reset recording-related timestamps
        self.recording_stop_time = 0
        self.listen_start = 0

        self._set_state("inactive")


    def transcribe(self):
        """
        Transcribes audio captured by this class instance using the `faster_whisper` model.

        Automatically starts recording upon voice activity if not manually started using `recorder.start()`.
        Automatically stops recording upon voice deactivity if not manually stopped with `recorder.stop()`.
        Processes the recorded audio to generate transcription.

        Args:
            on_transcription_finished (callable, optional): Callback function to be executed when transcription is ready.
                If provided, transcription will be performed asynchronously, and the callback will receive the transcription 
                as its argument. If omitted, the transcription will be performed synchronously, and the result will be returned.

        Returns (if no callback is set):
            str: The transcription of the recorded audio.

        Raises:
            Exception: If there is an error during the transcription process.
        """        
        self._set_state("transcribing")
        self.parent_transcription_pipe.send((self.audio, self.language))
        status, result = self.parent_transcription_pipe.recv()
        
        self._set_state("inactive")
        if status == 'success':
            return self._preprocess_output(result)
        else:
            logging.error(result)
            raise Exception(result)


    def text(self,
             on_transcription_finished = None,
        ):
        """
        Transcribes audio captured by this class instance using the `faster_whisper` model.

        - Automatically starts recording upon voice activity if not manually started using `recorder.start()`.
        - Automatically stops recording upon voice deactivity if not manually stopped with `recorder.stop()`.
        - Processes the recorded audio to generate transcription.

        Args:
            on_transcription_finished (callable, optional): Callback function to be executed when transcription is ready.
                If provided, transcription will be performed asynchronously, and the callback will receive the transcription 
                as its argument. If omitted, the transcription will be performed synchronously, and the result will be returned.

        Returns (if not callback is set):
            str: The transcription of the recorded audio 
        """

        self.wait_audio()

        if self.is_shut_down:
            return ""

        if on_transcription_finished:
            threading.Thread(target=on_transcription_finished, args=(self.transcribe(),)).start()
        else:
            return self.transcribe()


    def start(self):
        """
        Starts recording audio directly without waiting for voice activity.
        """

        # Ensure there's a minimum interval between stopping and starting recording
        if time.time() - self.recording_stop_time < self.min_gap_between_recordings:
            logging.info("Attempted to start recording too soon after stopping.")
            return self
        
        logging.info("recording started")
        self._set_state("recording")
        self.text_storage = []
        self.realtime_stabilized_text = ""
        self.realtime_stabilized_safetext = ""
        self.wakeword_detected = False
        self.wake_word_detect_time = 0
        self.frames = []
        self.is_recording = True        
        self.recording_start_time = time.time()
        self.is_silero_speech_active = False
        self.is_webrtc_speech_active = False
        self.stop_recording_event.clear()
        self.start_recording_event.set()

        if self.on_recording_start:
            self.on_recording_start()

        return self
    

    def stop(self):
        """
        Stops recording audio.
        """

        # Ensure there's a minimum interval between starting and stopping recording
        if time.time() - self.recording_start_time < self.min_length_of_recording:
            logging.info("Attempted to stop recording too soon after starting.")
            return self
                
        logging.info("recording stopped")                
        self.is_recording = False
        self.recording_stop_time = time.time()
        self.is_silero_speech_active = False
        self.is_webrtc_speech_active = False
        self.silero_check_time = 0
        self.start_recording_event.clear()
        self.stop_recording_event.set()

        if self.on_recording_stop:
            self.on_recording_stop()

        return self


    def shutdown(self):
        """
        Safely shuts down the audio recording by stopping the recording worker and closing the audio stream.
        """


        # Force wait_audio() and text() to exit
        self.is_shut_down = True
        self.start_recording_event.set()
        self.stop_recording_event.set()

        self.shutdown_event.set()
        self.is_recording = False
        self.is_running = False

        logging.debug('Finishing recording thread')
        if self.recording_thread:
            self.recording_thread.join()

        logging.debug('Terminating reader process')
        # Give it some time to finish the loop and cleanup.
        self.reader_process.join(timeout=10) 

        if self.reader_process.is_alive():
            logging.warning("Reader process did not terminate in time. Terminating forcefully.")
            self.reader_process.terminate()
        
        logging.debug('Terminating transcription process')
        self.transcript_process.join(timeout=10) 

        if self.transcript_process.is_alive():
            logging.warning("Transcript process did not terminate in time. Terminating forcefully.")
            self.transcript_process.terminate()

        self.parent_transcription_pipe.close()

        logging.debug('Finishing realtime thread')
        if self.realtime_thread:
            self.realtime_thread.join()






    def _recording_worker(self):
        """
        The main worker method which constantly monitors the audio input for voice activity and accordingly starts/stops the recording.
        """

        logging.debug('Starting recording worker')

        try:
            was_recording = False
            delay_was_passed = False

            # Continuously monitor audio for voice activity
            while self.is_running:

                data = self.audio_queue.get()

                # Handle queue overflow
                queue_overflow_logged = False
                while self.audio_queue.qsize() > self.allowed_latency_limit:
                    if not queue_overflow_logged:
                        logging.warning(f"Audio queue size exceeds latency limit. Current size: {self.audio_queue.qsize()}. Discarding old audio chunks.")
                        queue_overflow_logged = True
                    data = self.audio_queue.get()

                if not self.is_recording:
                    # Handle not recording state

                    time_since_listen_start = time.time() - self.listen_start if self.listen_start else 0
                    wake_word_activation_delay_passed = (time_since_listen_start > self.wake_word_activation_delay)

                    # Handle wake-word timeout callback
                    if wake_word_activation_delay_passed and not delay_was_passed:
                        if self.wake_words and self.wake_word_activation_delay:
                            if self.on_wakeword_timeout:
                                self.on_wakeword_timeout()
                    delay_was_passed = wake_word_activation_delay_passed

                    # Set state and spinner text 
                    if not self.recording_stop_time:
                        if self.wake_words and wake_word_activation_delay_passed and not self.wakeword_detected:
                            self._set_state("wakeword")
                        else:
                            if self.listen_start:
                                self._set_state("listening")
                            else:
                                self._set_state("inactive")

                    # Detect wake words if applicable
                    if self.wake_words and wake_word_activation_delay_passed:
                        try:
                            pcm = struct.unpack_from("h" * self.buffer_size, data)
                            wakeword_index = self.porcupine.process(pcm)

                        except struct.error:
                            logging.error("Error unpacking audio data for wake word processing.")
                            continue
                        
                        except Exception as e:
                            logging.error(f"Wake word processing error: {e}")
                            continue
                        
                        # If a wake word is detected
                        if wakeword_index >= 0:

                            # Removing the wake word from the recording
                            samples_for_0_1_sec = int(self.sample_rate * 0.1)
                            start_index = max(0, len(self.audio_buffer) - samples_for_0_1_sec)
                            temp_samples = collections.deque(itertools.islice(self.audio_buffer, start_index, None))
                            self.audio_buffer.clear()
                            self.audio_buffer.extend(temp_samples)

                            self.wake_word_detect_time = time.time()
                            self.wakeword_detected = True
                            if self.on_wakeword_detected:
                                self.on_wakeword_detected()

                    # Check for voice activity to trigger the start of recording
                    if ((not self.wake_words or not wake_word_activation_delay_passed) and self.start_recording_on_voice_activity) or self.wakeword_detected:

                        if self._is_voice_active():
                            logging.info("voice activity detected")

                            self.start()

                            if self.is_recording:
                                self.start_recording_on_voice_activity = False

                                # Add the buffered audio to the recording frames
                                self.frames.extend(list(self.audio_buffer))
                                self.audio_buffer.clear()

                            self.silero_vad_model.reset_states()
                        else:
                            data_copy = data[:]
                            self._check_voice_activity(data_copy)

                    self.speech_end_silence_start = 0

                else:
                    # If we are currently recording

                    # Stop the recording if silence is detected after speech
                    if self.stop_recording_on_voice_deactivity:

                        if not self._is_webrtc_speech(data, True):

                            # Voice deactivity was detected, so we start measuring silence time before stopping recording
                            if self.speech_end_silence_start == 0:
                                self.speech_end_silence_start = time.time()
                            
                        else:
                            self.speech_end_silence_start = 0

                        # Wait for silence to stop recording after speech
                        if self.speech_end_silence_start and time.time() - self.speech_end_silence_start > self.post_speech_silence_duration:
                            logging.info("voice deactivity detected")
                            self.stop()


                if not self.is_recording and was_recording:
                    # Reset after stopping recording to ensure clean state
                    self.stop_recording_on_voice_deactivity = False

                if time.time() - self.silero_check_time > 0.1:
                    self.silero_check_time = 0
                
                # Handle wake word timeout (waited to long initiating speech after wake word detection)
                if self.wake_word_detect_time and time.time() - self.wake_word_detect_time > self.wake_word_timeout:
                    self.wake_word_detect_time = 0
                    if self.wakeword_detected and self.on_wakeword_timeout:
                        self.on_wakeword_timeout()
                    self.wakeword_detected = False

                was_recording = self.is_recording


                if self.is_recording:
                    self.frames.append(data)

                if not self.is_recording or self.speech_end_silence_start:
                    self.audio_buffer.append(data)	


        except Exception as e:
            logging.error(f"Unhandled exeption in _recording_worker: {e}")
            raise


    def _realtime_worker(self):
        """
        Performs real-time transcription if the feature is enabled.

        The method is responsible transcribing recorded audio frames in real-time
         based on the specified resolution interval.
        The transcribed text is stored in `self.realtime_transcription_text` and a callback
        function is invoked with this text if specified.
        """

        try:

            logging.debug('Starting realtime worker')

            # Return immediately if real-time transcription is not enabled
            if not self.enable_realtime_transcription:
                return
                
            # Continue running as long as the main process is active
            while self.is_running:

                # Check if the recording is active
                if self.is_recording:
                    
                    # Sleep for the duration of the transcription resolution
                    time.sleep(self.realtime_processing_pause)
                    
                    # Convert the buffer frames to a NumPy array
                    audio_array = np.frombuffer(b''.join(self.frames), dtype=np.int16)
                    
                    # Normalize the array to a [-1, 1] range
                    audio_array = audio_array.astype(np.float32) / INT16_MAX_ABS_VALUE

                    # Perform transcription and assemble the text
                    segments = self.realtime_model_type.transcribe(
                        audio_array,
                        language=self.language if self.language else None
                    )

                    # double check recording state because it could have changed mid-transcription
                    if self.is_recording and time.time() - self.recording_start_time > 0.5:

                        logging.debug('Starting realtime transcription')
                        self.realtime_transcription_text = " ".join(seg.text for seg in segments[0]).strip()

                        self.text_storage.append(self.realtime_transcription_text)

                        # Take the last two texts in storage, if they exist
                        if len(self.text_storage) >= 2:
                            last_two_texts = self.text_storage[-2:]
                            
                            # Find the longest common prefix between the two texts
                            prefix = os.path.commonprefix([last_two_texts[0], last_two_texts[1]])

                            # This prefix is the text that was transcripted two times in the same way
                            # Store as "safely detected text" 
                            if len(prefix) >= len(self.realtime_stabilized_safetext):
                                # Only store when longer than the previous as additional security 
                                self.realtime_stabilized_safetext = prefix

                        # Find parts of the stabilized text in the freshly transscripted text
                        matching_position = self._find_tail_match_in_text(self.realtime_stabilized_safetext, self.realtime_transcription_text)
                        if matching_position < 0:
                            if self.realtime_stabilized_safetext:
                                self._on_realtime_transcription_stabilized(self._preprocess_output(self.realtime_stabilized_safetext, True))
                            else:
                                self._on_realtime_transcription_stabilized(self._preprocess_output(self.realtime_transcription_text, True))
                        else:
                            # We found parts of the stabilized text in the transcripted text
                            # We now take the stabilized text and add only the freshly transcripted part to it
                            output_text = self.realtime_stabilized_safetext + self.realtime_transcription_text[matching_position:]

                            # This yields us the "left" text part as stabilized AND at the same time delivers fresh detected parts 
                            # on the first run without the need for two transcriptions
                            self._on_realtime_transcription_stabilized(self._preprocess_output(output_text, True))

                        # Invoke the callback with the transcribed text
                        self._on_realtime_transcription_update(self._preprocess_output(self.realtime_transcription_text, True))


                # If not recording, sleep briefly before checking again
                else:
                    time.sleep(TIME_SLEEP)

        except Exception as e:
            logging.error(f"Unhandled exeption in _realtime_worker: {e}")
            raise


    def _is_silero_speech(self, data):
        """
        Returns true if speech is detected in the provided audio data

        Args:
            data (bytes): raw bytes of audio data (1024 raw bytes with 16000 sample rate and 16 bits per sample)
        """

        self.silero_working = True
        audio_chunk = np.frombuffer(data, dtype=np.int16)
        audio_chunk = audio_chunk.astype(np.float32) / INT16_MAX_ABS_VALUE  # Convert to float and normalize
        vad_prob = self.silero_vad_model(torch.from_numpy(audio_chunk), SAMPLE_RATE).item()
        is_silero_speech_active = vad_prob > (1 - self.silero_sensitivity)
        if is_silero_speech_active:
            self.is_silero_speech_active = True
        self.silero_working = False
        return is_silero_speech_active


    def _is_webrtc_speech(self, data, all_frames_must_be_true=False):
        """
        Returns true if speech is detected in the provided audio data

        Args:
            data (bytes): raw bytes of audio data (1024 raw bytes with 16000 sample rate and 16 bits per sample)
        """
        # Number of audio frames per millisecond
        frame_length = int(self.sample_rate * 0.01)  # for 10ms frame
        num_frames = int(len(data) / (2 * frame_length))
        speech_frames = 0        

        for i in range(num_frames):
            start_byte = i * frame_length * 2
            end_byte = start_byte + frame_length * 2
            frame = data[start_byte:end_byte]
            if self.webrtc_vad_model.is_speech(frame, self.sample_rate):
                speech_frames += 1
                if not all_frames_must_be_true:
                    return True
        if all_frames_must_be_true:
            return speech_frames == num_frames
        else:
            return False
    
        
    def _check_voice_activity(self, data):
        """
        Initiate check if voice is active based on the provided data.

        Args:
            data: The audio data to be checked for voice activity.
        """
        self.is_webrtc_speech_active = self._is_webrtc_speech(data)
        
        # First quick performing check for voice activity using WebRTC
        if self.is_webrtc_speech_active:
            
            if not self.silero_working:
                self.silero_working = True

                # Run the intensive check in a separate thread
                threading.Thread(target=self._is_silero_speech, args=(data,)).start()

    
    def _is_voice_active(self):
        """
        Determine if voice is active.

        Returns:
            bool: True if voice is active, False otherwise.
        """
        return self.is_webrtc_speech_active and self.is_silero_speech_active


    def _set_state(self, new_state):
        """
        Update the current state of the recorder and execute corresponding state-change callbacks.

        Args:
            new_state (str): The new state to set. 

        """
        # Check if the state has actually changed
        if new_state == self.state:
            return
        
        # Store the current state for later comparison
        old_state = self.state
        
        # Update to the new state
        self.state = new_state

        # Execute callbacks based on transitioning FROM a particular state
        if old_state == "listening":
            if self.on_vad_detect_stop:
                self.on_vad_detect_stop()
        elif old_state == "wakeword":
            if self.on_wakeword_detection_end:
                self.on_wakeword_detection_end()

        # Execute callbacks based on transitioning TO a particular state
        if new_state == "listening":
            if self.on_vad_detect_start:
                self.on_vad_detect_start()
            self._set_spinner("speak now")
            if self.spinner:
                self.halo._interval = 250
        elif new_state == "wakeword":
            if self.on_wakeword_detection_start:
                self.on_wakeword_detection_start()
            self._set_spinner(f"say {self.wake_words}")
            if self.spinner:
                self.halo._interval = 500
        elif new_state == "transcribing":
            if self.on_transcription_start:
                self.on_transcription_start()
            self._set_spinner("transcribing")
            if self.spinner:
                self.halo._interval = 50
        elif new_state == "recording":
            self._set_spinner("recording")
            if self.spinner:
                self.halo._interval = 100
        elif new_state == "inactive":
            if self.spinner and self.halo:
                self.halo.stop()
                self.halo = None


    def _set_spinner(self, text):
        """
        Update the spinner's text or create a new spinner with the provided text.

        Args:
            text (str): The text to be displayed alongside the spinner.
        """
        if self.spinner:
            # If the Halo spinner doesn't exist, create and start it
            if self.halo is None:
                self.halo = halo.Halo(text=text)
                self.halo.start()
            # If the Halo spinner already exists, just update the text
            else:
                self.halo.text = text


    def _preprocess_output(self, text, preview=False):
        """
        Preprocesses the output text by removing any leading or trailing whitespace,
        converting all whitespace sequences to a single space character, and capitalizing
        the first character of the text.

        Args:
            text (str): The text to be preprocessed.

        Returns:
            str: The preprocessed text.
        """
        text = re.sub(r'\s+', ' ', text.strip())

        if self.ensure_sentence_starting_uppercase:
            if text:
                text = text[0].upper() + text[1:]

        # Ensure the text ends with a proper punctuation if it ends with an alphanumeric character
        if not preview:
            if self.ensure_sentence_ends_with_period:
                if text and text[-1].isalnum():
                    text += '.'

        return text


    def _find_tail_match_in_text(self, text1, text2, length_of_match=10):
        """
        Find the position where the last 'n' characters of text1 match with a substring in text2.
        
        This method takes two texts, extracts the last 'n' characters from text1 (where 'n' is determined
        by the variable 'length_of_match'), and searches for an occurrence of this substring in text2,
        starting from the end of text2 and moving towards the beginning.

        Parameters:
        - text1 (str): The text containing the substring that we want to find in text2.
        - text2 (str): The text in which we want to find the matching substring.
        - length_of_match(int): The length of the matching string that we are looking for

        Returns:
        int: The position (0-based index) in text2 where the matching substring starts.
            If no match is found or either of the texts is too short, returns -1.
        """
        
        # Check if either of the texts is too short
        if len(text1) < length_of_match or len(text2) < length_of_match:
            return -1
        
        # The end portion of the first text that we want to compare
        target_substring = text1[-length_of_match:]
        
        # Loop through text2 from right to left
        for i in range(len(text2) - length_of_match + 1):
            # Extract the substring from text2 to compare with the target_substring
            current_substring = text2[len(text2) - i - length_of_match:len(text2) - i]
            
            # Compare the current_substring with the target_substring
            if current_substring == target_substring:
                return len(text2) - i  # Position in text2 where the match starts
        
        return -1
    

    def _on_realtime_transcription_stabilized(self, text):
        """
        Callback method invoked when the real-time transcription stabilizes.

        This method is called internally when the transcription text is considered "stable" 
        meaning it's less likely to change significantly with additional audio input. It 
        notifies any registered external listener about the stabilized text if recording is 
        still ongoing. This is particularly useful for applications that need to display 
        live transcription results to users and want to highlight parts of the transcription 
        that are less likely to change.

        Args:
            text (str): The stabilized transcription text.
        """        
        if self.on_realtime_transcription_stabilized:
            if self.is_recording:
                self.on_realtime_transcription_stabilized(text)


    def _on_realtime_transcription_update(self, text):
        """
        Callback method invoked when there's an update in the real-time transcription.

        This method is called internally whenever there's a change in the transcription text,
        notifying any registered external listener about the update if recording is still 
        ongoing. This provides a mechanism for applications to receive and possibly display 
        live transcription updates, which could be partial and still subject to change.

        Args:
            text (str): The updated transcription text.
        """        
        if self.on_realtime_transcription_update:
            if self.is_recording:
                self.on_realtime_transcription_update(text)


    def __enter__(self):
        """
        Method to setup the context manager protocol.

        This enables the instance to be used in a `with` statement, ensuring proper 
        resource management. When the `with` block is entered, this method is 
        automatically called.

        Returns:
            self: The current instance of the class.
        """
        return self


    def __exit__(self, exc_type, exc_value, traceback):
        """
        Method to define behavior when the context manager protocol exits.

        This is called when exiting the `with` block and ensures that any necessary 
        cleanup or resource release processes are executed, such as shutting down 
        the system properly.

        Args:
            exc_type (Exception or None): The type of the exception that caused the context to be exited, if any.
            exc_value (Exception or None): The exception instance that caused the context to be exited, if any.
            traceback (Traceback or None): The traceback corresponding to the exception, if any.
        """
        self.shutdown()