fix bug of queue lock

transcribe/serve.py

@@ -37,7 +37,8 @@ class WhisperTranscriptionService:
         # 音频处理相关
         self.sample_rate = config.SAMPLE_RATE
 
-        self.lock = threading.Lock()
+        self.frame_lock = threading.Lock()
+        self.segment_lock = threading.Lock()
         # 文本分隔符，根据语言设置
         self.text_separator = get_text_separator(language)
         self.loop = asyncio.get_event_loop()
@@ -72,61 +73,70 @@ class WhisperTranscriptionService:
     def _read_frame_processing_loop(self) -> None:
         """从队列获取音频帧并合并到缓冲区"""
         while not self._stop.is_set():
-            try:
-                frame_np = self.frame_queue.get(timeout=0.1)
-                frame_np, speech_status = self._apply_voice_activity_detection(frame_np)
+            frame_np = self.frame_queue.get()
+            frame_np, speech_status = self._apply_voice_activity_detection(frame_np)
 
-                if frame_np is None:
-                    continue
-                # logger.critical(f"frame np:{frame_np.shape}, {speech_status}")
-                with self.lock:
-                    self.frames_np = np.append(self.frames_np, frame_np)
-
-                     # 音频开始时间节点 用来统计时间来 达到最小断句时间长度
-                    if speech_status == "START" and self.frames_np_start_timestamp is None:
-                        self.frames_np_start_timestamp = time.time()
+            if frame_np is None:
+                continue
 
-                    # 音频最长时间缓冲区限制，超过了就强制断句
-                    if len(self.frames_np) >= self.sample_rate * config.MAX_SPEECH_DURATION_S:
+            with self.frame_lock:
+                self.frames_np = np.append(self.frames_np, frame_np)
+
+                # 音频开始时间节点 用来统计时间来 达到最小断句时间长度
+            if speech_status == "START" and self.frames_np_start_timestamp is None:
+                self.frames_np_start_timestamp = time.time()
+
+            # 音频最长时间缓冲区限制，超过了就强制断句
+            if len(self.frames_np) >= self.sample_rate * config.MAX_SPEECH_DURATION_S:
+                audio_array=self.frames_np.copy()
+                with self.segment_lock:
+                    self.full_segments_queue.appendleft(audio_array) # 根据时间是否满足三秒长度 来整合音频块
+                self.frames_np_start_timestamp = time.time()
+
+                with self.frame_lock:
+                    self.frames_np = np.array([], dtype=np.float32)
+
+            # 音频结束信号的时候 整合当前缓冲区
+            #   START -- END -- START -- END 通常
+            #   START -- END -- END  end块带有音频信息的通常是4096内断的一个短音
+            if speech_status == "END" and len(self.frames_np) > 0 and self.frames_np_start_timestamp:
+                time_diff = time.time() - self.frames_np_start_timestamp
+                if time_diff >= config.FRAME_SCOPE_TIME_THRESHOLD:
+                    with self.frame_lock:
                         audio_array=self.frames_np.copy()
-                        self.full_segments_queue.appendleft(audio_array) # 根据时间是否满足三秒长度 来整合音频块
-                        self.frames_np_start_timestamp = time.time()
                         self.frames_np = np.array([], dtype=np.float32)
 
-                    # 音频结束信号的时候 整合当前缓冲区
-                    #   START -- END -- START -- END 通常
-                    #   START -- END -- END  end块带有音频信息的通常是4096内断的一个短音
-                    elif speech_status == "END" and len(self.frames_np) > 0 and self.frames_np_start_timestamp:
-                        time_diff = time.time() - self.frames_np_start_timestamp
-                        if time_diff >= config.FRAME_SCOPE_TIME_THRESHOLD:
-                            audio_array=self.frames_np.copy()
-                            self.full_segments_queue.appendleft(audio_array) # 根据时间是否满足三秒长度 来整合音频块
-                            self.frames_np_start_timestamp = None
-                            self.frames_np = np.array([], dtype=np.float32)
-                        else:
-                            logger.debug(f"🥳 当前时间与上一句的时间差: {time_diff:.2f}s,继续保留在缓冲区")
-
-            except queue.Empty:
-                pass
+                    with self.segment_lock:
+                        self.full_segments_queue.appendleft(audio_array) # 根据时间是否满足三秒长度 来整合音频块
+                    logger.debug(f"🥳 增加整句到队列")
+                    self.frames_np_start_timestamp = None
+                   
+                else:
+                    logger.debug(f"🥳 当前时间与上一句的时间差: {time_diff:.2f}s,继续保留在缓冲区")
+
 
     def _transcription_processing_loop(self) -> None:
         """主转录处理循环"""
         frame_epoch = 1
 
         while not self._stop.is_set():
-
-            if len(self.frames_np) ==0:
-                time.sleep(0.1)
-                continue
-
-            with self.lock:
-                if len(self.full_segments_queue) > 0:
-                    audio_buffer = self.full_segments_queue.pop()
-                    partial = False
-                else:
+            time.sleep(0.1)
+          
+            with self.segment_lock:
+                segment_length = len(self.full_segments_queue) 
+            if segment_length > 0:
+                audio_buffer = self.full_segments_queue.pop()
+                partial = False
+            else:
+                with self.frame_lock:
+                    if len(self.frames_np) ==0:
+                        continue
                     audio_buffer = self.frames_np[:int(frame_epoch * 1.5 * self.sample_rate)].copy()# 获取 1.5s * epoch 个音频长度
                     partial = True
 
+            logger.debug(f"full_segments_queue size: {segment_length}")
+            logger.debug(f"audio buffer size: {len(self.frames_np) / self.sample_rate:.2f}s")
+            
             if len(audio_buffer) < int(self.sample_rate):
                 # Add a small buffer (e.g., 10ms worth of samples) to be safe
                 padding_samples = int(self.sample_rate * 0.01) # e.g., 160 samples for 10ms at 16kHz
@@ -137,7 +147,7 @@ class WhisperTranscriptionService:
                 silence_audio[-copy_length:] = audio_buffer[-copy_length:] # Copy from the end of audio_buffer
                 audio_buffer = silence_audio
 
-            logger.debug(f"audio buffer size: {len(audio_buffer) / self.sample_rate:.2f}s")
+            
             meta_item = self._transcribe_audio(audio_buffer)
             segments = meta_item.segments
             logger.debug(f"Segments: {segments}")