sdfsdf

sdsadas

sfdsadfd

import asyncio, json, numpy as np, websockets, webrtcvad, time from faster_whisper import WhisperModel MODEL_SIZE = "small" # баланс: tiny/base/small/medium/large-v3 DEVICE = "cuda" # "cpu" или "cuda" COMPUTE_TYPE = "auto" # для GPU: "float16" / "int8_float16" CHUNK_MS = 100 # длина входного фрейма (как на клиенте) WINDOW_SEC = 8.0 # длина скользящего окна для расшифровки STEP_SEC = 2.0 # как часто запускать инференс SAMPLE_RATE = 16000 model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE) vad = webrtcvad.Vad(2) # 0..3, выше — строже к шуму async def transcribe_session(ws): buf = np.zeros(0, dtype=np.int16) last_infer = 0.0 started = False async for msg in ws: # первые сообщения могут быть JSON (метаданные) if isinstance(msg, (str, bytes)) and isinstance(msg, str): try: meta = json.loads(msg) if meta.get("op") == "start": started = True await ws.send(json.dumps({"type":"ready"})) continue except Exception: pass # далее ожидаем бинарные PCM16LE чанки if isinstance(msg, (bytes, bytearray)): pcm = np.frombuffer(msg, dtype=np.int16) # простая VAD-обрезка тишины по краям чанка (опционально) frame_bytes = pcm.tobytes() is_speech = vad.is_speech(frame_bytes[:320*2], SAMPLE_RATE) if len(pcm) >= 320 else True if is_speech: buf = np.concatenate([buf, pcm]) now = time.time() if now - last_infer >= STEP_SEC and buf.size > 0: last_infer = now # ограничим окно max_samples = int(WINDOW_SEC * SAMPLE_RATE) if buf.size > max_samples: buf = buf[-max_samples:] # faster-whisper не «истинно» стримит, но быстро обрабатывает окно segments, info = model.transcribe( buf.astype(np.float32) / 32768.0, language="ru", vad_filter=True, beam_size=1, best_of=1, no_speech_threshold=0.6, condition_on_previous_text=False, word_timestamps=True ) # берём последнюю гипотезу как partial partial_text = "" ts, te = None, None for seg in segments: partial_text += seg.text ts, te = seg.start, seg.end await ws.send(json.dumps({ "type": "partial", "text": partial_text.strip(), "ts": ts, "te": te })) # на закрытие допрожимаем «финал» if buf.size > 0: segments, info = model.transcribe( buf.astype(np.float32) / 32768.0, language="ru", vad_filter=True, beam_size=1, best_of=1, condition_on_previous_text=False, word_timestamps=True ) final_text = "".join(seg.text for seg in segments).strip() if final_text: await ws.send(json.dumps({"type":"final","text":final_text})) async def main(): async with websockets.serve(transcribe_session, host="0.0.0.0", port=2700, max_size=None): print("WS STT server on ws://0.0.0.0:2700/stt") await asyncio.Future() if __name__ == "__main__": try: import uvloop; uvloop.install() except Exception: pass asyncio.run(main())