update code structures and some code

config/settings.py

@@ -60,8 +60,8 @@ WHISPER_MODEL_EN = 'large-v3-turbo-q5_0'
 WHISPER_MODEL_ZH = 'large-v3-turbo-q5_0'
 # LLM
 LLM_MODEL_PATH = (MODEL_DIR / "qwen2.5-1.5b-instruct-q5_0.gguf").as_posix()
-LLM_LARGE_MODEL_PATH = (MODEL_DIR / "qwen2.5-1.5b-instruct-q5_0.gguf").as_posix()
-# LLM_LARGE_MODEL_PATH = (MODEL_DIR / "qwen2.5-7b-instruct-q5_0-00001-of-00002.gguf").as_posix()
+# LLM_LARGE_MODEL_PATH = (MODEL_DIR / "qwen2.5-1.5b-instruct-q5_0.gguf").as_posix()
+LLM_LARGE_MODEL_PATH = (MODEL_DIR / "qwen2.5-7b-instruct-q5_0-00001-of-00002.gguf").as_posix()
 
 # VAD
 VAD_MODEL_PATH = (MODEL_DIR / "silero-vad" / "silero_vad.onnx").as_posix()

main.py

@@ -1,6 +1,5 @@
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect
-from urllib.parse import urlparse, parse_qsl
-from transcribe.whisper_llm_serve import WhisperTranscriptionService
+from transcribe.serve import WhisperTranscriptionService
 from uuid import uuid1
 from logging import getLogger
 import numpy as np

transcribe/pipelines/pipe_vad.py

@@ -72,7 +72,7 @@ class VadPipe(BasePipe):
                 logging.debug(" 🫷Speech ended, capturing audio up to frame: {}".format(rel_end_frame))
             else:
                 self._status = 'END'
-                target_audio = source_audio[rel_start_frame:rel_end_frame]
+                target_audio = source_audio[max(rel_start_frame-100, 0):rel_end_frame]
                 logging.debug(" 🔄 Speech segment captured from frame {} to frame {}".format(rel_start_frame, rel_end_frame))
                 # logging.debug("❌ No valid speech segment detected, setting status to END")
         else:

transcribe/{whisper_llm_serve.py → serve.py}

@@ -7,9 +7,9 @@ import asyncio
 import numpy as np
 import config
 import collections
-from api_model import TransResult, Message, DebugResult
+from api_model import TransResult, Message
 
-from .utils import log_block, save_to_wave, TestDataWriter, filter_words
+from .utils import log_block, start_thread, get_text_separator, filter_words
 from .translatepipes import TranslatePipes
 
 from transcribe.pipelines import MetaItem
@@ -18,26 +18,12 @@ from transcribe.pipelines import MetaItem
 logger = getLogger("TranscriptionService")
 
 
-def _get_text_separator(language: str) -> str:
-    """根据语言返回适当的文本分隔符"""
-    return "" if language == "zh" else " "
-
-
-def _start_thread(target_function) -> threading.Thread:
-    """启动守护线程执行指定函数"""
-    thread = threading.Thread(target=target_function)
-    thread.daemon = True
-    thread.start()
-    return thread
-
 
 class WhisperTranscriptionService:
     """
     Whisper语音转录服务类，处理音频流转录和翻译
     """
 
-    SERVER_READY = "SERVER_READY"
-    DISCONNECT = "DISCONNECT"
 
     def __init__(self, websocket, pipe: TranslatePipes, language=None, dst_lang=None, client_uid=None):
         print('>>>>>>>>>>>>>>>> init service >>>>>>>>>>>>>>>>>>>>>>')
@@ -47,52 +33,37 @@ class WhisperTranscriptionService:
         self.client_uid = client_uid
         # 转录结果稳定性管理
         self.websocket = websocket
-        self._translate_pipe = pipe
-
+        self.translate_pipe = pipe
         # 音频处理相关
         self.sample_rate = config.SAMPLE_RATE
 
         self.lock = threading.Lock()
         # 文本分隔符，根据语言设置
-        self.text_separator = _get_text_separator(language)
+        self.text_separator = get_text_separator(language)
         self.loop = asyncio.get_event_loop()
         #  原始音频队列
-        self._frame_queue = queue.Queue()
+        self.frame_queue = queue.Queue()
         #  音频队列缓冲区
         self.frames_np = np.array([], dtype=np.float32)
