add vad pipeline

transcribe/helpers/translator.py

@@ -1,6 +1,6 @@
 from logging import getLogger
 from llama_cpp import Llama
-import time
+from functools import lru_cache
 
 logger = getLogger(__name__)
 
@@ -19,7 +19,7 @@ class QwenTranslator:
             {"role": "user", "content": prompt},
         ]
 
-
+    @lru_cache(maxsize=10)
     def translate(self, prompt, src_lang, dst_lang) -> str:
         message = self.to_message(prompt, src_lang, dst_lang)
         output = self.llm.create_chat_completion(messages=message, temperature=0)

transcribe/helpers/vadprocessor.py

@@ -4,10 +4,35 @@ import numpy as np
 import onnxruntime
 from datetime import timedelta
 from pydub import AudioSegment
-from silero_vad import load_silero_vad, get_speech_timestamps
+from silero_vad import load_silero_vad, get_speech_timestamps, VADIterator
 import os
 import logging
 
+class FixedVADIterator(VADIterator):
+    '''It fixes VADIterator by allowing to process any audio length, not only exactly 512 frames at once.
+    If audio to be processed at once is long and multiple voiced segments detected, 
+    then __call__ returns the start of the first segment, and end (or middle, which means no end) of the last segment. 
+    '''
+
+    def reset_states(self):
+        super().reset_states()
+        self.buffer = np.array([],dtype=np.float32)
+
+    def __call__(self, x, return_seconds=False):
+        self.buffer = np.append(self.buffer, x) 
+        ret = None
+        while len(self.buffer) >= 512:
+            r = super().__call__(self.buffer[:512], return_seconds=return_seconds)
+            self.buffer = self.buffer[512:]
+            if ret is None:
+                ret = r
+            elif r is not None:
+                if 'end' in r:
+                    ret['end'] = r['end']  # the latter end
+                if 'start' in r and 'end' in ret:  # there is an earlier start.
+                    # Remove end, merging this segment with the previous one.
+                    del ret['end']
+        return ret if ret != {} else None
 
 class SileroVADProcessor:
     """

transcribe/pipelines/base.py

@@ -2,6 +2,10 @@
 from dataclasses import dataclass, field
 from multiprocessing import Process, Queue
 from multiprocessing import Event
+from logging import getLogger
+
+logger = getLogger(__name__)
+
 
 @dataclass
 class Segment:
@@ -53,7 +57,9 @@ class BasePipe(Process):
         raise NotImplementedError
 
     def run(self):
+        logger.info(f"start initial {self.__class__.__name__}")
         self.init()
+        logger.info(f"finish initial {self.__class__.__name__}")
         self.set_ready()
         while True:
             item = self.input_queue.get()

transcribe/pipelines/pipe_vad.py

@@ -1,15 +1,18 @@
 
 from .base import MetaItem, BasePipe
-from ..helpers.vadprocessor import SileroVADProcessor
+from ..helpers.vadprocessor import SileroVADProcessor, FixedVADIterator
 import numpy as np
-from silero_vad import read_audio, get_speech_timestamps,collect_chunks, VADIterator
+from silero_vad import get_speech_timestamps,collect_chunks
 import torch
+import noisereduce as nr
+
 
 class VadPipe(BasePipe):
     model = None
-    sample_rate=16000
+    sample_rate = 16000
     window_size_samples = 512
 
+
     @classmethod
     def init(cls):
         if cls.model is None:
@@ -21,50 +24,28 @@ class VadPipe(BasePipe):
             min_silence_duration=250,
             sample_rate=cls.sample_rate
         )
+            cls.vac = FixedVADIterator(cls.model.silero_vad, sampling_rate=cls.sample_rate,)
+            cls.vac.reset_states()
+   
+    
+    def get_previous_buffer(self):
+        if len(self.previous_buffer) == 2:
+            return self.previous_buffer[-1]
+        return np.array([], dtype=np.float32)
+    
 
-    @property
-    def vad_iterator(self):
-        return VADIterator(self.model.silero_vad, sampling_rate=self.sample_rate,)
+    def reduce_noise(self, data):
+        return nr.reduce_noise(y=data, sr=self.sample_rate)
+    
 
     def process(self, in_data: MetaItem) -> MetaItem:
         source_audio = in_data.source_audio
         source_audio = np.frombuffer(source_audio, dtype=np.float32)
-        speech_segments = []
-        is_speech_active = False
-        # current_segment_end = len(source_audio)
-       
-        for i in range(0, len(source_audio), self.window_size_samples):
-
-            window = source_audio[i:i+self.window_size_samples]
-         
-            if len(window) < self.window_size_samples:
-                padded_window = np.zeros(self.window_size_samples, dtype=np.float32)
-                padded_window[:len(window)] = window
-                window = padded_window  
-
-            speech_dict = self.vad_iterator(window, return_seconds=False)
-            if not speech_dict:
-                continue
-   
-            # 计算当前偏移量
-
-            if speech_dict and 'start' in speech_dict and not is_speech_active:
-                is_speech_active = True
-                # current_segment_start = speech_dict['start'] + i
-            
-            if is_speech_active:
-                speech_segments.append(window)
-            # # 如果检测到语音结束
-            # if speech_dict and 'end' in speech_dict and is_speech_active:
-            #     # 调整语音结束时间，加上窗口偏移
-            #     current_segment_end = min(speech_dict['end'] + i, current_segment_end)
-            #     is_speech_active = False
-            #     speech_audio = source_audio[current_segment_start: current_segment_end]
-            #     speech_segments.append(speech_audio)
-
-        self.vad_iterator.reset_states()
-        combied_audio = np.concatenate(speech_segments, axis=0).tobytes() if len(speech_segments) else b""
-        in_data.audio = combied_audio
+        send_audio = b""
+        speech_timestamps = get_speech_timestamps(source_audio, self.model.silero_vad, sampling_rate=16000)
+        if speech_timestamps:
+            send_audio = collect_chunks(speech_timestamps, torch.Tensor(source_audio))
+            send_audio = send_audio.numpy()
+            # send_audio = self.reduce_noise(send_audio).tobytes()
         in_data.source_audio = b""
-        
         return in_data

transcribe/pipelines/pipe_whisper.py

@@ -7,9 +7,6 @@ from ..helpers.whisper import WhisperCPP
 class WhisperPipe(BasePipe):
     whisper = None
 
-    def __init__(self, in_queue=None, out_queue=None) -> None:
-        super().__init__(in_queue, out_queue)
-
 
     @classmethod
     def init(cls):

transcribe/strategy.py

@@ -98,7 +98,7 @@ def segement_merge(segments):
 
     for seg in segments:
         temp_seq.append(seg)
-        if any([mk in seg.text for mk in config.SENTENCE_END_MARKERS]):
+        if any([mk in seg.text for mk in config.SENTENCE_END_MARKERS + config.PAUSE_END_MARKERS]):
             sequences.append(temp_seq.copy())
             temp_seq = []
     if temp_seq:
@@ -114,18 +114,18 @@ def segments_split(segments, audio_buffer: np.ndarray, sample_rate=16000):
 
     if (len(audio_buffer) / sample_rate) < 12:
         # 低于12s 使用短句符号比如逗号作为判断依据
-        markers = config.PAUSE_END_MARKERS
+        markers = config.PAUSE_END_MARKERS + config.SENTENCE_END_MARKERS
         is_end = False
 
         for idx, seg in enumerate(segments):
             left_watch_sequences.append(seg)
             if seg.text in markers:
                 seg_index = int(seg.t1 / 100 * sample_rate)
-                rest_buffer_duration = (len(audio_buffer) - seg_index) / sample_rate
+                # rest_buffer_duration = (len(audio_buffer) - seg_index) / sample_rate
                 # is_end = any(i in seg.text for i  in config.SENTENCE_END_MARKERS)
                 right_watch_sequences = segments[min(idx+1, len(segments)):]
-                if rest_buffer_duration >= 1.5:
-                    left_watch_idx = seg_index
+                # if rest_buffer_duration >= 1.5:
+                left_watch_idx = seg_index
                 break
     return left_watch_idx, left_watch_sequences, right_watch_sequences, is_end
 

transcribe/whisper_llm_serve.py

@@ -40,7 +40,6 @@ class PyWhiperCppServe(ServeClientBase):
         self.lock = threading.Lock()
         self.frames_np = None
         self._frame_queue = queue.Queue()
-        self._previous_frame_queue = collections.deque(maxlen=2)
         self.sample_rate = 16000
  
         self.send_ready_state()
@@ -69,24 +68,19 @@ class PyWhiperCppServe(ServeClientBase):
     def add_frames(self, frame_np):
         self._frame_queue.put(frame_np)
 
-    def get_prev_frame(self, ):
-        if len(self._previous_frame_queue) == 2:
-            return self._previous_frame_queue[-1]
-
+    def vad_merge(self):
+        with self.lock:
+            frame = self.frames_np.copy()
+            item = translate_pipes.voice_detect(frame.tobytes())
+            if item.audio != b'':
+                frame_np = np.frombuffer(item.audio, dtype=np.float32)
+                self.frames_np = frame_np.copy()
+     
     
     def get_frame_from_queue(self,):
         while True:
             try:
                 frame_np = self._frame_queue.get(timeout=0.1)
-                # frame_np = item.source_audio
-                # self._previous_frame_queue.appendleft(frame_np.copy())
-                # prev_frame_np = self.get_prev_frame()
-                # if prev_frame_np is not None:
-                #     frame_np = np.concatenate([prev_frame_np[int(-0.05 * self.sample_rate):],frame_np], axis=0)
-                # item = translate_pipes.voice_detect(frame_np.tobytes())
-                # if item.audio == b"":
-                #     continue 
-                # frame_np = np.frombuffer(item.audio, dtype=np.float32)
                 with self.lock:
                     if self.frames_np is None:
                         self.frames_np = frame_np.copy()
@@ -96,7 +90,6 @@ class PyWhiperCppServe(ServeClientBase):
                 pass
           
 
-
     def update_audio_buffer(self, last_offset):
         with self.lock:
             self.frames_np = self.frames_np[last_offset:]
@@ -244,6 +237,7 @@ class PyWhiperCppServe(ServeClientBase):
     def get_audio_chunk_for_processing(self):
         if self.frames_np.shape[0] >= self.sample_rate * 1:
             return self.frames_np.copy()
+        self.vad_merge()
         # 计算需要填充的样本数
         padding_length = self.sample_rate * 1 - len(self.frames_np)
         # 创建静音填充（零值）