add nemo_asr and silero_vad Engine

.gitignore

@@ -4,3 +4,4 @@ app/__pycache__/
 __pycache__/
 logs/
 .continue/
+tmp/

app.py

@@ -39,7 +39,7 @@ from app.ui_utils import (
     get_custom_theme,
     on_file_load
 )
-
+import nemo.collections.asr as nemo_asr
 # --------------------------------------------------------
 # Initialization
 # --------------------------------------------------------
@@ -47,6 +47,24 @@ reset_all_active_session_hash_code()
 
 theme,css_style = get_custom_theme()
 
+from omegaconf import OmegaConf
+cfg = OmegaConf.load('app/config.yaml')
+# logger.info(f'Hydra config: {OmegaConf.to_yaml(cfg)}')
+from app.canary_speech_engine import CanarySpeechEngine
+from app.silero_vad_engine import Silero_Vad_Engine
+from app.streaming_audio_processor import StreamingAudioProcessor,StreamingAudioProcessorConfig
+
+
+asr_model = nemo_asr.models.ASRModel.from_pretrained(cfg.pretrained_name)
+canary_speech_engine = CanarySpeechEngine(asr_model,cfg)
+silero_vad_engine = Silero_Vad_Engine()
+streaming_audio_processor_config = StreamingAudioProcessorConfig(
+    read_size=4000,
+    silence_threshold_chunks=1
+)
+streamer = StreamingAudioProcessor(speech_engine=canary_speech_engine,vad_engine=silero_vad_engine,cfg=streaming_audio_processor_config)
+
+
 with gr.Blocks(theme=theme, css=css_style) as demo:
     session_hash_code = gr.State()
     session_hash_code_box = gr.Textbox(label="Session ID", interactive=False, visible=DEBUG)