+        #  音频开始的时间点 用于约束最小断句时间
         self.frames_np_start_timestamp = None
         #  完整音频队列
         self.full_segments_queue = collections.deque()
         # 启动处理线程
-        self._translate_thread_stop = threading.Event()
-        self._frame_processing_thread_stop = threading.Event()
-
-        self.translate_thread = _start_thread(self._transcription_processing_loop)
-        self.frame_processing_thread = _start_thread(self._read_frame_processing_loop)
+        self._stop = threading.Event()
+        self.translate_thread = start_thread(self._transcription_processing_loop)
+        self.frame_processing_thread = start_thread(self._read_frame_processing_loop)
+        # 行号
         self.row_number = 0
-        # for test
-        self._transcribe_time_cost = 0.
-        self._translate_time_cost = 0.
-
-        if config.SAVE_DATA_SAVE:
-            self._save_task_stop = threading.Event()
-            self._save_queue = queue.Queue()
-            self._save_thread = _start_thread(self.save_data_loop)
-
 
-    def save_data_loop(self):
-        writer = TestDataWriter()
-        while not self._save_task_stop.is_set():
-            test_data = self._save_queue.get()
-            writer.write(test_data)  # Save test_data to CSV
 
     def add_frames(self, frame_np: np.ndarray) -> None:
         """添加音频帧到处理队列"""
-        self._frame_queue.put(frame_np)
+        self.frame_queue.put(frame_np)
 
     def _apply_voice_activity_detection(self, frame_np:np.array):
         """应用语音活动检测来优化音频缓冲区"""
-        processed_audio = self._translate_pipe.voice_detect(frame_np.tobytes())
+        processed_audio = self.translate_pipe.voice_detect(frame_np.tobytes())
         speech_audio =  np.frombuffer(processed_audio.audio, dtype=np.float32)
         speech_status = processed_audio.speech_status
         return speech_audio, speech_status
@@ -100,9 +71,9 @@ class WhisperTranscriptionService:
 
     def _read_frame_processing_loop(self) -> None:
         """从队列获取音频帧并合并到缓冲区"""
-        while not self._frame_processing_thread_stop.is_set():
+        while not self._stop.is_set():
             try:
-                frame_np = self._frame_queue.get(timeout=0.1)
+                frame_np = self.frame_queue.get(timeout=0.1)
                 frame_np, speech_status = self._apply_voice_activity_detection(frame_np)
 
                 if frame_np is None:
@@ -142,7 +113,7 @@ class WhisperTranscriptionService:
         """主转录处理循环"""
         frame_epoch = 1
 
-        while not self._translate_thread_stop.is_set():
+        while not self._stop.is_set():
       
             if len(self.frames_np) ==0:
                 time.sleep(0.1)
@@ -187,54 +158,35 @@ class WhisperTranscriptionService:
                     partial=partial
                 )
                 self._send_result_to_client(result)
-    
-
 
     def _transcribe_audio(self, audio_buffer: np.ndarray)->MetaItem:
         """转录音频并返回转录片段"""
         log_block("Audio buffer length", f"{audio_buffer.shape[0]/self.sample_rate:.2f}", "s")
-        start_time = time.perf_counter()
-
-        result = self._translate_pipe.transcribe(audio_buffer.tobytes(), self.source_language)
-        segments = result.segments
-        time_diff = (time.perf_counter() - start_time)
-        logger.debug(f"📝 transcribe Segments: {segments} ")
-        log_block("📝 transcribe output", f"{self.text_separator.join(seg.text for seg in segments)}", "")
-        log_block("📝 transcribe time", f"{time_diff:.3f}", "s")
-        self._transcribe_time_cost = round(time_diff, 3)
+        result = self.translate_pipe.transcribe(audio_buffer.tobytes(), self.source_language)
+        log_block("📝 transcribe output", f"{self.text_separator.join(seg.text for seg in result.segments)}", "")
+
         return result
 
     def _translate_text(self, text: str) -> str:
         """将文本翻译为目标语言"""
         if not text.strip():
             return ""
-
         log_block("🐧 Translation input ", f"{text}")
-        start_time = time.perf_counter()
 
-        result = self._translate_pipe.translate(text, self.source_language, self.target_language)
+        result = self.translate_pipe.translate(text, self.source_language, self.target_language)
         translated_text = result.translate_content
