import os import json import threading import datetime import time import numpy as np from vosk import Model, KaldiRecognizer import sounddevice as sd import uuid from collections import deque VOICE_COMMANDS_RESPONSE_TO_STORE = 3 class VoiceService: def __init__(self, model_path, samplerate=48000): self.model_path = model_path self.samplerate = samplerate self.model = Model(model_path) self.is_running = False self.current_file = None self.history = [] self.history_lock = threading.Lock() self.last_processed_timestamp = 0.0 self.responses = { "regular": deque(maxlen=VOICE_COMMANDS_RESPONSE_TO_STORE), "commands": deque(maxlen=VOICE_COMMANDS_RESPONSE_TO_STORE) } self.lock = threading.Lock() self._stop_event = threading.Event() def _load_history_from_file(self): """Парсит существующий файл лога в коллекцию self.history.""" if not self.current_file or not os.path.exists(self.current_file): return import re # Регулярка для парсинга строки типа: [10:20:30.123] GM: текст реплики pattern = re.compile(r"\[(\d{2}:\d{2}:\d{2}(?:\.\d+)?)\]\s+(GM|PCs):\s+(.*)") new_history = [] try: with open(self.current_file, "r", encoding="utf-8") as f: for line in f: match = pattern.match(line.strip()) if match: timestamp, role, text = match.groups() new_history.append({ 'role': role, 'text': text, 'time': timestamp }) with self.history_lock: self.history = new_history print(f"Loaded {len(self.history)} events from existing log.") except Exception as e: print(f"Error loading history from file: {e}") def start_session(self, continue_last=False): if self.is_running: return log_dir = "audio-logs" if not os.path.exists(log_dir): os.makedirs(log_dir) if continue_last: files = sorted([f for f in os.listdir(log_dir) if f.endswith(".txt")]) if files: self.current_file = os.path.join(log_dir, files[-1]) # Загружаем старые реплики в память self._load_history_from_file() else: self.current_file = self._create_new_file(log_dir) else: self.current_file = self._create_new_file(log_dir) with self.history_lock: self.history = [] # Сбрасываем историю для новой сессии self.is_running = True self._stop_event.clear() threading.Thread(target=self._run_transcription, daemon=True).start() def _create_new_file(self, log_dir): name = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S.txt") path = os.path.join(log_dir, name) with open(path, "w", encoding="utf-8") as f: f.write(f"--- Session Start: {name} ---\n\n") return path def stop_session(self): self.is_running = False self._stop_event.set() def _run_transcription(self): # 1. Поиск устройств (строго перед запуском потока) id_gm, id_pcs = None, None devices = sd.query_devices() hostapis = sd.query_hostapis() mme_api_index = next((i for i, h in enumerate(hostapis) if h['name'] == 'MME'), 0) for i, d in enumerate(devices): # Проверяем, что устройство относится к MME и содержит нужное имя if d['hostapi'] == mme_api_index: if "Voicemeeter Out B1" in d['name']: id_gm = i if "Voicemeeter Out B2" in d['name']: id_pcs = i if id_gm is None or id_pcs is None: # Если MME не нашел, пробуем любой for i, d in enumerate(devices): if "Voicemeeter Out B1" in d['name']: id_gm = i if "Voicemeeter Out B2" in d['name']: id_pcs = i # 2. Создаем рекогнайзеры ВНУТРИ потока rec_gm = KaldiRecognizer(self.model, self.samplerate) rec_pcs = KaldiRecognizer(self.model, self.samplerate) # 3. Фабрика колбэков def create_callback(rec, role): def callback_func(indata, frames, time_info, status): if status: print(f"Audio Status ({role}): {status}") if np.max(np.abs(indata)) < 0.001: return if rec.AcceptWaveform(bytes(indata)): res = json.loads(rec.Result()) text = res.get('text', '') if text: now = datetime.datetime.now().strftime("%H:%M:%S") # 1. Запись в файл (как и было) entry = f"[{now}] {role}: {text}\n" try: with self.lock: with open(self.current_file, "a", encoding="utf-8") as f: f.write(entry) except Exception as e: print(f"File write error: {e}") # 2. ЗАПИСЬ В ПАМЯТЬ with self.history_lock: self.history.append({ 'role': role, 'text': text, 'time': now }) return callback_func # 4. Запуск стримов try: # Используем те же параметры, что в рабочем демо with sd.RawInputStream(samplerate=self.samplerate, blocksize=8000, device=id_gm, dtype='int16', channels=1, callback=create_callback(rec_gm, "GM")), \ sd.RawInputStream(samplerate=self.samplerate, blocksize=8000, device=id_pcs, dtype='int16', channels=1, callback=create_callback(rec_pcs, "PCs")): print(f"✅🎤 Транскрибация запущена (IDs: {id_gm}, {id_pcs})") while not self._stop_event.is_set(): self._stop_event.wait(1.0) # Экономнее чем sleep except Exception as e: print(f"❌🎤 Критическая ошибка транскрибации: {e}") finally: self.is_running = False print("🏁🎤 Сессия транскрибации завершена") def get_context_by_limit(self, limit, separator = '', order='normal'): """ Собирает реплики с конца, пока не превысит лимит символов. Возвращает склеенную строку всех ролей. """ with self.history_lock: if not self.history: return "" selected_phrases = [] current_length = 0 # Идем с конца списка for item in reversed(self.history): phrase = f"{item['role']}: {item['text']}\n" selected_phrases.append(phrase) current_length += len(phrase) # Если превысили лимит, останавливаемся if current_length >= limit: break if order == 'normal': # Возвращаем в хронологическом порядке (как в чате) return separator.join(reversed(selected_phrases)) else: # Возвращаем как есть (новые сверху) return separator.join(selected_phrases) def get_new_gm_text(self, start_index): """Возвращает склеенный текст GM и новый индекс последнего элемента.""" with self.history_lock: current_max_index = len(self.history) - 1 if start_index >= current_max_index: return "", start_index # Собираем только реплики GM, начиная с start_index + 1 new_phrases = [ item['text'] for item in self.history[start_index + 1:] if item['role'] == "GM" ] return " ".join(new_phrases), current_max_index