import time import threading import json import queue import os from collections import Counter, deque import re from typing import Any, Deque, Dict, Optional, Set, List, Iterator from flask import Flask from flask_cors import CORS import ollama as _ollama from ollama import chat from ollama import ChatResponse import numpy as np import sounddevice as sd from faster_whisper import WhisperModel from gui import select_settings, prompt_input_sample_rate, run_runtime_dashboard from routes import register_routes from config import _SYSTEM_PROMPT, _LLM_EMPTY_SENTINELS, _HALLUCINATION_PHRASES TARGET_SAMPLE_RATE: int = 16000 CAPTURE_SAMPLE_RATE: int = 0 BUFFER_SECONDS: float = 10 MAX_SAMPLES: int = 0 PROCESS_INTERVAL_SECONDS: float = 2 SSE_EVENT_SUBTITLE: str = "subtitle" SSE_EVENT_AUDIO_ACTIVITY: str = "audio_activity" SSE_KEEPALIVE_SECONDS: int = 15 RUNTIME_SUBTITLE_LINES_MAX: int = 120 RUNTIME_LOG_LINES_MAX: int = 300 USE_OLLAMA_CLEANUP: bool = True OLLAMA_MODEL: str = "qwen2.5:7b-instruct" OLLAMA_CONTEXT_WINDOW: int = 6 # number of recent cleaned segments kept as context OLLAMA_OPTIONS: Dict[str, Any] = {"num_gpu": 1} RAW_BATCH_SIZE: int = 2 # accumulate this many raw Whisper lines before calling the LLM SETTINGS_PATH: str = os.path.join(os.path.dirname(__file__), "settings.json") DEFAULT_SETTINGS: Dict[str, Any] = { "audio_device_name": "", "model_name": "medium", "device": "cpu", "compute_type": "int8", "task": "translate", "beam_size": 3, "language": "", "context_seconds": 10, "update_interval_seconds": 2, "audio_activity_threshold": 0.003, "use_ollama_cleanup": True, "ollama_device": "GPU", "ollama_model": "qwen2.5:7b-instruct", "ollama_context_window": 5, "ollama_raw_batch_size": 1, } MODEL_CHOICES: List[str] = ["tiny", "base", "small", "medium", "large-v2", "large-v3", "distil-large-v3"] DEVICE_CHOICES: List[str] = ["cpu", "cuda", "auto"] COMPUTE_CHOICES: List[str] = ["int8", "int8_float16", "float16", "float32"] TASK_CHOICES: List[str] = ["translate", "transcribe"] audio_buffer: np.ndarray = np.zeros(0, dtype=np.float32) lock: threading.Lock = threading.Lock() model: Optional[WhisperModel] = None WHISPER_TASK: str = DEFAULT_SETTINGS["task"] WHISPER_BEAM_SIZE: int = DEFAULT_SETTINGS["beam_size"] WHISPER_LANGUAGE: str = DEFAULT_SETTINGS["language"] AUDIO_ACTIVITY_THRESHOLD: float = float(DEFAULT_SETTINGS["audio_activity_threshold"]) AUDIO_ACTIVITY_HOLD_SECONDS: float = 0.75 AUDIO_ACTIVITY_REPORT_INTERVAL_SECONDS: float = 0.5 last_subtitle_payload: Optional[Dict[str, Any]] = None last_audio_activity_payload: Dict[str, Any] = { "active": False, "rms": 0.0, "threshold": AUDIO_ACTIVITY_THRESHOLD, } _audio_active_until: float = 0.0 _audio_last_emit: float = 0.0 _audio_state_lock: threading.Lock = threading.Lock() recent_subtitle_lines: Deque[str] = deque(maxlen=RUNTIME_SUBTITLE_LINES_MAX) recent_subtitle_lines_lock: threading.Lock = threading.Lock() runtime_logs: Deque[str] = deque(maxlen=RUNTIME_LOG_LINES_MAX) runtime_logs_lock: threading.Lock = threading.Lock() clients: Set[queue.Queue] = set() clients_lock: threading.Lock = threading.Lock() SERVER_HOST: str = "127.0.0.1" SERVER_PORT: int = 5000 app: Flask = Flask(__name__) CORS(app) # OLLAMA stuff llm_input_queue: queue.Queue = queue.Queue(maxsize=1) subtitle_context: Deque[str] = deque(maxlen=OLLAMA_CONTEXT_WINDOW) # sliding window context subtitle_context_lock: threading.Lock = threading.Lock() _raw_batch: List[str] = [] _raw_batch_lock: threading.Lock = threading.Lock() def resample_audio(audio_np: np.ndarray, src_rate: int, dst_rate: int) -> np.ndarray: """ Resamples audio to TARGET_SAMPLE_RATE (default is 16000hz), speeds up inference time, fetched as a nd array """ if src_rate == dst_rate: return audio_np if len(audio_np) == 0: return audio_np dst_len = int(len(audio_np) * dst_rate / src_rate) if dst_len <= 0: return audio_np[:0] x_old = np.arange(len(audio_np)) x_new = np.linspace(0, len(audio_np) - 1, dst_len) return np.interp(x_new, x_old, audio_np).astype(np.float32) def load_settings() -> Dict[str, Any]: if not os.path.exists(SETTINGS_PATH): return DEFAULT_SETTINGS.copy() try: with open(SETTINGS_PATH, "r", encoding="utf-8") as handle: data = json.load(handle) except (OSError, json.JSONDecodeError): return DEFAULT_SETTINGS.copy() merged: Dict[str, Any] = DEFAULT_SETTINGS.copy() for key, value in data.items(): if key in merged: merged[key] = value return merged def save_settings(settings: Dict[str, Any]) -> None: try: with open(SETTINGS_PATH, "w", encoding="utf-8") as handle: json.dump(settings, handle, indent=2) except OSError as exc: print(f"Failed to save settings: {exc}") def cleanup_subtitle_with_ollama(raw_text: str, context: List[str]) -> Optional[str]: if context: context_block = "\n".join(f"- {seg}" for seg in context) else: context_block = "(none yet)" user_message = ( f"ALREADY SHOWN:\n{context_block}\n\n" "RAW INPUT (multiple consecutive transcriptions of the same rolling window — " f"deduplicate and extract only the genuinely new spoken content as one subtitle):\n{raw_text}" ) try: response: ChatResponse = chat( model=OLLAMA_MODEL, messages=[ {"role": "system", "content": _SYSTEM_PROMPT}, {"role": "user", "content": user_message}, ], options=OLLAMA_OPTIONS, ) return response.message.content.strip() except Exception as exc: print(f"⚠️ OLLAMA cleanup error: {exc}") return None def ensure_ollama_ready() -> None: """ Pulls Ollama model is necessary, checks model is downloaded """ try: local = _ollama.list() except Exception as exc: raise RuntimeError( f"Cannot reach Ollama — is the server running? ({exc})" ) from exc model_names: List[str] = [m.model for m in local.models] if not any(name.startswith(OLLAMA_MODEL) for name in model_names): print(f" '{OLLAMA_MODEL}' not found locally — pulling (this may take a while) ...") try: _ollama.pull(OLLAMA_MODEL) print(" Pull complete.") except Exception as exc: raise RuntimeError(f"Failed to pull model '{OLLAMA_MODEL}': {exc}") from exc else: print(f" Model found locally.") print(" Warming up model, almost done ...") try: chat( model=OLLAMA_MODEL, messages=[{"role": "user", "content": "Ready?"}], options=OLLAMA_OPTIONS, ) print(" ✅ Ollama is ready.") except Exception as exc: raise RuntimeError(f"Ollama warm-up failed: {exc}") from exc def normalize_llm_output(text: str) -> str: if text.strip().lower().rstrip(".") in _LLM_EMPTY_SENTINELS: return "" return text def add_runtime_log(kind: str, message: str) -> None: timestamp = time.strftime("%H:%M:%S") line = f"[{timestamp}] [{kind.upper()}] {message}" with runtime_logs_lock: runtime_logs.append(line) def is_hallucination(text: str) -> Optional[str]: """ Algorithmic hallucination detection by checking if the output from whisper is unusually long given sliding window length, or if there are too many repeating words/phrases False-alarms generally do not impact quality since the same information is likely captured in the previous subtitle """ words = text.split() if not words: return None max_expected = int(BUFFER_SECONDS * 4.5) if len(words) > max_expected: return f"too long: {len(words)} words > {max_expected}" clean = [re.sub(r"[^\w']+", "", w).lower() for w in words] clean = [w for w in clean if w] for n in [2, 3]: if len(clean) < n * 3: continue ngrams = [" ".join(clean[i : i + n]) for i in range(len(clean) - n + 1)] top, count = Counter(ngrams).most_common(1)[0] if count >= 3: tokens_covered = count * n if tokens_covered / max(1, len(clean)) > 0.35: return f"repeating phrase '{top}' x{count} (covers {tokens_covered}/{len(clean)} tokens)" top, count = Counter(clean).most_common(1)[0] if count >= 4 and count / len(clean) > 0.40: return f"repeating token '{top}' x{count} ({count/len(clean):.0%})" normalized = re.sub(r"[^\w\s]", "", text.lower()).strip() if normalized in _HALLUCINATION_PHRASES: return "blocked phrase pattern" return None def llm_processing_loop() -> None: print(f"LLM cleanup thread started (model={OLLAMA_MODEL})") while True: try: raw_text: str = llm_input_queue.get(timeout=1) except queue.Empty: continue with subtitle_context_lock: context = list(subtitle_context) cleaned: Optional[str] = cleanup_subtitle_with_ollama(raw_text, context) if cleaned is None: add_runtime_log("LLM", "cleanup failed, falling back to raw text") cleaned = raw_text else: cleaned = normalize_llm_output(cleaned) if cleaned: with subtitle_context_lock: subtitle_context.append(cleaned) add_runtime_log("FINAL", cleaned) broadcast_subtitle(cleaned) else: add_runtime_log("LLM", "no new content from cleanup") def run_whisper(audio_np: np.ndarray) -> str: transcribe_kwargs: Dict[str, Any] = {"task": WHISPER_TASK, "beam_size": WHISPER_BEAM_SIZE} if WHISPER_LANGUAGE: transcribe_kwargs["language"] = WHISPER_LANGUAGE assert model is not None, "Whisper model is not initialized" segments, _info = model.transcribe(audio_np, **transcribe_kwargs) text = " ".join(seg.text for seg in segments).strip() if not text: return text add_runtime_log("RAW", text) hallucination_reason = is_hallucination(text) if hallucination_reason: add_runtime_log("HALLUCINATION", f"{hallucination_reason} | text={text}") return text if USE_OLLAMA_CLEANUP: with _raw_batch_lock: _raw_batch.append(text) if len(_raw_batch) >= RAW_BATCH_SIZE: batch_text = "\n".join(_raw_batch) _raw_batch.clear() else: batch_text = None if batch_text is not None: add_runtime_log("RAW->LLM", batch_text.replace("\n", " || ")) try: llm_input_queue.put_nowait(batch_text) except queue.Full: add_runtime_log("LLM", "queue full, dropping previous batch") try: llm_input_queue.get_nowait() except queue.Empty: pass try: llm_input_queue.put_nowait(batch_text) except queue.Full: add_runtime_log("LLM", "queue still full, skipped batch") else: add_runtime_log("FINAL", text) broadcast_subtitle(text) return text def broadcast_event(event: str, payload: Dict[str, Any]) -> None: message: Dict[str, Any] = {"event": event, "payload": payload} with clients_lock: targets = list(clients) for client_queue in targets: try: client_queue.put_nowait(message) except queue.Full: pass def broadcast_subtitle(text: str) -> None: global last_subtitle_payload payload: Dict[str, Any] = {"text": text} last_subtitle_payload = payload with recent_subtitle_lines_lock: if not recent_subtitle_lines or recent_subtitle_lines[-1] != text: recent_subtitle_lines.append(text) broadcast_event(SSE_EVENT_SUBTITLE, payload) def get_audio_activity_snapshot() -> Dict[str, Any]: with _audio_state_lock: return dict(last_audio_activity_payload) def get_recent_subtitle_lines_snapshot() -> List[str]: with recent_subtitle_lines_lock: return list(recent_subtitle_lines) def get_runtime_logs_snapshot() -> List[str]: with runtime_logs_lock: return list(runtime_logs) def format_sse_event(event: str, payload: Dict[str, Any]) -> str: """ Creates an SSE event raw payload """ data = json.dumps(payload) return f"event: {event}\ndata: {data}\n\n" def event_stream() -> Iterator[str]: client_queue: queue.Queue = queue.Queue(maxsize=20) with clients_lock: clients.add(client_queue) if last_subtitle_payload: yield format_sse_event(SSE_EVENT_SUBTITLE, last_subtitle_payload) yield format_sse_event(SSE_EVENT_AUDIO_ACTIVITY, last_audio_activity_payload) try: while True: try: event_data = client_queue.get(timeout=SSE_KEEPALIVE_SECONDS) except queue.Empty: yield ": keep-alive\n\n" continue if not isinstance(event_data, dict): continue event_name = str(event_data.get("event", SSE_EVENT_SUBTITLE)) payload = event_data.get("payload", {}) if not isinstance(payload, dict): payload = {} yield format_sse_event(event_name, payload) finally: with clients_lock: clients.discard(client_queue) def start_subtitle_server() -> threading.Thread: """ Run flask app to expose processed data as a server-sent-events (subtitle) """ register_routes(app, event_stream) thread = threading.Thread( target=lambda: app.run( host=SERVER_HOST, port=SERVER_PORT, threaded=True, use_reloader=False, ), daemon=True, ) thread.start() print(f"SSE subtitle server listening on http://{SERVER_HOST}:{SERVER_PORT}/events") return thread def list_audio_devices() -> None: """ Get all audio devices """ devices = sd.query_devices() print("Available audio devices:") for idx, dev in enumerate(devices): io = [] if dev["max_input_channels"] > 0: io.append("input") if dev["max_output_channels"] > 0: io.append("output") io_str = "/".join(io) if io else "none" print(f"[{idx}] {dev['name']} ({io_str})") def publish_audio_activity(chunk_rms: float) -> None: global _audio_active_until, _audio_last_emit, last_audio_activity_payload now_mono = time.monotonic() if chunk_rms >= AUDIO_ACTIVITY_THRESHOLD: _audio_active_until = now_mono + AUDIO_ACTIVITY_HOLD_SECONDS with _audio_state_lock: active = now_mono <= _audio_active_until previous_active = bool(last_audio_activity_payload.get("active", False)) state_changed = active != previous_active report_due = (now_mono - _audio_last_emit) >= AUDIO_ACTIVITY_REPORT_INTERVAL_SECONDS if not state_changed and not report_due: return payload: Dict[str, Any] = { "active": active, "rms": round(chunk_rms, 6), "threshold": AUDIO_ACTIVITY_THRESHOLD, } last_audio_activity_payload = payload _audio_last_emit = now_mono broadcast_event(SSE_EVENT_AUDIO_ACTIVITY, payload) def audio_callback(indata: np.ndarray, frames: int, time_info: Any, status: Any) -> None: """ Callback definition for audio sink. Unload all data into global audio_buffer """ if status: print(f"Audio status: {status}") # Take first channel chunk: np.ndarray = indata[:, 0].copy() chunk_rms: float = float(np.sqrt(np.mean(np.square(chunk)))) if len(chunk) > 0 else 0.0 publish_audio_activity(chunk_rms) global audio_buffer with lock: audio_buffer = np.concatenate([audio_buffer, chunk]) if len(audio_buffer) > MAX_SAMPLES: audio_buffer = audio_buffer[-MAX_SAMPLES:] def is_silent(audio_16k: Optional[np.ndarray]) -> bool: """ Basic rudimentary silence detection, do not run whisper if rms value isn't reached """ if audio_16k is None or len(audio_16k) == 0: return False rms: float = float(np.sqrt(np.mean(np.square(audio_16k)))) # root mean square return rms < AUDIO_ACTIVITY_THRESHOLD def processing_loop() -> None: """ Core logic for processing incoming data Capture Audio -> If not silent -> Run Whisper on audio buffer """ while True: time.sleep(PROCESS_INTERVAL_SECONDS) with lock: if len(audio_buffer) == 0 or CAPTURE_SAMPLE_RATE <= 0: continue audio_copy: np.ndarray = audio_buffer.copy() capture_rate: int = CAPTURE_SAMPLE_RATE audio_16k: np.ndarray = resample_audio(audio_copy, capture_rate, TARGET_SAMPLE_RATE) if is_silent(audio_16k): continue run_whisper(audio_16k) def select_input_sample_rate(device_index: int, preferred_rate: int) -> int: """ Attempts to automatically identify the sample rate of audio sink. Otherwise prompt for input """ common_rates: List[int] = [48000, 44100, 32000, 24000, 22050, 16000, 12000, 8000] tried: Set[int] = set() for rate in [preferred_rate] + common_rates: if rate in tried or rate <= 0: continue tried.add(rate) try: sd.check_input_settings(device=device_index, channels=1, samplerate=rate, dtype="float32") return rate except sd.PortAudioError: continue return prompt_input_sample_rate(device_index, common_rates) def main() -> None: global CAPTURE_SAMPLE_RATE, MAX_SAMPLES, model, WHISPER_TASK, WHISPER_BEAM_SIZE, WHISPER_LANGUAGE global BUFFER_SECONDS, PROCESS_INTERVAL_SECONDS, USE_OLLAMA_CLEANUP global OLLAMA_MODEL, OLLAMA_CONTEXT_WINDOW, RAW_BATCH_SIZE, subtitle_context global AUDIO_ACTIVITY_THRESHOLD, last_audio_activity_payload, _audio_active_until, _audio_last_emit start_subtitle_server() settings: Dict[str, Any] = load_settings() devices = sd.query_devices() input_devices = [(idx, dev) for idx, dev in enumerate(devices) if dev["max_input_channels"] > 0] settings = select_settings( settings, input_devices, DEFAULT_SETTINGS, MODEL_CHOICES, DEVICE_CHOICES, COMPUTE_CHOICES, TASK_CHOICES, ) save_settings(settings) USE_OLLAMA_CLEANUP = bool(settings.get("use_ollama_cleanup", True)) OLLAMA_OPTIONS["num_gpu"] = 0 if settings.get("ollama_device", "CPU").upper() == "CPU" else 1 OLLAMA_MODEL = "qwen2.5:7b-instruct" if str(settings.get("ollama_model", OLLAMA_MODEL)) is None else str(settings.get("ollama_model", OLLAMA_MODEL)) OLLAMA_CONTEXT_WINDOW = int(settings.get("ollama_context_window", 6)) subtitle_context = deque(maxlen=OLLAMA_CONTEXT_WINDOW) RAW_BATCH_SIZE = int(settings.get("ollama_raw_batch_size", 3)) if USE_OLLAMA_CLEANUP: ensure_ollama_ready() llm_thread = threading.Thread(target=llm_processing_loop, daemon=True) llm_thread.start() device_name: str = settings.get("audio_device_name", "") matched_index: Optional[int] = None for idx, dev in enumerate(devices): if dev.get("name") == device_name and dev.get("max_input_channels", 0) > 0: matched_index = idx break if matched_index is None: raise RuntimeError("Saved audio device not found. Please reselect in the settings window.") device_index: int = matched_index model_name: str = settings["model_name"] whisper_device: str = settings["device"] compute_type: str = settings["compute_type"] WHISPER_TASK = settings["task"] WHISPER_BEAM_SIZE = int(settings["beam_size"]) WHISPER_LANGUAGE = settings["language"].strip() if settings["language"] else "" BUFFER_SECONDS = float(settings.get("context_seconds", BUFFER_SECONDS)) PROCESS_INTERVAL_SECONDS = float(settings.get("update_interval_seconds", PROCESS_INTERVAL_SECONDS)) AUDIO_ACTIVITY_THRESHOLD = float(settings.get("audio_activity_threshold", AUDIO_ACTIVITY_THRESHOLD)) if BUFFER_SECONDS <= 0: BUFFER_SECONDS = DEFAULT_SETTINGS["context_seconds"] if PROCESS_INTERVAL_SECONDS <= 0: PROCESS_INTERVAL_SECONDS = DEFAULT_SETTINGS["update_interval_seconds"] if AUDIO_ACTIVITY_THRESHOLD <= 0: AUDIO_ACTIVITY_THRESHOLD = float(DEFAULT_SETTINGS["audio_activity_threshold"]) last_audio_activity_payload = { "active": False, "rms": 0.0, "threshold": AUDIO_ACTIVITY_THRESHOLD, } _audio_active_until = 0.0 _audio_last_emit = 0.0 broadcast_event(SSE_EVENT_AUDIO_ACTIVITY, last_audio_activity_payload) model = WhisperModel(model_name, device=whisper_device, compute_type=compute_type) device_info = sd.query_devices(device_index) preferred_rate: int = int(device_info["default_samplerate"]) if preferred_rate <= 0: preferred_rate = 48000 CAPTURE_SAMPLE_RATE = select_input_sample_rate(device_index, preferred_rate) MAX_SAMPLES = int(CAPTURE_SAMPLE_RATE * BUFFER_SECONDS) print(f"Using device {device_index}: {device_info['name']}") print(f"Model: {model_name} | task={WHISPER_TASK} | beam_size={WHISPER_BEAM_SIZE}") print(f"Compute: device={whisper_device} | compute_type={compute_type}") print(f"Capture sample rate: {CAPTURE_SAMPLE_RATE} Hz (resampling to {TARGET_SAMPLE_RATE} Hz)") print(f"Audio activity threshold (RMS): {AUDIO_ACTIVITY_THRESHOLD}") print(f"Ollama cleanup: {'enabled' if USE_OLLAMA_CLEANUP else 'disabled'} (model={OLLAMA_MODEL})") processing_thread = threading.Thread(target=processing_loop, daemon=True) processing_thread.start() with recent_subtitle_lines_lock: recent_subtitle_lines.clear() with runtime_logs_lock: runtime_logs.clear() add_runtime_log("SYSTEM", "Runtime dashboard started") add_runtime_log("SYSTEM", f"Device: {device_info['name']} @ {CAPTURE_SAMPLE_RATE} Hz") add_runtime_log("SYSTEM", f"Task={WHISPER_TASK} | Beam={WHISPER_BEAM_SIZE} | Cleanup={'on' if USE_OLLAMA_CLEANUP else 'off'}") stream = sd.InputStream( device=device_index, channels=1, samplerate=CAPTURE_SAMPLE_RATE, dtype="float32", callback=audio_callback, blocksize=int(CAPTURE_SAMPLE_RATE * 0.5), ) def _on_dashboard_close() -> None: print("Stopping.") try: stream.start() print("Listening... Close the runtime window to stop.") run_runtime_dashboard( get_audio_activity=get_audio_activity_snapshot, get_runtime_logs=get_runtime_logs_snapshot, get_subtitle_lines=get_recent_subtitle_lines_snapshot, on_close=_on_dashboard_close, ) finally: try: stream.stop() except Exception: pass try: stream.close() except Exception: pass if __name__ == "__main__": main()