-        time_diff = (time.perf_counter() - start_time)
-        log_block("🐧 Translation time ", f"{time_diff:.3f}", "s")
+ 
         log_block("🐧 Translation out ", f"{translated_text}")
-        self._translate_time_cost = round(time_diff, 3)
         return translated_text
 
     def _translate_text_large(self, text: str) -> str:
         """将文本翻译为目标语言"""
         if not text.strip():
             return ""
-
         log_block("Translation input", f"{text}")
-        start_time = time.perf_counter()
-
-        result = self._translate_pipe.translate_large(text, self.source_language, self.target_language)
+        result = self.translate_pipe.translate_large(text, self.source_language, self.target_language)
         translated_text = result.translate_content
-
-        time_diff = (time.perf_counter() - start_time)
-        log_block("Translation large model time ", f"{time_diff:.3f}", "s")
         log_block("Translation large model output", f"{translated_text}")
-        self._translate_time_cost = round(time_diff, 3)
         return translated_text
 
 
@@ -252,8 +204,5 @@ class WhisperTranscriptionService:
 
     def stop(self) -> None:
         """停止所有处理线程并清理资源"""
-        self._translate_thread_stop.set()
-        self._frame_processing_thread_stop.set()
-        if config.SAVE_DATA_SAVE:
-            self._save_task_stop.set()
+        self._stop.set()
         logger.info(f"Stopping transcription service for client: {self.client_uid}")

transcribe/translatepipes.py

@@ -1,5 +1,5 @@
-from transcribe.pipelines import WhisperPipe, MetaItem, WhisperChinese, Translate7BPipe, FunASRPipe, VadPipe, TranslatePipe
-
+from .pipelines import WhisperPipe, MetaItem, WhisperChinese, Translate7BPipe, FunASRPipe, VadPipe, TranslatePipe
+from .utils import timer
 
 class TranslatePipes:
     def __init__(self) -> None:
@@ -26,6 +26,7 @@ class TranslatePipes:
         for p in self._process:
             p.wait()
 
+    @timer(name="🐧 Translate")
     def translate(self, text, src_lang, dst_lang) -> MetaItem:
         item = MetaItem(
             transcribe_content=text,
@@ -34,6 +35,7 @@ class TranslatePipes:
         self._translate_pipe.input_queue.put(item)
         return self._translate_pipe.output_queue.get()
 
+    @timer(name="🐧 Translate-large")
     def translate_large(self, text, src_lang, dst_lang) -> MetaItem:
         item = MetaItem(
             transcribe_content=text,
@@ -47,6 +49,7 @@ class TranslatePipes:
             return self._funasr_pipe
         return self._whisper_pipe_en
 
+    @timer(name="📝 transcribe")
     def transcribe(self, audio_buffer: bytes, src_lang: str) -> MetaItem:
         transcription_model = self.get_transcription_model(src_lang)
         item = MetaItem(audio=audio_buffer, source_language=src_lang)

transcribe/utils.py

@@ -8,8 +8,9 @@ import config
 import csv
 import av
 import re
-import json
-
+from functools import wraps
+import time
+import threading
 # Compile regex patterns once outside the loop for better performance
 p_pattern = re.compile(r"(\s*\[.*?\])")
 p_start_pattern = re.compile(r"(\s*\[.*)")
@@ -178,6 +179,32 @@ def pcm_bytes_to_np_array(pcm_bytes: bytes, dtype=np.float32, channels=1):
         audio_np = audio_np.reshape(-1, channels)
     return audio_np
 
+def timer(name: str):
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            start_time = time.perf_counter()
+            result = func(*args, **kwargs)
+            end_time = time.perf_counter()
+            duration = end_time - start_time
+            log_block(f"{name} cost:", f"{duration:.2f} s")
+            return result
+        return wrapper
+    return decorator
+
+def get_text_separator(language: str) -> str:
+    """根据语言返回适当的文本分隔符"""
+    return "" if language == "zh" else " "
+
+
+def start_thread(target_function) -> threading.Thread:
+    """启动守护线程执行指定函数"""
+    thread = threading.Thread(target=target_function)
+    thread.daemon = True
+    thread.start()
+    return thread
+
+
 class TestDataWriter:
     def __init__(self, file_path='test_data.csv'):
         self.file_path = file_path