@@ -125,7 +143,7 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
 
             webrtc_stream.stream(
                 fn=read_and_stream_audio,
-                inputs=[active_filepath, session_hash_code, stop_streaming_flags],
+                inputs=[active_filepath, session_hash_code, stop_streaming_flags,gr.State(streaming_audio_processor_config.read_size)],
                 outputs=[webrtc_stream],
                 trigger=start_stream_button.click,
                 concurrency_id="audio_stream",
@@ -273,7 +291,7 @@ with gr.Blocks(theme=theme, css=css_style) as demo:
                         yield f"Starting {task_type.lower()}...\n\n",gr.update(visible=False),gr.update(visible=True)
 
                         # Boucle sur le générateur de `task()`
-                        for msg in task(session_hash_code):
+                        for msg in task(session_hash_code,streamer=streamer):
                             accumulated += msg
                             yield accumulated,gr.update(visible=False),gr.update(visible=True)
 

app/canary_speech_engine.py

@@ -0,0 +1,288 @@
+import time
+from typing import Optional, Tuple
+from app.interfaces import IStreamingSpeechEngine
+import numpy as np
+import torch
+import gc
+from omegaconf import OmegaConf
+
+
+from nemo.collections.asr.models.aed_multitask_models import lens_to_mask
+from nemo.collections.asr.parts.submodules.aed_decoding import (
+    GreedyBatchedStreamingAEDComputer,
+    return_decoder_input_ids,
+)
+from nemo.collections.asr.parts.submodules.multitask_decoding import (
+    AEDStreamingDecodingConfig,
+    MultiTaskDecodingConfig,
+)
+# from nemo.collections.asr.parts.utils.eval_utils import cal_write_wer # Not used
+from nemo.collections.asr.parts.utils.streaming_utils import (
+    ContextSize,
+    StreamingBatchedAudioBuffer,
+)
+import nemo.collections.asr as nemo_asr
+from nemo.collections.asr.parts.utils.transcribe_utils import (
+    get_inference_device,
+    get_inference_dtype,
+)
+from app.logger_config import (
+    logger as logging,
+    DEBUG
+) 
+
+
+def make_divisible_by(num: int, factor: int) -> int:
+    """Make num divisible by factor"""
+    return (num // factor) * factor
+
+
+class CanarySpeechEngine(IStreamingSpeechEngine):
+    """
+    Encapsulates the state and logic for streaming audio transcription
+    using an internally loaded Canary model.
+    """
+    def __init__(self,asr_model, cfg: OmegaConf):
+        """
+        Initializes the speech engine and loads the ASR model.
+        
+        Args:
+            cfg: An OmegaConf object containing 'model' and 'streaming' configs.
+        """
+        self.cfg = cfg # Store the full config
+        
+        # Setup device and dtype from config
+        self.map_location = get_inference_device(cuda=self.cfg.cuda, allow_mps=self.cfg.allow_mps)
+        self.compute_dtype = get_inference_dtype(self.cfg.compute_dtype, device=self.map_location)
+        logging.info(f"Inference will be on device: {self.map_location} with dtype: {self.compute_dtype}")
+
+        # Load the model internally
+        asr_model, _ = self._setup_model(asr_model,self.cfg, self.map_location)
+        self.asr_model = asr_model
+        
+        self.full_transcription = [] # Stores finalized segments
+        self._setup_streaming_params()
+        
+        # The initial full reset (buffer + decoder)
+        self.reset()
+        
+        logging.info("CanarySpeechEngine initialized and ready.")
+        logging.info(f"Model-adjusted chunk size: {self.context_samples.chunk} samples.")
+
+    def _setup_model(self,asr_model, model_cfg: OmegaConf, map_location: str):
+        """Loads the pretrained ASR model and configures it for inference."""
+        logging.info(f"Loading model {model_cfg.pretrained_name}...")
+        start_time = time.time()
+        try:
+            asr_model = asr_model.to(map_location)
+            asr_model.eval()
+            
+            # Change decoding strategy to greedy for streaming
+            if hasattr(asr_model, 'change_decoding_strategy'):
+                multitask_decoding = MultiTaskDecodingConfig()
+                multitask_decoding.strategy = "greedy"
+                asr_model.change_decoding_strategy(multitask_decoding)
+                logging.info("Model decoding strategy set to 'greedy'.")
+
+            if map_location == "cuda":
+                torch.cuda.synchronize()
+            
+            end_time = time.time()
+            logging.info("Model loaded successfully.")
+            load_time = end_time - start_time
+            logging.info("\n" + "="*30)
+            logging.info(f"Total model load time: {load_time:.2f} seconds")
+            logging.info("="*30)
+            return asr_model, None
+
+        except Exception as e:
+            logging.error(f"Error loading model: {e}")
+            logging.error("Ensure NeMo is installed (pip install nemo_toolkit['asr'])")
+            return None, None
+    
+    def _setup_streaming_params(self):
+        """Helper to calculate model-specific streaming parameters."""
+        model_cfg = self.asr_model.cfg
+        audio_sample_rate = model_cfg.preprocessor['sample_rate']
+        feature_stride_sec = model_cfg.preprocessor['window_stride']
+        features_per_sec = 1.0 / feature_stride_sec
+        self.encoder_subsampling_factor = self.asr_model.encoder.subsampling_factor
+        
+        self.features_frame2audio_samples = make_divisible_by(
+            int(audio_sample_rate * feature_stride_sec), factor=self.encoder_subsampling_factor
+        )
+        encoder_frame2audio_samples = self.features_frame2audio_samples * self.encoder_subsampling_factor
+
+        # Use self.cfg.streaming instead of self.streaming_cfg
+        streaming_cfg = self.cfg
+        self.context_encoder_frames = ContextSize(
+            left=int(streaming_cfg.left_context_secs * features_per_sec / self.encoder_subsampling_factor),
+            chunk=int(streaming_cfg.chunk_secs * features_per_sec / self.encoder_subsampling_factor),
+            right=int(streaming_cfg.right_context_secs * features_per_sec / self.encoder_subsampling_factor),
+        )
+        self.context_samples = ContextSize(
+            left=self.context_encoder_frames.left * encoder_frame2audio_samples,
+            chunk=self.context_encoder_frames.chunk * encoder_frame2audio_samples,
+            right=self.context_encoder_frames.right * encoder_frame2audio_samples,
+        )
+
+    def _reset_decoder_state(self):
+        """
+        Resets ONLY the decoder state, preserving the audio buffer.
+        This prevents slowdowns on long audio streams.
+        """
+        start_time = time.perf_counter()
+        logging.debug("--- Resetting decoder state (audio buffer preserved) ---")
+        
+        # Reset tracking for this segment
+        self.last_transcription = ""
+        self.chunk_count = 0 
+        batch_size = 1 # Hardcoded for this script
+        
+        # Use self.cfg.streaming instead of self.streaming_cfg
+        streaming_cfg = self.cfg
+
+        # 1. Recreate the initial prompt for the decoder
+        self.decoder_input_ids = return_decoder_input_ids(streaming_cfg, self.asr_model)
+        
+        # 2. Recreate the "computer" object that manages decoding
+        self.decoding_computer = GreedyBatchedStreamingAEDComputer(
+            self.asr_model,
+            frame_chunk_size=self.context_encoder_frames.chunk,
+            decoding_cfg=streaming_cfg.decoding,
+        )
+
+        # 3. Recreate an EMPTY STATE object (model_state)
+        self.model_state = GreedyBatchedStreamingAEDComputer.initialize_aed_model_state(
+            asr_model=self.asr_model,
+            decoder_input_ids=self.decoder_input_ids,
+            batch_size=batch_size,
+            context_encoder_frames=self.context_encoder_frames,
+            chunk_secs=streaming_cfg.chunk_secs,
+            right_context_secs=streaming_cfg.right_context_secs,
+        )
+
+        # Clear CUDA cache if possible
+        if torch.cuda.is_available():
+            gc.collect()
+            torch.cuda.empty_cache()
+            torch.cuda.synchronize()
+
+        end_time = time.perf_counter()
+        duration_ms = (end_time - start_time) * 1000  # Convert to milliseconds
+        logging.debug(f"--- Decoder reset finished in {duration_ms:.2f} ms ---")
+
+    def reset(self):
+        """
+        Resets the transcriber's state completely (audio buffer + decoder state).
+        Called only on initialization.
+        """
+        start_time = time.perf_counter()
+        logging.debug("--- FULL RESET (Audio Buffer + Decoder State) ---")
+        
+        # Operation 1: Reset the decoder (this now includes GC)
+        self._reset_decoder_state()
+        
+        # Operation 2: Reset the audio buffer
+        self.buffer = StreamingBatchedAudioBuffer(
+            batch_size=1, # Hardcoded for this script
+            context_samples=self.context_samples,
+            dtype=torch.float32,
+            device=self.map_location,
+        )
+        
+        end_time = time.perf_counter()
+        duration_ms = (end_time * 1000)
+        logging.debug(f"--- RESET Complete: took {duration_ms:.2f} ms ---")
+
+    def transcribe_chunk(self, chunk: np.ndarray, is_last_chunk: bool = False) -> Tuple[str, str]:
+        """
+        Processes a single audio chunk and returns the newly predicted text.
+
+        Returns:
+            Tuple[str, str]: 
+                (current_transcription: The full transcription for the current segment,
+                 new_text: The newly appended text since the last chunk)
+        """
+        start_time = time.perf_counter()
+        self.chunk_count += 1
+        
+        # Preprocess audio
+        signal = torch.from_numpy(chunk.astype(np.float32) / 32768.0)
+        audio_batch = signal.unsqueeze(0).to(self.map_location)
+        audio_batch_lengths = torch.tensor([signal.shape[0]], device=self.map_location)
+
+        # 1. Add the chunk to the persistent buffer
+        self.buffer.add_audio_batch_(
+            audio_batch,
+            audio_lengths=audio_batch_lengths,
+            is_last_chunk=is_last_chunk,
+            is_last_chunk_batch=torch.tensor([is_last_chunk], device=self.map_location)
+        )
+
+        self.model_state.is_last_chunk_batch = torch.tensor([is_last_chunk], device=self.map_location)
+
+        # 2. Pass the buffer to the encoder
+        _, encoded_len, enc_states, _ = self.asr_model(
+            input_signal=self.buffer.samples, input_signal_length=self.buffer.context_size_batch.total()
+        )
+        encoder_context_batch = self.buffer.context_size_batch.subsample(factor=self.features_frame2audio_samples * self.encoder_subsampling_factor)
+        encoded_len_no_rc = encoder_context_batch.left + encoder_context_batch.chunk
+        encoded_length_corrected = torch.where(self.model_state.is_last_chunk_batch, encoded_len, encoded_len_no_rc)
+        encoder_input_mask = lens_to_mask(encoded_length_corrected, enc_states.shape[1]).to(enc_states.dtype)
+
+        # 3. Pass to the decoding computer
+        self.model_state = self.decoding_computer(
+            encoder_output=enc_states,
+            encoder_output_len=encoded_length_corrected,
+            encoder_input_mask=encoder_input_mask,
+            prev_batched_state=self.model_state,
+        )
+
+        # 4. Calculate the new text
+        current_tokens = self.model_state.pred_tokens_ids[0, self.decoder_input_ids.size(-1): self.model_state.current_context_lengths[0]]
+        
+        # OPTIMIZATION: Move tokens to CPU before converting to list
+        current_transcription = self.asr_model.tokenizer.ids_to_text(current_tokens.cpu().tolist()).strip()
+        
+        # Calculate the NEW text by "subtracting" the old history
+        new_text = ""
+        if current_transcription.startswith(self.last_transcription):
+             new_text = current_transcription[len(self.last_transcription):]
+        else:
+            # Model corrected itself, send the full new transcription
+            new_text = current_transcription
+        
+        # Memorize the FULL current transcription as the new history
+        if new_text:
+            self.last_transcription = current_transcription
+        
+        end_time = time.perf_counter()
+        duration_ms = (end_time - start_time) * 1000
+        # logging.info(f"--- transcribe_chunk: took {duration_ms:.2f} ms ---")
+
+        # Return both the full segment transcription and the new diff
+        return current_transcription, new_text
+    
+    def finalize_segment(self):
+        """
+        Finalizes the current transcription segment (e.g., on silence)
+        and adds it to the full history.
+        """
+        if self.last_transcription:
+            self.full_transcription.append(self.last_transcription)
+        self.last_transcription = ""
+        # We must reset the decoder state to start a new segment
+        self._reset_decoder_state()
+
+    def get_full_transcription(self) -> str:
+        """
+        Returns the full accumulated transcription from all finalized segments.
+        Does NOT include the currently active (unfinalized) segment.
+        """
+        return " ".join(self.full_transcription)
+
+    def get_current_segment_text(self) -> str:
+        """Returns the text of the segment currently being transcribed."""
+        return self.last_transcription
+

app/config.yaml

@@ -0,0 +1,28 @@
+pretrained_name: 'nvidia/canary-1b-v2'
+model_path: null
+batch_size: 32
+
+chunk_secs: 1.0
+left_context_secs: 20.0
+right_context_secs: 0.5
+cuda: null
+allow_mps: true
+compute_dtype: null
+matmul_precision: high
+decoding:
+  streaming_policy: alignatt
+  alignatt_thr: 8.0
+  waitk_lagging: 2
+  exclude_sink_frames: 8
+  xatt_scores_layer: -2
+  max_tokens_per_alignatt_step: 30
+  max_generation_length: 512
+  use_avgpool_for_alignatt: false
+  hallucinations_detector: true
+prompt:
+  pnc: 'no'
+  task: asr
+  source_lang: fr
+  target_lang: fr
+  timestamps: yes 
+debug_mode: false

app/interfaces.py

@@ -0,0 +1,27 @@
+from abc import ABC, abstractmethod
+from typing import Any, Iterable
+
+class IVoiceActivityEngine(ABC):
+    """Contract for a Voice Activity Detector (VAD)."""
+
+    @abstractmethod
+    def __call__(self, audio_chunk: bytes) -> bool:
+        """
+        Analyzes an audio chunk and returns True if speech is detected, 
+        False otherwise.
+        """
+        pass
+
+class IStreamingSpeechEngine(ABC):
+    """Contract for a streaming transcription service."""
+
+    @abstractmethod
+    def transcribe_chunk(self, audio_chunk: bytes) -> str:
+        """Processes an audio chunk and returns a transcription (partial or final)."""
+        pass
+    
+    @abstractmethod
+    def finalize_segment(self) -> str:
+        """Called at the end of the stream to get the final transcription."""
+        pass
+

app/logger_config.py

@@ -5,10 +5,10 @@ load_dotenv(find_dotenv())
 import logging
 from logging.handlers import RotatingFileHandler
 import os
-# from nemo.utils.nemo_logging import Logger
+from nemo.utils.nemo_logging import Logger
 
-# nemo_logger = Logger()
-# nemo_logger.remove_stream_handlers()
+nemo_logger = Logger()
+nemo_logger.remove_stream_handlers()
 
 DEBUG = os.getenv("DEBUG", "false").lower() == "true"
 # Create "logs" directory if it doesn't exist

app/silero_vad_engine.py

@@ -0,0 +1,189 @@
+import os
+import subprocess
+import torch
+import numpy as np
+import onnxruntime
+import warnings
+from app.interfaces import IVoiceActivityEngine
+from app.logger_config import (
+    logger as logging,
+    DEBUG
+) 
+
+class VoiceActivityDetection():
+
+    def __init__(self, force_onnx_cpu=True):
+        logging.info("Initializing VoiceActivityDetection...")
+        path = self.download()
+
+        opts = onnxruntime.SessionOptions()
+        opts.log_severity_level = 3  # Suppress ONNX runtime's own logs
+
+        opts.inter_op_num_threads = 1
+        opts.intra_op_num_threads = 1
+
+        try:
+            if force_onnx_cpu and 'CPUExecutionProvider' in onnxruntime.get_available_providers():
+                self.session = onnxruntime.InferenceSession(path, providers=['CPUExecutionProvider'], sess_options=opts)
+                logging.info("ONNX VAD session created with CPUExecutionProvider.")
+            else:
+                self.session = onnxruntime.InferenceSession(path, providers=['CUDAExecutionProvider'], sess_options=opts)
+                logging.info("ONNX VAD session created with CUDAExecutionProvider.")
+        except Exception as e:
+            logging.critical(f"Failed to create ONNX InferenceSession: {e}", exc_info=True)
+            raise
+
+        self.reset_states()
+        if '16k' in path:
+            logging.warning('This VAD model supports only 16000 sampling rate!')
+            self.sample_rates = [16000]
+        else:
+            logging.info("VAD model supports 8000Hz and 16000Hz.")
+            self.sample_rates = [8000, 16000]
+
+    def _validate_input(self, x, sr: int):
+        if x.dim() == 1:
+            x = x.unsqueeze(0)
+        if x.dim() > 2:
+            logging.error(f"Too many dimensions for input audio chunk: {x.dim()}")
+            raise ValueError(f"Too many dimensions for input audio chunk {x.dim()}")
+
+        if sr != 16000 and (sr % 16000 == 0):
+            step = sr // 16000
+            x = x[:,::step]
+            sr = 16000
+            logging.debug(f"Downsampled input audio to 16000Hz from {sr}Hz.")
+
+        if sr not in self.sample_rates:
+            logging.error(f"Unsupported sampling rate: {sr}. Supported: {self.sample_rates}")
+            raise ValueError(f"Supported sampling rates: {self.sample_rates} (or multiply of 16000)")
+        
+        return x, sr
+
+    def reset_states(self, batch_size=1):
+        logging.debug(f"Resetting VAD states for batch_size: {batch_size}")
+        self._state = torch.zeros((2, batch_size, 128)).float()
+        self._context = torch.zeros(0)
+        self._last_sr = 0
+        self._last_batch_size = 0
+
+    def __call__(self, x, sr: int):
+
+        x, sr = self._validate_input(x, sr)
+        num_samples = 512 if sr == 16000 else 256
+
+        if x.shape[-1] != num_samples:
+            logging.error(f"Invalid audio chunk size: {x.shape[-1]}. Expected {num_samples} for {sr}Hz.")
+            raise ValueError(f"Provided number of samples is {x.shape[-1]} (Supported values: 256 for 8000 sample rate, 512 for 16000)")
+
+        batch_size = x.shape[0]
+        context_size = 64 if sr == 16000 else 32
+
+        if not self._last_batch_size:
+            logging.debug("First call, resetting states.")
+            self.reset_states(batch_size)
+        if (self._last_sr) and (self._last_sr != sr):
+            logging.warning(f"Sample rate changed ({self._last_sr}Hz -> {sr}Hz). Resetting states.")
+            self.reset_states(batch_size)
+        if (self._last_batch_size) and (self._last_batch_size != batch_size):
+            logging.warning(f"Batch size changed ({self._last_batch_size} -> {batch_size}). Resetting states.")
+            self.reset_states(batch_size)
+
+        if not len(self._context):
+            self._context = torch.zeros(batch_size, context_size)
+
+        x = torch.cat([self._context, x], dim=1)
+        if sr in [8000, 16000]:
+            ort_inputs = {'input': x.numpy(), 'state': self._state.numpy(), 'sr': np.array(sr, dtype='int64')}
+            ort_outs = self.session.run(None, ort_inputs)
+            out, state = ort_outs
+            self._state = torch.from_numpy(state)
+        else:
+            # This should be caught by _validate_input, but as a safeguard:
+            logging.critical(f"Unexpected sample rate in VAD __call__: {sr}")
+            raise ValueError()
+
+        self._context = x[..., -context_size:]
+        self._last_sr = sr
+        self._last_batch_size = batch_size
+
+        out = torch.from_numpy(out)
+        return out
+
+    def audio_forward(self, x, sr: int):
+        outs = []
+        x, sr = self._validate_input(x, sr)
+        self.reset_states()
+        num_samples = 512 if sr == 16000 else 256
+
+        if x.shape[1] % num_samples:
+            pad_num = num_samples - (x.shape[1] % num_samples)
+            logging.debug(f"Padding audio input with {pad_num} samples.")
+            x = torch.nn.functional.pad(x, (0, pad_num), 'constant', value=0.0)
+
+        for i in range(0, x.shape[1], num_samples):
+            wavs_batch = x[:, i:i+num_samples]
+            out_chunk = self.__call__(wavs_batch, sr)
+            outs.append(out_chunk)
+
+        stacked = torch.cat(outs, dim=1)
+        return stacked.cpu()
+
+    @staticmethod
+    def download(model_url="https://github.com/snakers4/silero-vad/raw/v5.0/files/silero_vad.onnx"):
+        target_dir = os.path.expanduser("~/.cache/silero_vad/")
+        os.makedirs(target_dir, exist_ok=True)
+        model_filename = os.path.join(target_dir, "silero_vad.onnx")
+
+        if not os.path.exists(model_filename):
+            logging.info(f"Downloading VAD model to {model_filename}...")
+            try:
+                subprocess.run(["wget", "-O", model_filename, model_url], check=True)
+                logging.info("VAD model downloaded successfully.")
+            except subprocess.CalledProcessError as e:
+                logging.critical(f"Failed to download the model using wget: {e}")
+                raise
+        else:
+            logging.info(f"VAD model already exists at {model_filename}.")
+        return model_filename
+
+
+class Silero_Vad_Engine(IVoiceActivityEngine):
+    def __init__(self, threshold :float =0.5, frame_rate: int =16000):
+        """
+        Initializes the Silero_Vad_Engine with a voice activity detection model and a threshold.
+
+        Args:
+            threshold (float, optional): The probability threshold for detecting voice activity. Defaults to 0.5.
+        """
+        logging.info(f"Initializing Silero_Vad_Engine with threshold: {threshold} and frame_rate: {frame_rate}Hz.")
+        self.model = VoiceActivityDetection()
+        self.threshold = threshold
+        self.frame_rate = frame_rate
+
+    def __call__(self, audio_frame):
+        """
+        Determines if the given audio frame contains speech by comparing the detected speech probability against
+        the threshold.
+
+        Args:
+            audio_frame (np.ndarray): The audio frame to be analyzed for voice activity. It is expected to be a
+                                      NumPy array of audio samples.
+
+        Returns:
+            bool: True if the speech probability exceeds the threshold, indicating the presence of voice activity;
+                  False otherwise.
+        """
+        # Convert frame to tensor
+        audio_tensor = torch.from_numpy(audio_frame.copy())
+        
+        # Get speech probabilities
+        speech_probs = self.model.audio_forward(audio_tensor, self.frame_rate)[0]
+        
+        # Check against threshold
+        is_speech = torch.any(speech_probs > self.threshold).item()
+        
+        logging.debug(f"VAD check result: {is_speech} (Max prob: {torch.max(speech_probs).item():.4f})")
+        
+        return is_speech
+

app/streaming_audio_processor.py

@@ -0,0 +1,173 @@
+import numpy as np
+from app.logger_config import (
+    logger as logging,
+    DEBUG
+) 
+from app.interfaces import IStreamingSpeechEngine,IVoiceActivityEngine
+
+from dataclasses import dataclass
+
+@dataclass
+class StreamingAudioProcessorConfig:
+    """Configuration settings for the StreamingAudioProcessor."""
+    read_size: int = 8000
+    silence_threshold_chunks: int = 2
+    sample_rate: int = 16000
+    # Add other streaming-related parameters here
+    # e.g., vad_padding_chunks: int = 0
+
+
+class StreamingAudioProcessor:
+    """
+    Manages streaming transcription by combining a speech engine
+    and a voice activity detector (VAD).
+
+    This class handles internal audio buffering and VAD state.
+    """
+
+    def __init__(self, speech_engine: IStreamingSpeechEngine, vad_engine :IVoiceActivityEngine, cfg : StreamingAudioProcessorConfig):
+        """
+        Initializes the streaming processor.
+
+        Args:
+            speech_engine: The ASR speech engine (must have .transcribe_chunk() and .reset()).
+            vad_engine: The VAD engine (returns True/False for a chunk).
+            cfg: The configuration object for this processor.
+        """
+        logging.info("Initializing StreamingAudioProcessor...")
+        self.speech_engine = speech_engine
+        self.vad_engine = vad_engine
+
+        # Store config
+        self.VAD_SAMPLE_RATE = cfg.sample_rate
+        self.read_size = cfg.read_size
+        self.SILENCE_THRESHOLD_CHUNKS = cfg.silence_threshold_chunks
+        
+        # Internal buffer state (Optimized: using numpy array)
+        self.internal_buffer = np.array([], dtype='int16')
+        
+        # Internal logic state
+        self.is_first_logical_chunk = True
+        self.logical_chunk_size = self.speech_engine.context_samples.chunk
+        self.initial_logical_chunk_size = self.speech_engine.context_samples.chunk + self.speech_engine.context_samples.right
+        
+        # Internal VAD state
+        self.silent_chunks_count = 0
+        self.chunks_count = 0
+
+        logging.info(f"  Config: VAD Sample Rate={self.VAD_SAMPLE_RATE}Hz")
+        logging.info(f"  Config: Physical Read Size={self.read_size} samples")
+        logging.info(f"  Config: Silence Threshold={self.SILENCE_THRESHOLD_CHUNKS} chunks")
+        logging.info(f"  Config: Initial Logical Chunk={self.initial_logical_chunk_size} samples")
+        logging.info(f"  Config: Subsequent Logical Chunk={self.logical_chunk_size} samples")
+
+
+    def _append_to_buffer(self, chunk_np, asr_chunk_len):
+        """
+        Appends a NumPy chunk to the internal buffer and returns a logical chunk if ready.
+        (Optimized to use numpy concatenation).
+        """
+        logging.debug(f"Appending {len(chunk_np)} samples to internal buffer (current size: {len(self.internal_buffer)}).")
+        self.internal_buffer = np.concatenate((self.internal_buffer, chunk_np))
+        
+        if len(self.internal_buffer) >= asr_chunk_len:
+            asr_signal_chunk = self.internal_buffer[:asr_chunk_len]
+            self.internal_buffer = self.internal_buffer[asr_chunk_len:]
+            logging.debug(f"Extracted logical chunk of {len(asr_signal_chunk)} samples. Buffer remaining: {len(self.internal_buffer)}.")
+            return asr_signal_chunk
+        else:
+            logging.debug(f"Buffer size ({len(self.internal_buffer)}) < target ({asr_chunk_len}). Holding.")
+            return None
+
+    def _flush_and_reset(self):
+        """
+        Flushes the remaining buffer to the transcriber, resets the state,
+        and returns the last transcribed text.
+        """
+        new_text = ""
+        if len(self.internal_buffer) > 0:
+            # Buffer is already a numpy array
+            final_segment_chunk = self.internal_buffer
+            logging.info(f"Flushing segment remainder of {len(final_segment_chunk)} samples.")
+            seg, new_text = self.speech_engine.transcribe_chunk(final_segment_chunk, is_last_chunk=True)
+        else:
+            # Buffer is empty, but send a silent "flush"
+            # to force the transcriber to finalize its internal state.
+            logging.info("Buffer empty, sending silent flush to finalize segment.")
+            flush_chunk = np.zeros(self.logical_chunk_size, dtype='int16')
+            seg, new_text = self.speech_engine.transcribe_chunk(flush_chunk, is_last_chunk=True)
+
+        # Full state reset
+        logging.debug("Resetting speech engine state...")
+        self.speech_engine.reset() # Resets the speech engine (decoder state)
+        
+        logging.debug("Resetting internal buffer and VAD state.")
+        self.internal_buffer = np.array([], dtype='int16') # Reset buffer
+        self.is_first_logical_chunk = True
+        self.silent_chunks_count = 0
+        
+        return new_text
+
+    def process_chunk(self, chunk: np.ndarray):
+        """
+        Processes a single physical chunk (e.g., 8000 samples).
+        Manages VAD, buffering, and transcription.
+        
+        Args:
+            chunk (np.ndarray): The audio chunk (int16).
+        
+        Returns:
+            list: A list of new transcribed text segments.
+                  (Often empty, may contain one or more segments).
+        """
+        new_text_segments = []
+        self.chunks_count += 1
+        logging.debug(f"--- Processing Physical Chunk {self.chunks_count} ---")
+        
+        # --- 1. VAD Logic ---
+        has_speech = self.vad_engine(chunk) 
+        logging.debug(f"VAD result: {'SPEECH' if has_speech else 'SILENCE'}")
+
+        if has_speech:
+            self.silent_chunks_count = 0
+        else:
+            self.silent_chunks_count += 1
+            logging.debug(f"Silent chunks count: {self.silent_chunks_count}/{self.SILENCE_THRESHOLD_CHUNKS}")
+            
+        silence_reset = self.silent_chunks_count >= self.SILENCE_THRESHOLD_CHUNKS
+
+        # --- 2. Buffering & Transcription Logic ---
+        target_size = self.initial_logical_chunk_size if self.is_first_logical_chunk else self.logical_chunk_size
+        asr_chunk_np = self._append_to_buffer(chunk, target_size) # Now returns np.ndarray or None
+
+        if asr_chunk_np is not None:
+            logging.debug(f"Sending logical chunk (size: {len(asr_chunk_np)}) to speech engine...")
+            seg, new_text = self.speech_engine.transcribe_chunk(asr_chunk_np, is_last_chunk=False)
+            if new_text:
+                logging.info(f"Received new text segment: '{new_text}'")
+                new_text_segments.append(new_text)
+            self.is_first_logical_chunk = False
+
+        # --- 3. VAD Reset Logic ---
+        if silence_reset and not self.is_first_logical_chunk:
+            logging.info(f"\n[VAD RESET: SILENCE detected ({self.silent_chunks_count} empty chunks) at {(self.chunks_count * (self.read_size/self.VAD_SAMPLE_RATE)):.2f}s]")
+            
+            # Flush the buffer, reset state, and get final text
+            reset_text = self._flush_and_reset()
+            if reset_text:
+                logging.info(f"Received final reset text: '{reset_text}'")
+                new_text_segments.append(reset_text)
+
+        return new_text_segments
+
+    def finalize_stream(self):
+        """
+        Must be called at the very end of the stream (after the loop breaks).
+        Flushes anything remaining in the buffer.
+        """
+        logging.info("Finalizing stream. Flushing final buffer...")
+        final_text = self._flush_and_reset()
+        if final_text:
+            logging.info(f"Received final flushed text: '{final_text}'")
+        return final_text
+

app/utils.py

@@ -12,7 +12,8 @@ import base64
 import os
 import time
 import random
-
+import torch
+from app.streaming_audio_processor import StreamingAudioProcessor
 from app.session_utils import (
     get_active_task_flag_file,
     get_folder_chunks
@@ -57,7 +58,7 @@ def generate_coturn_config():
 
 
 
-def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streaming_flags: dict):
+def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streaming_flags: dict,read_size:int =8000, sample_rate:int =16000):
     """
     Read an audio file and stream it chunk by chunk (1s per chunk).
     Handles errors safely and reports structured messages to the client.
@@ -72,7 +73,7 @@ def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streami
     transcribe_flag = get_active_task_flag_file(session_id)
     try:
         segment = AudioSegment.from_file(filepath_to_stream)
-        chunk_duration_ms = 1000
+        chunk_duration_ms = int((read_size/sample_rate)*1000)
         total_chunks = len(segment) // chunk_duration_ms + 1
         logging.info(f"[{session_id}] Starting audio streaming {filepath_to_stream} ({total_chunks} chunks).")
 
@@ -88,9 +89,9 @@ def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streami
                 break
             
             yield ((frame_rate, samples), AdditionalOutputs({"progressed": True, "value": progress} ))
-            logging.debug(f"[{session_id}] Sent chunk {i+1}/{total_chunks} ({progress}%).")
+            # logging.debug(f"[{session_id}] Sent chunk {i+1}/{total_chunks} ({progress}%).")
 
-            time.sleep(1)
+            time.sleep(chunk_duration_ms/1000)
             #  Save only if transcription is active
             if os.path.exists(transcribe_flag) :
                 chunk_dir = get_folder_chunks(session_id)
@@ -99,7 +100,7 @@ def read_and_stream_audio(filepath_to_stream: str, session_id: str, stop_streami
                 npz_path = os.path.join(chunk_dir, f"chunk_{i:05d}.npz")
                 chunk_array = np.array(chunk.get_array_of_samples(), dtype=np.int16)
                 np.savez_compressed(npz_path, data=chunk_array, rate=frame_rate)
-                logging.debug(f"[{session_id}] Saved chunk {i}/{total_chunks} (transcribe active)")
+                logging.debug(f"[{session_id}] Saved chunk {i}/{total_chunks} (transcribe active) ({progress}%) ({npz_path}).")
 
             # raise_function()  # Optional injected test exception
 
@@ -148,14 +149,15 @@ def _is_stop_requested(stop_streaming_flags: dict) -> bool:
 
 # --- Decorator compatibility layer ---
 if os.environ.get("SPACE_ID", "").startswith("zero-gpu"):
-    logging.warning("Running on ZeroGPU — disabling @spaces.GPU")
-    def gpu_decorator(f): return f
+    logging.warning("Running on ZeroGPU — gpu_fork_decorator @spaces.GPU")
+    def gpu_fork_decorator(f): return f
 else:
     gpu_decorator = spaces.GPU
 
+
 # --- Audio Stream Function ---
-@gpu_decorator
-def task(session_id: str):
+@spaces.GPU
+def task(session_id: str, streamer:  StreamingAudioProcessor):
     """Continuously read and delete .npz chunks while task is active."""
     active_flag = get_active_task_flag_file(session_id)
     with open(active_flag, "w") as f:
@@ -171,11 +173,11 @@ def task(session_id: str):
             if not os.path.exists(chunk_dir):
                 logging.warning(f"[{session_id}] No chunk directory found for task.")
                 yield "No audio chunks yet... waiting for stream.\n"
-                time.sleep(0.25)
+                time.sleep(0.1)
                 continue
             files = sorted(f for f in os.listdir(chunk_dir) if f.endswith(".npz"))
             if not files:
-                time.sleep(0.25)
+                time.sleep(0.1)
                 continue
 
             for fname in files:
@@ -186,19 +188,26 @@ def task(session_id: str):
                     rate = int(npz["rate"])
 
                     text = f"Transcribed {fname}: {len(samples)} samples @ {rate}Hz"
-                    yield f"{text}\n"
-                    logging.debug(f"[{session_id}] {text}")
-
+                    new_texts = streamer.process_chunk(samples)
+                    for text in new_texts:
+                        print(text, end='', flush=True)
+                        yield f"{text}"
+                        logging.debug(f"[{session_id}] {new_texts}")
+                    # yield f"{text}\n"   
                     os.remove(fpath)
                     logging.debug(f"[{session_id}] Deleted processed chunk: {fname}")
                 except Exception as e:
                     logging.error(f"[{session_id}] Error processing {fname}: {e}")
                     yield f"Error processing {fname}: {e}\n"
                     continue
-
-            time.sleep(0.25)
+            
+            time.sleep(0.1)
             # raise_function()
-        yield "\nTask stopped by user or stream ended.\n"
+        final_text = streamer.finalize_stream()
+        if final_text:
+            print(final_text, end='', flush=True)
+        yield f"\n{final_text}"
+        # yield f"\n"
         logging.info(f"[{session_id}] task loop ended (flag removed).")
 
     except Exception as e:

requirements.txt

@@ -3,3 +3,6 @@ spaces
 torch
 python-dotenv
 fastrtc==0.0.33
+Cython
+nemo_toolkit[asr,nlp] @ git+https://github.com/NVIDIA/NeMo.git@237e2c08ed8e5b6bad66b124d75b02f6510b9b56 
+onnxruntime