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 import io import base64 # Для генерации графиков на сервере без UI! import matplotlib matplotlib.use('Agg') # Для работы без GUI import matplotlib.pyplot as plt import pandas as pd 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.metrics_data = { "graphs": [], # Массив Base64 строчек с HTML-декором "text": "Анализ пока не выполнялся или недостаточно данных..." } self.lock = threading.Lock() self._stop_event = threading.Event() self.abs_logs_path = None def update_logs_config(self, abs_path): """Обновление путей из настроек Obsidian.""" self.abs_logs_path = abs_path 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 -> 10:20:35.456] GM: текст pattern = re.compile(r"\[(\d{2}:\d{2}:\d{2}(?:\.\d+)?)\s*(?:->|-->)\s*(\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: start_time, end_time, role, text = match.groups() new_history.append({ 'role': role, 'text': text, 'startTime': start_time, 'endTime': end_time }) 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 # Очистка вкладок при новом старте if not continue_last: self.responses["regular"].clear() self.responses["commands"].clear() self.metrics_data = {"graphs": [], "text": "Ожидание достаточного количества данных..."} log_dir = self.abs_logs_path if self.abs_logs_path else "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(".md")]) 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.md") 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) rec_states = { "GM": {"start": datetime.datetime.now()}, "PCs": {"start": datetime.datetime.now()} } # 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: end_dt = datetime.datetime.now() start_dt = rec_states[role]["start"] # Форматируем с миллисекундами start_str = start_dt.strftime("%H:%M:%S.%f")[:-3] end_str = end_dt.strftime("%H:%M:%S.%f")[:-3] # Обновляем старт для следующей фразы rec_states[role]["start"] = end_dt # 1. Запись в файл с двумя метками entry = f"[{start_str} --> {end_str}] {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, 'startTime': start_str, 'endTime': end_str }) 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 def generate_metrics_graphs(self): """Анализирует history и генерирует графики (Matplotlib -> Base64)""" with self.history_lock: if not self.history: return history_copy = list(self.history) if len(history_copy) < 3: return try: df = pd.DataFrame(history_copy) df['tokens'] = df['text'].apply(lambda x: len(str(x).split())) df['chars'] = df['text'].apply(lambda x: len(str(x))) stats = df.groupby('role').agg( chars_total=('chars', 'sum'), tokens_total=('tokens', 'sum'), replica_count=('text', 'count') ) total_tokens = stats['tokens_total'].sum() stats['percent_tokens'] = (stats['tokens_total'] / total_tokens * 100).fillna(0) stats['avg_len'] = (stats['tokens_total'] / stats['replica_count']).fillna(0) # Расчет WPM (Скорость речи) wpm_data = {} for role in stats.index: role_df = df[df['role'] == role] total_duration_sec = 0.0 for _, row in role_df.iterrows(): try: t_format = "%H:%M:%S.%f" if "." in row['startTime'] else "%H:%M:%S" e_format = "%H:%M:%S.%f" if "." in row['endTime'] else "%H:%M:%S" t1 = datetime.datetime.strptime(row['startTime'], t_format) t2 = datetime.datetime.strptime(row['endTime'], e_format) total_duration_sec += (t2 - t1).total_seconds() except: pass duration_min = total_duration_sec / 60.0 wpm_data[role] = stats.loc[role, 'tokens_total'] / max(duration_min, 0.05) generated_images = [] def fig_to_base64(fig): buf = io.BytesIO() fig.savefig(buf, format='png', bbox_inches='tight', dpi=100) plt.close(fig) buf.seek(0) return base64.b64encode(buf.read()).decode('utf-8') # --- ПЕРВАЯ КАРТИНКА: Распределение (слева) и Средняя длина (справа) --- fig1, (ax1_1, ax1_2) = plt.subplots(1, 2, figsize=(12, 5)) # Слева: Распределение реплик (%) ax1_1.barh(stats.index, stats["percent_tokens"], color="skyblue", alpha=0.7) ax1_1.set_title("Распределение реплик (по словам, %)") ax1_1.set_xlim(0, 100) ax1_1.grid(axis="x", linestyle="--", alpha=0.6) # Справа: Средняя длина ax1_2.bar(stats.index, stats["avg_len"], color="teal", alpha=0.6) ax1_2.set_title("Средняя длина (слов на реплику)") fig1.tight_layout() generated_images.append(f'') # --- ВТОРАЯ КАРТИНКА: Общее кол-во (слева) и Скорость речи (справа) --- fig2, (ax2_1, ax2_2) = plt.subplots(1, 2, figsize=(12, 5)) # Слева: Общее количество реплик ax2_1.bar(stats.index, stats["replica_count"], color="coral", alpha=0.6) ax2_1.set_title("Общее кол-во реплик") # Справа: Скорость речи (WPM) roles_list = stats.index.tolist() wpm_values = [wpm_data.get(r, 0) for r in roles_list] ax2_2.bar(roles_list, wpm_values, color='#9b59b6', alpha=0.7) ax2_2.set_title("Скорость речи (WPM)") ax2_2.set_ylabel("Слов в минуту") ax2_2.grid(axis='y', linestyle='--', alpha=0.6) fig2.tight_layout() generated_images.append(f'') self.metrics_data["graphs"] = generated_images print("✅ Аналитика (сдвоенные графики) обновлена.") except Exception as e: print(f"❌ Ошибка генерации графиков: {e}")