speech_to_text.py
Rama main · réplica de lectura del repositorio original.
import argparse
import os
import sys
import warnings
from typing import ClassVar
# Determine application directory (works for both dev and PyInstaller bundle).
# APP_DIR = user data dir: config, models, logs, per-session DBs.
# BUNDLE_DIR = bundled read-only assets: templates, static, config.default.
# Every non-override run (frozen OR run-from-repo) uses ~/.stt, so the data dir
# is always the per-user, always-writable location and a run-from-repo server no
# longer writes data into the checkout. STT_DATA_DIR (set by the watchdog) still
# overrides, so a managed worker follows whatever path the watchdog chose.
_data_override = os.environ.get("STT_DATA_DIR")
_script_dir = os.path.dirname(os.path.abspath(__file__))
_is_frozen = getattr(sys, "frozen", False)
# BUNDLE_DIR follows the freeze, not the data override: a frozen build's templates,
# static assets and config templates only exist under _MEIPASS, so resolving them
# next to this file would leave a frozen server with no pages to render. The two
# settings are independent — the watchdog sets STT_DATA_DIR but runs the app
# unfrozen from a checkout, and the demo build is frozen with its own data dir.
BUNDLE_DIR = sys._MEIPASS if _is_frozen else _script_dir
if _data_override:
APP_DIR = os.path.abspath(os.path.expanduser(_data_override))
else:
APP_DIR = os.path.join(os.path.expanduser("~"), ".stt")
# Root running a checkout somebody else owns (a User=root service from an older
# install.sh) must not leave root-owned __pycache__ in it; stt/owner_exec.py has the
# rest of that story. Before the first stt import, or that import writes the first one.
if not _is_frozen and getattr(os, "geteuid", None) is not None and os.geteuid() == 0:
try:
if os.stat(BUNDLE_DIR).st_uid != 0:
sys.dont_write_bytecode = True
except OSError:
pass
# A demo replays a recorded service instead of transcribing one: no worker process,
# no models, no microphone. Its data root is chosen (and wiped) before anything is
# created, so a demo can never write into a real install. See stt/demo_mode.py.
from stt import demo_guard as _demo_guard # noqa: E402
from stt import demo_mode as _demo_mode # noqa: E402
DEMO = _demo_mode.enabled()
DEMO_PORT = 0
if DEMO:
_demo_install_id = _demo_mode.read_install_id() # before the wipe takes it
APP_DIR = _demo_mode.prepare_data_dir(BUNDLE_DIR)
# --port / STT_DEMO_PORT if either was given, otherwise the usual 8099 search.
# A requested port that is taken still falls through to the next free one: the
# control window then shows where the demo actually is, which beats dying at bind.
DEMO_PORT = _demo_mode.pick_port(start=_demo_mode.requested_port() or _demo_mode.DEFAULT_PORT)
_demo_mode.write_config(APP_DIR, BUNDLE_DIR, DEMO_PORT, install_id=_demo_install_id)
os.makedirs(APP_DIR, exist_ok=True)
MODELS_DIR = os.path.join(APP_DIR, "models")
os.makedirs(MODELS_DIR, exist_ok=True)
# Default base directory for database + audio backups (rooted in APP_DIR so compiled
# builds keep all data under ~/.stt instead of the launch directory).
BACKUP_DIR = os.path.join(APP_DIR, "_AUTOMATIC_BACKUP")
# Live (user-mutated, gitignored) config files live in APP_DIR/config next to
# their tracked *.default.json templates. Templates ship with the checkout;
# live files are seeded from them on first run and never touched by updates.
CONFIG_DIR = os.path.join(APP_DIR, "config")
os.makedirs(CONFIG_DIR, exist_ok=True)
# One-time layout migration: live config files used to sit in APP_DIR root.
# stt/watchdog.py performs the same move-if-absent migration; whichever
# process starts first wins, the other finds nothing left to move.
for _mig_name in ("config.json", "custom_dictionary.json", "word_highlighting.json",
"whisper_models.json", "faster_whisper_models.json"):
_mig_old = os.path.join(APP_DIR, _mig_name)
_mig_new = os.path.join(CONFIG_DIR, _mig_name)
if os.path.exists(_mig_old) and not os.path.exists(_mig_new):
try:
import shutil as _mig_shutil
_mig_shutil.move(_mig_old, _mig_new)
print(f"[MIGRATE] {_mig_name} -> config/{_mig_name}")
except OSError as _mig_err:
print(f"[MIGRATE] Could not move {_mig_name}: {_mig_err}")
# Path-containment guards live in stt/paths.py (importable, unit-tested);
# safe_model_path is re-imported and safe_managed_path keeps its APP_DIR default.
from stt import paths as _paths
from stt.paths import safe_model_path # noqa: F401
from stt.coercion import coerce_bool, coerce_float, coerce_int
from stt.server_port import DEFAULT_PORT as _DEFAULT_PORT
from stt import server_restart as _server_restart
from stt import pair_tokens as _pair_tokens_mod
# How a dying worker process reports itself, and why excepthooks are chained
# rather than replaced.
from stt import worker_crash as _worker_crash
from stt import shutdown_channel as _shutdown_channel
from stt import fd_limit as _fd_limit
# Clear-all-screens markers: which captions the live displays may still show.
from stt import display_clear as _display_clear
# What may appear in an operator's support report, by allowlist.
from stt import diagnostics as _diagnostics
from stt import hq_audio as _hq_audio
from stt import model_disk as _model_disk
from stt import engineio_compat as _engineio_compat
from stt import caption_order as _caption_order_mod
from stt.http_params import merge_request_params, parse_json_body as _parse_json_body
from stt.model_disk import _CT2_MARKER, dir_has_weights, dir_is_writable, has_weight_file, is_weight_file, model_presence # noqa: F401
from stt import model_catalog as _model_catalog
from stt import model_files as _model_files
from stt import session_edit as _session_edit
from stt import start_watch as _start_watch
from stt import digital_silence as _digital_silence
def safe_managed_path(path, base_dir=None):
"""Resolve a file-manager path and return its realpath only if it stays
inside base_dir (defaults to APP_DIR). See stt/paths.py."""
return _paths.safe_managed_path(path, base_dir if base_dir is not None else APP_DIR)
def _seed_from_bundle(filename):
"""Return the live path of a config file (CONFIG_DIR/<filename>), seeding it
from the bundled template (config/<stem>.default.json) on first run."""
import shutil
stem, ext = os.path.splitext(filename)
dst = os.path.join(CONFIG_DIR, filename)
src = os.path.join(BUNDLE_DIR, "config", f"{stem}.default{ext}")
if not os.path.exists(dst) and os.path.exists(src):
try:
shutil.copy2(src, dst)
except Exception as e:
print(f"[INIT] Could not seed {filename} from bundle: {e}")
return dst
def _chmod_quiet(path, mode):
"""Best-effort chmod; ignore failures (not owner, fs without unix perms, etc.)."""
try:
os.chmod(path, mode)
except OSError:
pass
def make_db_world_readable(db_path):
"""Make a DB file and its WAL/SHM/journal sidecars world-readable (a+r, 0644),
so databases written by the service (often as root) can be read by all users
and downstream consumers."""
if not db_path:
return
for suffix in ("", "-wal", "-shm", "-journal"):
p = db_path + suffix
if os.path.exists(p):
_chmod_quiet(p, 0o644)
def make_dirs_world_readable(leaf_dir, base_dir=None):
"""Make leaf_dir world-readable and traversable (a+rx, 0755). When base_dir is
given and contains leaf_dir, every directory from base_dir down to leaf_dir is
updated too, so DB files written inside are reachable by all users."""
if not leaf_dir:
return
leaf = os.path.abspath(leaf_dir)
if not base_dir:
_chmod_quiet(leaf, 0o755)
return
base = os.path.abspath(base_dir)
_chmod_quiet(base, 0o755)
if leaf == base:
return
if leaf.startswith(base + os.sep):
cur = base
for part in os.path.relpath(leaf, base).split(os.sep):
cur = os.path.join(cur, part)
_chmod_quiet(cur, 0o755)
else:
_chmod_quiet(leaf, 0o755)
def make_tree_world_readable(root):
"""Recursively make every directory (a+rx, 0755) and file (a+r, 0644) under
root readable by all users. Best-effort; skips entries it cannot chmod. Used to
sweep the whole DB/backup folder, including files created during stop cleanup."""
if not root or not os.path.isdir(root):
return
_chmod_quiet(root, 0o755)
for dirpath, dirnames, filenames in os.walk(root):
for d in dirnames:
_chmod_quiet(os.path.join(dirpath, d), 0o755)
for f in filenames:
_chmod_quiet(os.path.join(dirpath, f), 0o644)
# Suppress NNPACK warnings from PyTorch (harmless but spammy)
# These are C++ warnings so we need to disable at the PyTorch level
os.environ['NNPACK_DISABLE'] = '1'
os.environ['PYTORCH_NNPACK_WARN'] = '0'
# macOS: allow fork() after Objective-C threads are initialized (Flask, PyTorch, etc.)
# Without this, multiprocessing with fork start method crashes on macOS
os.environ['OBJC_DISABLE_INITIALIZE_FORK_SAFETY'] = 'YES'
warnings.filterwarnings('ignore', message='.*NNPACK.*')
# Set HuggingFace cache to local models directory BEFORE any HF imports
# This prevents models from being downloaded to ~/.cache/huggingface/hub
_models_cache_dir = os.path.join(MODELS_DIR, ".hf_cache")
os.makedirs(_models_cache_dir, exist_ok=True)
os.environ["HF_HUB_CACHE"] = _models_cache_dir
os.environ["HF_HOME"] = _models_cache_dir
os.environ["HUGGINGFACE_HUB_CACHE"] = _models_cache_dir
# TTS models directory (for piper models)
_tts_cache_dir = os.path.join(MODELS_DIR, "tts")
os.makedirs(_tts_cache_dir, exist_ok=True)
import sqlite3
import logging
import signal
import threading
import multiprocessing
from multiprocessing import Queue as MPQueue
import json
import re
import secrets
import shutil
import socket
import statistics
from pathlib import Path
from datetime import timedelta, datetime
from queue import Queue, Empty, Full
from time import sleep
import time
from sys import platform
from stt.file_mover import execute_file_move_now, execute_file_move
# Tracks the most recent file mover run so the UI can show live activity/status.
# Updated by both the automatic (transcription-stop) path and the manual trigger.
_file_mover_runtime_lock = threading.Lock()
file_mover_runtime = {
"state": "idle", # idle | running | success | error
"trigger": None, # "auto" | "manual"
"started_at": None, # ISO timestamp
"finished_at": None, # ISO timestamp
"moved": 0,
"failed": 0,
"message": "",
}
def set_file_mover_running(trigger):
"""Mark the file mover as actively running (so the UI shows a live indicator)."""
with _file_mover_runtime_lock:
file_mover_runtime.update({
"state": "running",
"trigger": trigger,
"started_at": datetime.now().isoformat(),
"finished_at": None,
"message": "File move in progress...",
})
def set_file_mover_result(trigger, result):
"""Record the outcome of a completed file mover run."""
with _file_mover_runtime_lock:
file_mover_runtime.update({
"state": "success" if result.get("success") else "error",
"trigger": trigger,
"finished_at": datetime.now().isoformat(),
"moved": result.get("moved", 0),
"failed": result.get("failed", 0),
"message": result.get("message", ""),
})
def get_file_mover_runtime():
"""Return a thread-safe copy of the current file mover runtime status."""
with _file_mover_runtime_lock:
return dict(file_mover_runtime)
import functools
import inspect
from flask import Flask, render_template, jsonify, request, redirect, send_from_directory, make_response, g, Response, stream_with_context
from flask_socketio import SocketIO, emit
# Both are used only by the transcription worker (capture buffer, PANNs, the audio
# loop) and by local TTS — never by the web layer or the demo's playback. Guarding
# them lets the demo build leave numpy out of the bundle entirely (~35MB). A real
# install that is missing them fails at the top of thread1_function with a clear
# message rather than silently here.
try:
import speech_recognition as sr
except ImportError: # pragma: no cover - present in every real install
sr = None
try:
import numpy as np
except ImportError: # pragma: no cover - present in every real install
np = None
# Heavy ML imports will be loaded lazily when needed
torch = None
whisper = None
AutoModelForSpeechSeq2Seq = None
AutoProcessor = None
AutoModelForCTC = None
Wav2Vec2Processor = None
pipeline = None
HfApi = None
model_info = None
def _lazy_import_ml_libraries():
"""Import heavy ML libraries only when needed"""
global \
torch, \
whisper, \
AutoModelForSpeechSeq2Seq, \
AutoProcessor, \
AutoModelForCTC, \
Wav2Vec2Processor, \
pipeline, \
HfApi, \
model_info
if torch is None:
import torch as _torch
torch = _torch
print("[INFO] PyTorch loaded")
if whisper is None:
import whisper as _whisper
whisper = _whisper
print("[INFO] Whisper loaded")
if AutoModelForSpeechSeq2Seq is None:
from transformers import (
AutoModelForSpeechSeq2Seq as _AutoModelForSpeechSeq2Seq,
AutoProcessor as _AutoProcessor,
AutoModelForCTC as _AutoModelForCTC,
Wav2Vec2Processor as _Wav2Vec2Processor,
pipeline as _pipeline,
)
from huggingface_hub import HfApi as _HfApi, model_info as _model_info
AutoModelForSpeechSeq2Seq = _AutoModelForSpeechSeq2Seq
AutoProcessor = _AutoProcessor
AutoModelForCTC = _AutoModelForCTC
Wav2Vec2Processor = _Wav2Vec2Processor
pipeline = _pipeline
HfApi = _HfApi
model_info = _model_info
print("[INFO] Transformers and HuggingFace Hub loaded")
# Suppress pydub ffmpeg warning - ffmpeg is only needed for specific audio formats
# WAV files (which we use) don't require ffmpeg
warnings.filterwarnings("ignore", message="Couldn't find ffmpeg or avconv")
# AudioSegment will be imported lazily when needed
AudioSegment = None
def _lazy_import_audio():
"""Import audio processing libraries only when needed"""
global AudioSegment
if AudioSegment is None:
from pydub import AudioSegment as _AudioSegment
AudioSegment = _AudioSegment
print("[INFO] AudioSegment loaded")
import tempfile
import uuid
import random
# Whisper decoding parameters optimized for streaming (3s chunks)
LIVE_TRANSCRIPTION_PARAMS = {
"beam_size": 3, # Matches config.default.json — accuracy win over greedy, live loop tolerates it
"best_of": 1, # No sampling
"temperature": 0.0, # Deterministic
"condition_on_previous_text": False, # Chunks lack context
"compression_ratio_threshold": 2.4,
"logprob_threshold": -1.0,
"no_speech_threshold": 0.6,
}
# Whisper decoding parameters optimized for file chunks (30s)
FILE_TRANSCRIPTION_PARAMS = {
"beam_size": 5, # Better quality for longer audio
"temperature": (0.0, 0.2, 0.4, 0.6, 0.8), # Fallback on failure
"condition_on_previous_text": True, # Chunks have context
"compression_ratio_threshold": 2.4,
"logprob_threshold": -1.0,
"no_speech_threshold": 0.6,
}
# Configuration file management
# The canonical default/template config lives in config.default.json (bundled at
# BUNDLE_DIR when compiled). It is used only to seed a fresh config.json on first
# run, or to recover from a missing/corrupted config.json — see load_config().
CONFIG_FILE = os.path.join(CONFIG_DIR, "config.json")
CONFIG_TEMPLATE_FILE = os.path.join(BUNDLE_DIR, "config", "config.default.json")
# Serializes all writers to config.json so concurrent endpoint saves cannot
# interleave and corrupt the file.
_config_file_lock = threading.Lock()
# Guards in-memory read-modify-write of the global `config` dict at the hotspots
# (the deep-merge in update_config, save_config's serialization, and the
# snapshots pushed to the transcription worker) so a reader never observes a
# half-mutated config. Reentrant: nested save_config / snapshot calls on one
# thread must not deadlock. Acquire order is always _config_lock -> _config_file_lock.
_config_lock = threading.RLock()
def _config_snapshot():
"""A coherent deep copy of the global config, taken under _config_lock, for
pushing to the hot-reload queue (a shallow copy would still share nested
dicts and could be observed mid-mutation)."""
import copy as _copy
with _config_lock:
return _copy.deepcopy(config)
# Config/validation/version helpers live in stt/config_utils.py (importable,
# unit-tested); names are re-imported here so call sites stay unchanged.
from stt import config_utils as _config_utils
# What the live-map pings say and when there is anything to say (stt/livemap.py);
# the HTTP call and the thread stay here.
from stt import livemap as _livemap
from stt import request_origin
from stt.session_reset import SessionTracker
from stt import tunnel as tunnel_mod
from stt.config_utils import ( # noqa: F401
SUPPORTED_AUDIO_FORMATS,
SUPPORTED_VIDEO_FORMATS,
_atomic_write_json,
_merge_missing_keys,
drop_retired_keys,
validate_file,
is_known_timezone as _is_known_timezone,
resolve_timezone as _resolve_timezone,
system_timezone as _system_timezone,
)
def save_config(config_to_save):
"""Save configuration to config.json atomically with error handling."""
try:
with _config_lock, _config_file_lock:
_atomic_write_json(CONFIG_FILE, config_to_save)
print(f"[OK] Configuration saved to '{CONFIG_FILE}'")
# Single choke point for session provenance: any settings change that
# reaches disk is recorded against the running session, whichever route
# made it. Guarded because config persistence is critical and must not
# fail on a provenance problem.
#
# The probe is not defensive padding: this module *does* call save_config
# during its own import — the password-generation block below runs at
# import and writes the generated password — and the sync function is not
# defined until later in the file. Catching the resulting NameError and
# warning made every fresh start print a provenance failure that had not
# happened: there is no session to record against during import
# (transcription_state is not bound yet), so there is nothing to sync and
# nothing lost. Distinguishing "not yet defined" from a real failure keeps
# the warning meaningful for the cases that are.
_sync = globals().get("_sync_session_meta_from_config")
if _sync is not None:
try:
_sync()
except Exception as e:
print(f"[SESSION-META] WARNING: could not sync provenance after save: {e}")
return True
except Exception as e:
print(f"[ERROR] Failed to save config: {e}")
return False
# Serialises first-time id generation. The app-start and transcription-start pings can
# both fire within a second of each other at boot when audio.autostart is on, and two
# threads racing here would generate two ids and persist whichever wrote last — leaving
# the map counting one machine as two installs.
_install_id_lock = threading.Lock()
def _get_install_id():
"""Stable anonymous UUID for the live-map ping; generated once and persisted."""
with _install_id_lock:
analytics = config.setdefault("analytics", {})
iid, changed = _livemap.ensure_install_id(analytics, lambda: str(uuid.uuid4()))
if changed:
save_config(config)
return iid
# Word highlighting uses a separate config file
WORD_HIGHLIGHTING_FILE = _seed_from_bundle("word_highlighting.json")
def load_word_highlighting():
"""Load word highlighting configuration from separate file"""
if os.path.exists(WORD_HIGHLIGHTING_FILE):
try:
with open(WORD_HIGHLIGHTING_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
print(f"[ERROR] Failed to load word highlighting config: {e}")
return {"enabled": True, "words": []}
def save_word_highlighting(data):
"""Save word highlighting configuration to separate file"""
try:
_atomic_write_json(WORD_HIGHLIGHTING_FILE, data, ensure_ascii=False)
return True
except Exception as e:
print(f"[ERROR] Failed to save word highlighting config: {e}")
return False
def _restore_config_from_template(reason=""):
"""Copy the bundled config.default.json over CONFIG_FILE. Returns True on success."""
return _config_utils.restore_config_from_template(CONFIG_TEMPLATE_FILE, CONFIG_FILE, reason)
def load_config():
"""Load configuration from config.json.
config.default.json (bundled, read from BUNDLE_DIR) is the canonical template.
It seeds a fresh config.json on first run and is used to recover from a
missing or corrupted config.json. On every load, settings present in the
template but missing from config.json (added in later releases) are patched
in and written back; user-set values are never overwritten."""
# First run: no config.json yet -> seed from the bundled template.
if not os.path.exists(CONFIG_FILE):
if _restore_config_from_template("create config.json"):
print(f"[OK] Created '{CONFIG_FILE}' from config.default.json")
print("[NOTE] Edit this file to configure your settings.")
else:
raise FileNotFoundError(
f"Neither '{CONFIG_FILE}' nor template '{CONFIG_TEMPLATE_FILE}' exist; cannot start."
)
try:
# encoding= is load-bearing: config.json is seeded as a byte-copy of the
# UTF-8 template, whose hallucination-phrase list is Cyrillic. Without it,
# Windows reads cp1252 and dies on bytes cp1252 leaves undefined (0x81…),
# which quarantined the freshly seeded config in an infinite crash loop
# (seen in the field on 26.1.146).
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
config = json.load(f)
print(f"[OK] Loaded configuration from '{CONFIG_FILE}'")
except OSError as e:
# A read failure (permissions, I/O) is NOT corruption: the file's content
# may be perfectly good, so displacing it and seeding defaults would
# silently discard the user's settings. Fail loudly instead.
raise RuntimeError(
f"Cannot read '{CONFIG_FILE}': {e}. If the file is owned by another "
f"user (e.g. root after running the server with sudo), fix it with: "
f"sudo chmod 644 {CONFIG_FILE}"
) from e
except Exception as e:
# Corrupted config.json (readable but unparseable): back up the bad file,
# then rewrite a fresh one from the template so the app can still start.
print(f"[CONFIG] ERROR: could not parse '{CONFIG_FILE}': {e}")
from datetime import datetime
import shutil
try:
corrupt_path = f"{CONFIG_FILE}.corrupt.{datetime.now().strftime('%Y%m%d_%H%M%S')}"
shutil.move(CONFIG_FILE, corrupt_path)
print(f"[CONFIG] Backed up corrupt config to '{corrupt_path}'")
except Exception as move_err:
print(f"[CONFIG] WARNING: could not back up corrupt config: {move_err}")
if not _restore_config_from_template("recover config.json"):
raise
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
config = json.load(f)
print(f"[CONFIG] Recovered '{CONFIG_FILE}' from config.default.json")
# Migrate old use_english_model format to new .en model names
migrated = False
for section in ["model", "file_transcription"]:
if section in config:
# Handle top-level model section
if section == "model" and "whisper" in config[section]:
whisper = config[section]["whisper"]
if whisper.get("use_english_model", False):
model = whisper.get("model", "base")
if model in ["tiny", "base", "small", "medium"] and not model.endswith(".en"):
whisper["model"] = f"{model}.en"
print(f"[MIGRATION] Converted model '{model}' to '{model}.en'")
migrated = True
# Remove the old flag
if "use_english_model" in whisper:
whisper.pop("use_english_model")
migrated = True
# Handle file_transcription model section
elif section == "file_transcription" and "model" in config[section]:
if "whisper" in config[section]["model"]:
whisper = config[section]["model"]["whisper"]
if whisper.get("use_english_model", False):
model = whisper.get("model", "base")
if model in ["tiny", "base", "small", "medium"] and not model.endswith(".en"):
whisper["model"] = f"{model}.en"
print(f"[MIGRATION] Converted file transcription model '{model}' to '{model}.en'")
migrated = True
# Remove the old flag
if "use_english_model" in whisper:
whisper.pop("use_english_model")
migrated = True
# Drop settings that were removed from the template (stt/config_utils.RETIRED_KEYS)
if drop_retired_keys(config):
print("[MIGRATION] Removed retired settings from config.json")
migrated = True
# Save migrated config
if migrated:
try:
with _config_file_lock:
_atomic_write_json(CONFIG_FILE, config)
print("[MIGRATION] Config file updated and saved")
except Exception as e:
print(f"[MIGRATION] Warning: Could not save migrated config: {e}")
# Patch in settings added since this config.json was created (e.g. a new
# auto_update block), so older installs pick up new defaults without manual
# edits. Only missing keys are added; user-set values are never overwritten.
try:
with open(CONFIG_TEMPLATE_FILE, "r", encoding="utf-8") as f:
template = json.load(f)
if _merge_missing_keys(config, template):
with _config_file_lock:
_atomic_write_json(CONFIG_FILE, config)
print("[CONFIG] Added new default settings missing from config.json")
except Exception as e:
print(f"[CONFIG] Warning: could not patch missing defaults: {e}")
return config
# Load configuration
config = load_config()
# Project default DSN (ingest-only key). Overridable via crash_reporting.sentry_dsn;
# disable everything with crash_reporting.sentry_enabled = false.
_SENTRY_DEFAULT_DSN = "https://eff01fdec5e9330b80ffd96093038588@o4511050918723584.ingest.us.sentry.io/4511714251702272"
from stt.crash_reports import scrub_event as _sentry_scrub_request # noqa: E402
from stt.crash_reports import sentry_environment as _sentry_environment # noqa: E402
def _init_sentry():
"""Sentry error reporting, logs, tracing, and profiling — on by default,
disabled via crash_reporting.sentry_enabled = false. Runs in the web
process and again in the transcription worker (which re-imports this
module). Never blocks boot.
The worker needs one extra thing to actually report: sentry_sdk hooks
uncaught exceptions through sys.excepthook, and multiprocessing never calls
it in a child. _report_worker_crash captures and flushes by hand, and
install_crash_diagnostics chains sys.excepthook rather than replacing it so
this init is not silently undone. Both fixed together — see
stt/worker_crash.py.
Deliberately configured to carry no user content: no request bodies, no
frame locals, no PII (IP addresses/headers). Scrubbing, and the dropping of
engineio's websocket-upgrade exceptions, live in stt/crash_reports.py."""
cr = config.get("crash_reporting", {})
if not cr.get("sentry_enabled", True):
return
dsn = (cr.get("sentry_dsn", "") or "").strip() or _SENTRY_DEFAULT_DSN
try:
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
try:
with open(os.path.join(BUNDLE_DIR, "VERSION"), encoding="utf-8") as _vf:
release = "stt@" + _vf.read().strip()
except OSError:
release = None
sentry_sdk.init(
dsn=dsn,
integrations=[FlaskIntegration()],
release=release,
# Left unset the SDK files everything under "production", which put a
# visitor's demo crash in the same environment as a running service.
environment=_sentry_environment(DEMO),
# PII off: no client IP addresses, headers or cookies. Opt-in via a
# new key, so installs carrying the old sentry_send_pii=true default
# in their config.json are healed rather than grandfathered.
send_default_pii=bool(cr.get("sentry_send_pii_optin", False)),
# Request bodies can hold transcript text (/api/translate),
# dictionary entries and file paths — never send them.
max_request_body_size="never",
# Frame locals routinely hold the text being transcribed/translated.
include_local_variables=False,
# Runs on errors and on sampled transactions alike.
before_send=_sentry_scrub_request,
before_send_transaction=_sentry_scrub_request,
# Left unset the SDK fills this with socket.gethostname(), which on
# a church PC is often the building or the operator's name.
server_name="stt",
# Forward logging-module records as Sentry Logs (watchdog uses
# logging heavily; the server mostly print()s, which is not captured)
enable_logs=bool(cr.get("sentry_enable_logs", True)),
# NOT 1.0: the web UI polls ~2x/second, so full tracing would be
# ~100k+ transactions per viewer per day
traces_sample_rate=float(cr.get("sentry_traces_sample_rate", 0.1)),
# Continuous profiling is OFF unless explicitly opted in: the SDK's
# profiler thread (continuous_profiler._sample_stack/extract_stack)
# repeatedly segfaulted the server while sampling other threads'
# stacks — every faulthandler dump named it as the crashing thread.
# Gated behind a new key (not the old sample-rate) so installs with
# the old 1.0 default materialized in config.json are healed too.
profile_session_sample_rate=(
float(cr.get("sentry_profile_session_sample_rate", 1.0))
if cr.get("sentry_profiling_enabled", False) else 0.0
),
profile_lifecycle=cr.get("sentry_profile_lifecycle", "trace"),
)
sentry_sdk.set_tag("process", "server")
# A crash in the demo is a packaging bug on a machine we will never see, which
# is the whole reason to report it — but it must never be counted against the
# real application. Tagged rather than filtered client-side so the demo's
# crashes stay searchable on their own.
sentry_sdk.set_tag("demo", "yes" if DEMO else "no")
_prof = "profiling" if cr.get("sentry_profiling_enabled", False) else "no profiling"
print(f"[SENTRY] Error reporting enabled (logs + traces + {_prof})"
+ (" [demo]" if DEMO else ""))
except ImportError:
print("[SENTRY] sentry-sdk is not installed — rerun the installer or: uv pip install 'sentry-sdk[flask]'")
except Exception as e:
print(f"[SENTRY] Init failed (continuing without): {e}")
_init_sentry()
# Create empty word highlighting file if it doesn't exist
if not os.path.exists(WORD_HIGHLIGHTING_FILE):
print(f"[INIT] Creating empty {WORD_HIGHLIGHTING_FILE}")
save_word_highlighting({"enabled": True, "words": []})
# Calibration mode state
calibration_mode = False
calibration_data = None
# Remote translation state
_pending_pair_requests = {} # {ip: {"code": str, "expires": float, "attempts": int}}
_pending_pair_lock = threading.Lock()
PAIR_MAX_PENDING = 20 # cap unauthenticated pending requests
PAIR_MAX_ATTEMPTS = 5 # wrong-code guesses before the pending code is voided
_translation_clients = {} # {ip: last_seen_timestamp}
_translation_clients_lock = threading.Lock()
_trusted_translation_clients = set(
config.get("live_translation", {}).get("trusted_clients", [])
)
_session_glossary_override = None # {"glossary": {...}} pushed by a paired Machine A for this session; None = use custom_dictionary.json
# What a paired Machine A has actually taken over on this machine, as opposed to
# what it could take over. A pairing on its own dictates nothing: A pushes a
# language only when it switches one, and a glossary only when one is edited.
# Reported by /api/translation/status so this machine's settings page locks the
# controls A really owns instead of every control it might — locking on the mere
# existence of a pairing shut the operator out of the LLM settings, which A never
# sends at all.
#
# The model is deliberately absent. An offload server runs its own engine and its
# own model, and A has no say in either: it asks for a translation and is told
# what produced it. One machine owning that choice is the whole point — two
# machines negotiating it was a picker on A listing B's files, a push route, a
# reload, and a lock group, all to arrive at a setting B already had.
_a_pushed = {"language": False, "glossary": False}
def _is_trusted_translation_client(ip):
"""Whether ``ip`` is on the paired list. Address only — see _paired_client_ok."""
return ip in _trusted_translation_clients
# A token this machine has just issued but whose owner may not have stored it:
# for this long, that address may ask again. Long enough for Machine A's ~20 s
# heartbeat to retry, short enough that it is not a standing way in. In memory
# only — a restart ends the window, which is the safe direction.
PAIR_TOKEN_GRACE_SECONDS = 120
_pair_token_grace = {} # {ip: expiry}
_pair_token_grace_lock = threading.Lock()
def _pair_token_grace_ips():
"""Addresses still inside the re-claim window, dropping the expired ones."""
now = time.time()
with _pair_token_grace_lock:
for ip in [ip for ip, expiry in _pair_token_grace.items() if expiry <= now]:
del _pair_token_grace[ip]
return set(_pair_token_grace)
def _pair_tokens():
"""Fingerprint -> address, for the tokens this machine has handed out."""
stored = config.get("live_translation", {}).get("trusted_client_tokens")
return dict(stored) if isinstance(stored, dict) else {}
def _save_pair_tokens(tokens):
"""Persist the token store."""
config.setdefault("live_translation", {})["trusted_client_tokens"] = dict(tokens)
save_config(config)
def _paired_client_ok(ip=None):
"""Whether the request in hand may act as the paired Machine A.
The bearer token pairing handed out is the real answer; a machine that has
not been given one yet is still recognised by address, which is what lets an
older build keep working until it updates (see stt/pair_tokens). A token
arriving from a new address moves the pairing to that address, so a machine
that changed IP does not have to be paired again.
"""
client_ip = request.remote_addr if ip is None else ip
tokens = _pair_tokens()
decision = _pair_tokens_mod.authorize(
tokens, _trusted_translation_clients, client_ip,
_pair_tokens_mod.parse_bearer(request.headers.get("Authorization")) if ip is None else None,
)
if decision.authorized and decision.rebind_from:
_rebind_trusted_client(decision.rebind_from, client_ip)
if not decision.authorized:
_log_pair_refusal(client_ip, decision.reason)
return decision.authorized
def _log_pair_refusal(client_ip, reason):
"""Say out loud why a would-be paired machine was turned away.
The reason never goes back over HTTP — telling an unauthenticated caller
which half of the check it failed is a gift to whoever is guessing. But
discarding it entirely meant B could refuse its own paired A on every call
and no log anywhere said so: the machine looked simply unreachable, and
finding the truth took a packet-level diagnosis across both boxes. So it is
written here, where only the operator can read it, and thinned through the
same suppressor as refused HTTP so a peer that retries cannot flood the log.
"""
try:
if _pair_refusal_log_filter.decide((client_ip, reason)).log:
print(f"[PAIR] Refused {client_ip}: {reason}")
except Exception:
pass # a diagnostic must never be the thing that breaks a request
def _rebind_trusted_client(old_ip, new_ip):
"""Follow a paired machine to a new address, keeping what we knew about it."""
if not new_ip or old_ip == new_ip:
return
_trusted_translation_clients.discard(old_ip)
_trusted_translation_clients.add(new_ip)
live = config.setdefault("live_translation", {})
trusted = live.setdefault("trusted_clients", [])
if old_ip in trusted:
trusted.remove(old_ip)
if new_ip not in trusted:
trusted.append(new_ip)
port = _translation_client_ports.get(old_ip)
_forget_client_port(old_ip)
if port:
_remember_client_port(new_ip, port, persist=False)
live["trusted_client_tokens"] = _pair_tokens_mod.rebind(_pair_tokens(), old_ip, new_ip)
save_config(config)
print(f"[PAIR] Paired machine moved from {old_ip} to {new_ip}")
def _add_trusted_client(ip, port=None):
_trusted_translation_clients.add(ip)
if "live_translation" not in config:
config["live_translation"] = {}
trusted = config["live_translation"].setdefault("trusted_clients", [])
if ip not in trusted:
trusted.append(ip)
if port:
_remember_client_port(ip, port, persist=False)
save_config(config)
def _forget_client_port(ip):
"""Drop a client's stored UI port when it is unpaired."""
_translation_client_ports.pop(ip, None)
stored = config.get("live_translation", {}).get("trusted_client_ports")
if isinstance(stored, dict) and stored.pop(ip, None) is not None:
save_config(config)
def _remember_client_port(ip, port, persist=True):
"""Record where a paired client's own UI lives, so we can link back to it.
Learned at pairing and refreshed by the heartbeat. Stored in its own config
map rather than folded into trusted_clients, which is a plain list of IPs
that older configs and _is_trusted_translation_client both rely on.
"""
port = coerce_int(port, 0, lo=1, hi=65535)
if not port:
return
_translation_client_ports[ip] = port
stored = config.setdefault("live_translation", {}).setdefault("trusted_client_ports", {})
if stored.get(ip) != port:
stored[ip] = port
if persist:
save_config(config)
# {ip: port} for paired clients that have told us where their own UI lives. Kept
# beside _translation_clients rather than in it, because that maps ip -> last-seen
# timestamp and several readers do arithmetic on the value. Seeded from config so
# a paired client can be linked to before it has run a session since our restart.
_translation_client_ports = {
str(ip): int(port)
for ip, port in (config.get("live_translation", {}).get("trusted_client_ports", {}) or {}).items()
if str(port).isdigit()
}
def _register_translation_client(ip, port=None):
with _translation_clients_lock:
_translation_clients[ip] = time.time()
if port:
_translation_client_ports[ip] = port
# Generate random password if not configured
password_auth_config = config.get("web_server", {}).get("password_auth", {})
if password_auth_config.get("enabled", False) and not password_auth_config.get("password", ""):
import secrets
import string
# Generate a random 12-character password
alphabet = string.ascii_letters + string.digits
random_password = ''.join(secrets.choice(alphabet) for i in range(12))
# Update config with generated password
if "web_server" not in config:
config["web_server"] = {}
if "password_auth" not in config["web_server"]:
config["web_server"]["password_auth"] = {}
config["web_server"]["password_auth"]["password"] = random_password
# Save config
save_config(config)
print("=" * 80)
print("[AUTH] Password authentication enabled with auto-generated password:")
print(f"[AUTH] Password: {random_password}")
print("[AUTH] Save this password to access settings from non-whitelisted IPs.")
print("[AUTH] You can change it in config.json under web_server.password_auth.password")
print("=" * 80)
# Timezone Helper Function
def get_configured_timezone():
"""The timezone transcript rows are stamped in.
Honours config["timezone"]; mode "auto" (the default) means the machine's own
zone, anything else uses "value" as an IANA name. A name this machine cannot
load falls back to the system zone with a warning rather than raising —
timestamps are written on every row, so a typo in a setting must not be able to
stop a service from recording.
"""
tz, note = _resolve_timezone(config.get("timezone"), _system_timezone())
if note:
print(f"[TIMEZONE] {note}")
return tz
# Load configured timezone. Read once at import: the value is stamped onto rows in
# the worker, which re-imports this module, so a change needs a restart to take
# effect on both sides — which is what the save endpoint tells the operator.
configured_timezone = get_configured_timezone()
print(f"[OK] Using timezone: {configured_timezone}")
# ====================================================================================
# NLLB Translation Support - Translate to 200+ languages
# ====================================================================================
# NLLB language tables + model catalog live in stt/nllb_catalog.py (importable, tested).
from stt.nllb_catalog import ( # noqa: F401
NLLB_LANG_CODES,
TRANSLATION_LANGUAGES,
get_nllb_model_description,
get_default_nllb_models,
build_madlad_input,
is_madlad_model,
get_default_madlad_models,
madlad_anti_repetition_defaults,
supported_target,
languages_for_method,
resolve_translation_model_id as _resolve_translation_model_id,
)
from stt.ct2_translate import ( # noqa: F401
resolve_compute_type as _ct2_resolve_compute_type,
ct2_model_dir as _ct2_model_dir,
nllb_ct2_target_prefix as _ct2_nllb_target_prefix,
strip_target_prefix as _ct2_strip_target_prefix,
nllb_source_tokens as _ct2_nllb_source_tokens,
madlad_source_tokens as _ct2_madlad_source_tokens,
decode_ct2_tokens as _ct2_decode_tokens,
score_to_confidence as _ct2_score_to_confidence,
)
# Session provenance: which models/decode settings produced a given transcript.
from stt.llm_translate import (
effective_context_window as _effective_context_window,
DEFAULT_SYSTEM_PROMPT_TEMPLATE as _DEFAULT_LLM_SYSTEM_PROMPT,
build_chat_messages as _llm_chat_messages,
build_chat_payload as _llm_chat_payload,
build_system_prompt as _llm_system_prompt,
check_translation as _llm_check,
retry_system_prompt as _llm_retry_prompt,
fit_context_prefix as _llm_fit_context,
input_fits as _llm_input_fits,
input_token_budget as _llm_input_budget,
local_model_path as _llm_local_model_path,
looks_like_reasoning_model as _llm_looks_like_reasoning,
looks_like_reasoning_name as _llm_reasoning_name,
resolve_gpu_layers as _llm_resolve_gpu_layers,
scan_gguf_models as _scan_gguf_models,
extract_chat_text as _llm_extract_text,
uses_local_llm as _uses_local_llm,
# Prompt styles: a chat model is instructed with a system prompt, a
# TranslateGemma with four fields and no instruction at all.
PROMPT_STYLE_TRANSLATEGEMMA as _LLM_STYLE_TRANSLATEGEMMA,
build_translategemma_messages as _llm_tg_messages,
build_translategemma_prompt as _llm_tg_prompt,
is_model_gguf as _is_model_gguf,
PROMPT_STYLES as _LLM_PROMPT_STYLES,
resolve_prompt_style as _llm_prompt_style,
# What a declined caption does: translate it with the NMT model, or show the
# source and keep those weights out of memory entirely.
FALLBACK_SKIP as _LLM_FALLBACK_SKIP,
resolve_fallback as _llm_resolve_fallback,
uses_system_prompt as _llm_uses_system_prompt,
)
# Service-phase detection lives in stt/service_phase.py (importable, unit-tested);
# the monolith supplies the connection and the live config and does nothing else.
from stt.service_phase import (
analyze as _service_phase_analyze,
delete_correction as _service_phase_delete_correction,
delete_correction_by_id as _service_phase_delete_correction_by_id,
ensure_tables as _service_phase_ensure_tables,
load_analysis as _service_phase_load,
load_corrections as _service_phase_corrections,
read_rows as _service_phase_rows,
save_analysis as _service_phase_save,
save_correction as _service_phase_save_correction,
save_group_correction as _service_phase_save_group,
)
from stt.optional_deps import RuntimeProbe as _RuntimeProbe
# The operator's live marks. They resolve into ordinary spans, which is why nothing
# downstream — the timeline, sermon_ranges, the summariser — has to learn what a mark is.
from stt.phase_marks import (
END_LABEL as _MARK_END,
describe as _phase_marks_describe,
is_mark as _phase_is_mark,
resolve as _phase_marks_resolve,
)
from stt.phase_rules import load_rules as _phase_rules_load
from stt.control_auth import refuse_reason as _control_refuse_reason
from stt.phase_profiles import (
DEFAULT_PROFILE as _phase_default_profile,
apply_to_profile as _phase_apply_to_profile,
list_profiles as _phase_list_profiles,
load_profile as _phase_load_profile,
merge_config as _phase_merge_config,
profile_path as _phase_profile_path,
seed_missing as _phase_seed_profile,
select_profile as _phase_select_profile,
slugify_profile as _phase_slugify_profile,
)
from stt.session_index import (
describe as _session_describe,
index as _session_index,
resolve_session as _session_resolve,
unreadable as _session_unreadable,
)
# Sermon summary and chapters live in stt/sermon_summary.py, beside the phase detector that
# tells it where the sermon was; the monolith supplies the connection, the config and the
# model call.
from stt.sermon_summary import (
MAP_SYSTEM as _SERMON_MAP_SYSTEM,
STATUS_DONE as _SERMON_DONE,
STATUS_ERROR as _SERMON_ERROR,
STATUS_PENDING as _SERMON_PENDING,
STATUS_RUNNING as _SERMON_RUNNING,
build_map_prompt as _sermon_map_prompt,
build_reduce_prompt as _sermon_reduce_prompt,
build_translate_prompt as _sermon_translate_prompt,
chunk_rows as _sermon_chunk_rows,
delete_summary as _sermon_delete,
ensure_tables as _sermon_ensure_tables,
explain_no_sermons as _sermon_explain,
fingerprint as _sermon_fingerprint,
format_offset as _sermon_format_offset,
has_summaries as _sermon_has_summaries,
load_summaries as _sermon_load_all,
load_summary as _sermon_load,
mark_error as _sermon_mark_error,
parse_chapters as _sermon_parse_chapters,
parse_sections as _sermon_parse_sections,
parse_translation as _sermon_parse_translation,
progress_text as _sermon_progress_text,
set_progress as _sermon_set_progress,
dominant_source_language as _sermon_source_language,
read_sermon_rows as _sermon_read_rows,
row_signature as _sermon_row_signature,
same_language as _sermon_same_language,
sermon_ranges as _sermon_ranges,
ready_sermons as _sermon_ready,
wait_while as _sermon_wait_while,
save_summary as _sermon_save,
supersede as _sermon_supersede,
snap_chapters as _sermon_snap_chapters,
chapter_range as _sermon_chapter_range,
transcript_text as _sermon_transcript_text,
unfinished as _sermon_unfinished,
ACTION_NOTE as _SERMON_ACTION_NOTE,
ACTION_SKIP as _SERMON_ACTION_SKIP,
covering_summary as _sermon_covering,
note_range_change as _sermon_note_range_change,
range_change as _sermon_range_change,
scan_action as _sermon_scan_action,
)
from stt.llm_priority import (
KIND_CAPTION as _ACT_CAPTION,
KIND_HEARTBEAT as _ACT_HEARTBEAT,
PeerActivity as _PeerActivity,
hold_poll_seconds as _hold_poll_seconds,
quiet_window as _quiet_window,
stale_window as _stale_window,
summariser_wait_reason as _summariser_wait_reason,
)
# Evidence that a paired machine has a service running, kept apart from
# _translation_clients on purpose: that map is refreshed by pairing traffic and by any
# route calling _register_translation_client, so a summariser reading it would see its own
# footprint and never find this machine quiet. Only real caption work and the transcription
# heartbeat land here — see stt/llm_priority.py.
_peer_activity = _PeerActivity()
from stt.sermon_scheduling import (
DEFAULT_PAUSE_CEILING_SECONDS as _SERMON_PAUSE_CEILING,
defer_scan as _sermon_defer_scan,
finalise_expired as _sermon_finalise_expired,
hold_reason as _sermon_hold_reason,
pause_fields as _sermon_pause_fields,
)
from stt.phase_learn import (
apply_proposals as _phase_learn_apply,
collect as _phase_learn_collect,
group_by_service as _phase_learn_shapes,
propose_all as _phase_learn_propose,
read_marked_phases as _phase_learn_marks,
)
from stt.db_maintenance import (
checkpoint_and_release as _db_checkpoint_and_release,
open_readonly as _db_open_readonly,
sweep_orphaned_sidecars as _db_sweep_sidecars,
_iter_databases as _db_iter_databases,
)
from stt.session_meta import (
append_changes as _session_meta_append,
append_new_changes as _session_meta_append_new,
build_session_meta as _build_session_meta,
changed_keys as _session_meta_changed_keys,
glossary_provenance as _glossary_provenance,
is_offloaded as _translation_is_offloaded,
uses_llm as _translation_uses_llm,
latest_values as _session_meta_latest,
load_session_meta as _load_session_meta,
remote_provenance as _remote_provenance,
write_missing as _session_meta_write_missing,
write_session_meta as _write_session_meta,
asr_row_label as _session_asr_row_label,
mt_row_label as _session_mt_row_label,
peer_provenance as _session_peer_provenance,
row_label_if_changed as _session_row_label_if_changed,
MT_ENGINE_LLM,
MT_ENGINE_NMT,
MT_ENGINE_NONE,
MT_ENGINE_REMOTE,
MT_ENGINE_WHISPER,
)
# Default MADLAD-400 model when the translation engine is set to "madlad".
MADLAD_DEFAULT_MODEL = "google/madlad400-3b-mt"
def _local_fallback_ready():
"""Whether this machine could translate locally if the remote went away.
Answers the question the "fall back to local translation" setting quietly
assumes: is a local model actually on disk. Whisper-translate modes need no
separate model — the ASR pass produces the translation — and an LLM session
is ready when its GGUF is there. Checked on disk rather than by loading,
because this runs on a settings page, not on the caption path.
"""
lt = config.get("live_translation", {})
method = (lt.get("translation_method") or "nllb").strip().lower()
if method in ("whisper_translate", "whisper_forced_lang"):
return True
if method == "llm":
llm_cfg = lt.get("llm") or {}
if (llm_cfg.get("provider") or "endpoint").strip().lower() != "local":
return bool((llm_cfg.get("endpoint") or "").strip())
path = (llm_cfg.get("gguf_path") or "").strip() or _llm_local_model_path(
MODELS_DIR, (llm_cfg.get("gguf_repo") or "").strip(),
(llm_cfg.get("gguf_file") or "").strip())
return bool(path and os.path.isfile(path))
model_id = _resolve_live_translation_model_id(lt)
if not model_id:
return False
# A CTranslate2 conversion counts: it is what actually runs, and the HF
# weights are often deleted afterwards to reclaim the space.
return model_presence(MODELS_DIR, model_id.replace("/", "--")).downloaded
def _resolve_live_translation_model_id(lt_cfg):
"""Effective live-translation model id for the configured engine.
Thin wrapper over stt.nllb_catalog.resolve_translation_model_id so session
provenance (stt/session_meta.py) records the model that actually loads
rather than a possibly-stale configured string, with no chance of the two
resolutions drifting apart.
"""
return _resolve_translation_model_id(lt_cfg, MADLAD_DEFAULT_MODEL)
def _maybe_half_translation_model(model, device, use_fp16):
"""Convert the model to fp16 on GPU when requested (CUDA/MPS only — see
_should_use_fp16). Returns (model, applied). Never raises: on failure it
keeps the fp32 model so a bad fp16 path degrades instead of failing the load."""
if not _should_use_fp16(use_fp16, device):
return model, False
try:
return model.half(), True
except Exception as e:
print(f"[LIVE-TRANSLATION] fp16 conversion failed, staying fp32: {e}")
return model, False
def _load_ct2_translator(hf_model_path, model_id, use_gpu, ct2_compute_type):
"""Load (converting once, cached) a CTranslate2 Translator for the model at
hf_model_path. CT2 has no Metal backend, so Apple Silicon runs CPU int8 —
which is the whole point (MADLAD-3B in ~3 GB). Sets _live_translation_device.
Raises if the HF model isn't downloaded locally (conversion needs the weights)."""
global _live_translation_device
import ctranslate2
device = "cuda" if (use_gpu and torch.cuda.is_available()) else "cpu"
# The card's capability decides whether fp16 is worth naming: a pre-Volta GPU
# has no fast fp16 path, and int8_float16 there only earns a second cache
# directory holding the same weights.
_capability = None
if device == "cuda":
try:
_props = torch.cuda.get_device_properties(0)
_capability = float(f"{_props.major}.{_props.minor}")
except Exception:
_capability = None # unknown keeps the long-standing answer
compute_type = _ct2_resolve_compute_type(ct2_compute_type or "auto", device, _capability)
if not os.path.isdir(hf_model_path):
raise RuntimeError(
f"CTranslate2 backend needs '{model_id}' downloaded locally first "
"(Model Manager). Cannot convert a hub-only model.")
ct2_dir = _ct2_model_dir(hf_model_path, compute_type)
# A conversion that was interrupted — the process killed, the machine
# restarted, the disk filled — left a directory that exists and is not a
# model. The only test was that the path was there, so the remains were then
# handed to ctranslate2.Translator, which is a C++ reader: a crash rather
# than an exception, and nothing to repair it since the path existed.
if not _model_files.dir_status(ct2_dir, "ct2").complete:
if os.path.isdir(ct2_dir):
print(f"[CT2] Previous conversion at {ct2_dir} is incomplete; redoing it",
flush=True)
shutil.rmtree(ct2_dir, ignore_errors=True)
# One-time conversion from the official HF weights (transiently needs
# ~model-size RAM). Cached beside the HF model, keyed by compute type.
#
# Built under a temporary name and moved into place, so an interruption
# leaves no directory rather than half a model — the same rule the file
# downloads follow with their .part staging.
print(f"[CT2] Converting {model_id} -> {ct2_dir} ({compute_type})... one-time", flush=True)
from ctranslate2.converters import TransformersConverter
staging = f"{ct2_dir}.converting"
shutil.rmtree(staging, ignore_errors=True)
TransformersConverter(hf_model_path).convert(staging, quantization=compute_type, force=False)
os.replace(staging, ct2_dir)
_lt = config.get("live_translation", {})
intra = max(0, int(_lt.get("ct2_intra_threads", 4))) # CPU compute threads (P-cores)
inter = max(1, int(_lt.get("ct2_inter_threads", 1))) # parallel batches
translator = ctranslate2.Translator(ct2_dir, device=device, compute_type=compute_type,
intra_threads=intra, inter_threads=inter)
_live_translation_device = device
print(f"[CT2] Translator loaded ({device}, {compute_type}, intra={intra} inter={inter})")
return translator
def load_translation_model(use_gpu=True, model_id=None, use_fp16=False, use_ct2=False, ct2_compute_type="auto"):
"""
Load a translation model (NLLB-200 or MADLAD-400).
Args:
use_gpu: Whether to use GPU acceleration
model_id: HuggingFace model ID (e.g., "facebook/nllb-200-distilled-600M")
If None, defaults to facebook/nllb-200-distilled-600M
use_fp16: Load in half precision on GPU (MPS/CUDA only; ignored on CPU)
use_ct2: Use the CTranslate2 int8 backend (returns a Translator instead
of a transformers model; converts once from local HF weights)
ct2_compute_type: CT2 quantization ("auto" -> int8 on CPU, int8_float16 on CUDA)
Returns:
Tuple of (model, tokenizer) — model is a transformers model or, with
use_ct2, a ctranslate2.Translator.
"""
_lazy_import_ml_libraries()
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
# Use provided model_id or default
if model_id is None:
model_id = "facebook/nllb-200-distilled-600M"
# Check ./models/ directory first for local copy
local_dir_name = model_id.replace("/", "--")
local_model_path = os.path.join(MODELS_DIR, local_dir_name)
if os.path.exists(local_model_path):
model_path = local_model_path
print(f"[INFO] Loading translation model from local: {model_path}")
# Refuse a directory whose files no longer match what was downloaded.
# An interrupted transfer used to leave a truncated pytorch_model.bin
# under its real name; the failure surfaced minutes or days later as
# "PytorchStreamReader failed reading zip archive: failed finding central
# directory" from torch.load, on whichever thread happened to touch
# translation next — a message with nothing in it about downloads.
# Reported from the settings-save thread in the wild, which is why the
# check belongs here rather than in any start-up precondition.
#
# Not on the CT2 path. There the weights are read from the conversion and
# this directory supplies only the tokenizer, so an operator who reclaimed
# the HF weights after converting — which the Model Manager treats as
# downloaded, and which is the documented way to save several gigabytes —
# would have a working setup refused for missing a file nothing reads.
if not use_ct2:
# No weights at all: checked first because transformers' own error
# ("no file named model.safetensors, or pytorch_model.bin") names no
# remedy. Usually weights reclaimed after a CT2 conversion. STT-2S.
_no_weights = _model_disk.missing_weights_message(
model_presence(MODELS_DIR, local_dir_name), model_id)
if _no_weights:
raise _model_disk.ModelUnavailable(_no_weights)
_incomplete = _model_files.manifest_mismatches(model_path)
if _incomplete:
raise _model_disk.ModelUnavailable(
f"The translation model '{model_id}' is incomplete on disk — "
f"{_model_files.describe_missing(_incomplete)} "
f"{'are' if len(_incomplete) > 1 else 'is'} missing or truncated. "
f"Download it again in the Model Manager (Settings -> Model Manager); "
f"the download now repairs what is already there."
)
else:
model_path = model_id
print(f"[INFO] Loading translation model from HuggingFace: {model_path}")
tokenizer = AutoTokenizer.from_pretrained(model_path)
global _live_translation_is_ct2
if use_ct2:
translator = _load_ct2_translator(model_path, model_id, use_gpu, ct2_compute_type)
_live_translation_is_ct2 = True
return translator, tokenizer
_live_translation_is_ct2 = False
bin_path = os.path.join(model_path, "pytorch_model.bin") if os.path.isdir(model_path) else None
if bin_path and os.path.exists(bin_path) and not os.path.exists(os.path.join(model_path, "model.safetensors")):
from transformers import AutoConfig
cfg = AutoConfig.from_pretrained(model_path)
model = AutoModelForSeq2SeqLM.from_config(cfg)
state_dict = torch.load(bin_path, map_location="cpu", weights_only=False)
model.load_state_dict(state_dict)
else:
# Load straight to fp16 when the model is headed for a GPU and fp16 was
# asked for, instead of loading fp32 and halving afterwards. That order
# holds both copies at once: MADLAD-3B is ~11.8 GB of fp32 weights, so the
# peak is ~17.7 GB — more than a 16 GB unified-memory Mac has, which put
# MPS out of reach for that model even though its fp16 weights are ~5.9 GB.
# low_cpu_mem_usage streams the checkpoint rather than building a second
# full copy in RAM first.
_load_dtype = _translation_load_dtype(
use_fp16, use_gpu, torch.cuda.is_available(),
bool(getattr(torch.backends, "mps", None) and torch.backends.mps.is_available()),
)
_load_kwargs = {}
if _load_dtype:
_load_kwargs = {"torch_dtype": getattr(torch, _load_dtype), "low_cpu_mem_usage": True}
model = AutoModelForSeq2SeqLM.from_pretrained(model_path, **_load_kwargs)
global _live_translation_device
# _maybe_half_translation_model stays as the fallback: it is a no-op when the
# weights already loaded as fp16, and still does the conversion for the
# state-dict branch above (which cannot take a load dtype).
if use_gpu and torch.cuda.is_available():
model = model.to("cuda")
_live_translation_device = "cuda"
model, _fp16 = _maybe_half_translation_model(model, "cuda", use_fp16)
print(f"[INFO] Translation model loaded on GPU (CUDA) (fp16={_fp16})")
elif use_gpu and hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
model = model.to("mps")
_live_translation_device = "mps"
model, _fp16 = _maybe_half_translation_model(model, "mps", use_fp16)
print(f"[INFO] Translation model loaded on GPU (MPS) (fp16={_fp16})")
else:
_live_translation_device = "cpu"
if use_fp16:
print("[LIVE-TRANSLATION] use_fp16 ignored on CPU")
print("[INFO] Translation model loaded on CPU")
if use_gpu:
# GPU was requested but no accelerator was usable — translations will
# be seconds-per-sentence instead of sub-second. This is the silent
# failure mode behind "translations got slower" field reports (e.g. a
# torch upgrade that dropped CUDA support for the installed driver).
print(f"[WARNING] Translation model '{model_id}' requested GPU but is running on CPU — "
"expect slow translations. Check torch CUDA/MPS availability "
"(.venv python -c 'import torch; print(torch.cuda.is_available())').", flush=True)
return model, tokenizer
# Glossary application and the TranslationCache live in stt/translation_utils.py
# (importable, unit-tested); the wrapper below resolves config/session state.
from stt.translation_utils import (
TranslationCache,
TextTranslationCache,
apply_glossary as _apply_glossary_dict,
should_cache_translation as _should_cache_translation,
should_use_fp16 as _should_use_fp16,
translation_load_dtype as _translation_load_dtype,
)
from stt.tts_queue import SpokenTracker
from stt.translation_backfill import BackfillAttempts, select_backfill_ids
from stt.offload_jobs import OffloadJobs as _OffloadJobs
from stt.peer_load import (
FAILURE_DOWN as _PEER_DOWN,
FAILURE_TIMEOUT as _PEER_TIMEOUT,
LOAD_KEY as _PEER_LOAD_KEY,
WorkTracker as _WorkTracker,
classify_failure as _classify_peer_failure,
parse_load as _parse_peer_load,
peer_is_working as _peer_is_working,
)
from stt.translation_attempts import (
BACKFILL_SUCCESS_SET as _MT_BACKFILL_SUCCESS_SET,
LIVE_SUCCESS_SET as _MT_LIVE_SUCCESS_SET,
LiveTranslationAttempts as _LiveTranslationAttempts,
missing_attempt_columns as _missing_attempt_columns,
persist_decision as _persist_decision,
record_failed_attempt as _record_failed_attempt,
)
def _apply_glossary(text, source_lang, target_lang):
"""Apply NLLB glossary post-processing replacements from custom dictionary."""
try:
dict_config = config.get("custom_dictionary", {})
override = _session_glossary_override
if override is not None:
# Offloaded session: honor the CLIENT's glossary-enabled flag when it
# sent one, else fall back to this machine's own config.
enabled = override.get("nllb_glossary_enabled")
if enabled is None:
enabled = dict_config.get("nllb_glossary_enabled", False)
if not enabled:
return text
dictionary = override
elif not dict_config.get("nllb_glossary_enabled", False):
return text
else:
dict_file = dict_config.get("file", "custom_dictionary.json")
if not os.path.isabs(dict_file):
dict_file = os.path.join(CONFIG_DIR, dict_file)
if not os.path.exists(dict_file):
return text
import json as _json
with open(dict_file, "r", encoding="utf-8") as f:
dictionary = _json.load(f)
return _apply_glossary_dict(text, source_lang, target_lang, dictionary)
except Exception as e:
print(f"[WARNING] Glossary application failed: {e}")
return text
def _session_meta_enabled(session_config=None):
"""Whether provenance recording is on (config/config.default.json: session_meta)."""
return bool((session_config or config).get("session_meta", {}).get("enabled", True))
def _current_session_meta(session_config=None):
"""Provenance mapping for the running config, or {} when disabled.
The ASR model loads at init step 3 and the database is created at step 4, so
what actually loaded is already known here and belongs in the session-start
values rather than in the change log. The translation model loads lazily
(usually after the db exists), so its effective values are appended later by
_record_session_meta_change().
``session_config`` describes a config other than this process's global one.
The transcription worker outlives a Start/Stop cycle and is reused, so its
module-level `config` dates from when the process spawned — it reloads a
fresh copy into process_config at each session start (see thread1_function)
and must pass it here. Reading the global instead recorded a session as
translating locally with MADLAD while it was in fact offloading every caption
to a paired remote, which is precisely the misattribution this table exists
to prevent.
"""
if not _session_meta_enabled(session_config):
return {}
meta = _build_session_meta(
session_config or config, SERVER_VERSION, SERVER_COMMIT, SERVER_DESCRIBE,
socket.gethostname(), MADLAD_DEFAULT_MODEL,
)
try:
if transcription_state is not None:
loaded = transcription_state.get("loaded_model")
if loaded:
meta["asr.effective.model"] = str(loaded)
device = transcription_state.get("loaded_model_device")
if device:
meta["asr.effective.device"] = str(device)
except Exception:
pass # provenance is best-effort; a missing state proxy is not an error
try:
# The glossary terms rewrite translated captions and live in a file, not in
# config, so build_session_meta() cannot see them. Absent glossary.file
# means this session applies no glossary at all (a local LLM session), and
# there is nothing to describe. An override is a paired client's own table,
# pushed for this session: the local file is then not what ran.
if "glossary.file" in meta:
override = _session_glossary_override
meta.update(_glossary_provenance(
override if override is not None else load_custom_dictionary(),
"paired-client" if override is not None else "local"))
except Exception as e:
print(f"[SESSION-META] WARNING: could not read glossary provenance: {e}")
return meta
# Last successful answer from the paired machine about what it translates with.
# Kept because every offloaded row records the model that produced it, and probing
# per caption would put a network round trip in front of every one.
_remote_effective = {}
def _fetch_remote_provenance():
"""Ask the paired remote what it is translating with. {} if unreachable.
On an offloaded session the remote's model does the work, so without this the
database records nothing about what actually translated — and with a blank
remote model ("use Machine B's own model") this box can't infer it either.
"""
endpoint = _get_remote_endpoint_safe()
if not endpoint:
return {}
try:
r = _peer_request("GET", endpoint, "/api/translation/status", timeout=5)
provenance = _remote_provenance(r.json())
except Exception as e:
print(f"[SESSION-META] Could not read remote translation provenance: {e}")
return {}
if provenance:
_remote_effective.update(provenance)
return provenance
def _record_remote_provenance_async(db_path):
"""Probe the paired remote off-thread and record it as a session-start fact.
Off-thread because a 5s network timeout must not delay the start of
transcription; recorded with write_missing so it lands as a base key rather
than a timestamped change — nothing changed, we only just found out.
"""
if not (db_path and _session_meta_enabled()):
return
def _probe():
remote = _fetch_remote_provenance()
if remote and _session_meta_write_missing(db_path, remote):
print(f"[DB] OK: Recorded remote translation provenance ({remote.get('mt.remote.effective.model', '?')})")
threading.Thread(target=_probe, daemon=True).start()
def _reprobe_remote_provenance_async():
"""Re-probe the remote after an offload change and append what differs.
A change here means a different box (or a different model on the same box) is
now translating, so this genuinely is a mid-session change and belongs in the
timeline rather than as a base key.
"""
if not _session_meta_enabled():
return
active_db = transcription_state.get("db_name") if transcription_state else None
if not active_db:
return
def _probe():
remote = _fetch_remote_provenance()
if remote:
# Guarded append: this fires on every offload-touching save, and the
# remote usually answers with what it answered last time. It used to
# diff against the raw table, whose base key still holds the
# session-start value, so an unchanged remote was re-recorded under a
# fresh timestamp on every save.
_session_meta_append_new(active_db, remote)
threading.Thread(target=_probe, daemon=True).start()
def _record_session_meta_change(**values):
"""Append a mid-session change to the active session db.
For runtime facts that aren't in config — which model actually loaded, on
which device. Config-driven settings are handled wholesale by
_sync_session_meta_from_config() and must not be recorded here too.
Append-only: the base key keeps meaning "at session start", so a session
that began in one configuration and changed reads differently from one that
started in the final configuration. Never raises — a failed provenance write
must not break a live language switch.
Records nothing when the value already stands: the translation model reloads
for reasons other than a change of model (an unload to free VRAM, a settings
save that rebuilds it), and each reload used to append another identical
mt.effective.model row.
"""
if not _session_meta_enabled():
return
active_db = transcription_state.get("db_name") if transcription_state else None
if not active_db or not values:
return
_session_meta_append_new(
active_db, {k: "" if v is None else str(v) for k, v in values.items()})
# Regenerated on every call, so it can never be compared for equality.
_SESSION_META_VOLATILE = frozenset({"session.started_at"})
def _sync_session_meta_from_config():
"""Append any provenance-relevant setting that just changed in config.
Hooked into save_config (and /api/config's direct write) rather than into
each settings route. There are ~30 config writers, and hooking them one at a
time is exactly how the transcription-language switch and the three
hallucination-filter toggles went unrecorded — a session could flip the ASR
language mid-service while asr.language still read as the starting value,
contradicting the per-row source_language column. One choke point covers
every present and future path by construction.
Diffs against the latest recorded value (base key as superseded by any
appended change), so a setting that already changed isn't re-appended on
every subsequent save. Settings that appear for the first time become base
keys — they were never anything else — while genuine changes append.
"""
if not _session_meta_enabled():
return
active_db = transcription_state.get("db_name") if transcription_state else None
if not active_db:
return
try:
stored, read_error = _load_session_meta(active_db)
if read_error or not stored:
return # no session provenance yet: nothing to diff against
latest = _session_meta_latest(stored)
current = _current_session_meta()
changes, additions = {}, {}
for key, value in current.items():
if key in _SESSION_META_VOLATILE:
continue
if key not in latest:
additions[key] = value
elif latest[key] != value:
changes[key] = value
if additions:
_session_meta_write_missing(active_db, additions)
if changes:
_session_meta_append(active_db, changes)
except Exception as e:
print(f"[SESSION-META] Could not sync settings change: {e}")
def _translate_text_ct2(text, source_lang, target_lang, translator, tokenizer,
return_confidence=False, num_alternatives=0, generation_params=None):
"""Translate one string via the CTranslate2 int8 backend.
CT2 works on token strings: encode with the HF tokenizer, translate_batch,
decode. NLLB forces the target via target_prefix (stripped on decode);
MADLAD carries it as a "<2xx>" prefix in the source text. Mirrors the
transformers path's return shape (text, or dict with confidence/alternatives).
"""
_is_madlad = is_madlad_model(_live_translation_model_id or "")
gp = generation_params or {}
num_beams = max(1, int(gp.get("num_beams", 2)))
no_repeat = int(gp.get("no_repeat_ngram_size", 0))
rep_pen = float(gp.get("repetition_penalty", 1.0))
if _is_madlad:
# MADLAD can loop; apply safe anti-repetition defaults unless tuned.
rep_pen, no_repeat = madlad_anti_repetition_defaults(rep_pen, no_repeat)
want_extras = return_confidence or num_alternatives > 0
num_hyp = min(num_alternatives + 1, 5) if num_alternatives > 0 else 1
beam_size = max(num_beams, num_hyp)
tgt_code = None
if _is_madlad:
source = _ct2_madlad_source_tokens(tokenizer, text, target_lang)
target_prefix = None
else:
src_code = NLLB_LANG_CODES.get(source_lang, "eng_Latn")
tgt_code = NLLB_LANG_CODES.get(target_lang, "eng_Latn")
source = _ct2_nllb_source_tokens(tokenizer, text, src_code)
target_prefix = _ct2_nllb_target_prefix(tgt_code)
_kwargs = {"beam_size": beam_size, "num_hypotheses": num_hyp,
"return_scores": want_extras, "max_decoding_length": 1024}
if target_prefix is not None:
_kwargs["target_prefix"] = target_prefix
if no_repeat > 0:
_kwargs["no_repeat_ngram_size"] = no_repeat
if rep_pen != 1.0:
_kwargs["repetition_penalty"] = rep_pen
_tr_t0 = time.perf_counter()
results = translator.translate_batch([source], **_kwargs)
try:
_record_local_translate_ms((time.perf_counter() - _tr_t0) * 1000.0)
except Exception:
pass
res = results[0]
hyps = res.hypotheses
scores = getattr(res, "scores", None) or []
def _decode(tok_list):
toks = _ct2_strip_target_prefix(tok_list, tgt_code) if tgt_code else tok_list
return _ct2_decode_tokens(tokenizer, toks)
if not hyps:
return {"text": text, "confidence": None, "alternatives": []} if want_extras else text
best = _apply_glossary(_decode(hyps[0]), source_lang, target_lang)
if not want_extras:
return best
confidence = _ct2_score_to_confidence(scores[0]) if scores else None
alternatives = [_apply_glossary(_decode(h), source_lang, target_lang) for h in hyps[1:]]
return {"text": best, "confidence": confidence, "alternatives": alternatives}
def translate_text(text, source_lang, target_lang, model, tokenizer, return_confidence=False,
num_alternatives=0, generation_params=None, model_id=None, is_ct2=None):
"""
Translate text using NLLB-200
Args:
text: Text to translate
source_lang: Source language ISO code (e.g., 'en', 'fr')
target_lang: Target language ISO code
model: Loaded NLLB model
tokenizer: Loaded NLLB tokenizer
return_confidence: If True, return (text, confidence) tuple
num_alternatives: Number of alternative translations to return (0 = none)
generation_params: Dict of generation parameters (num_beams, length_penalty, etc.)
Returns:
Translated text string, or dict with text/confidence/alternatives if extras requested
"""
if not text or not text.strip():
if return_confidence or num_alternatives > 0:
return {"text": text, "confidence": None, "alternatives": []}
return text
# Which engine this call is for. model/tokenizer are parameters, but these two
# facts were read from the globals describing the LIVE model — so a caller that
# loaded its own model (batch file transcription does) got the live model's
# engine applied to a different model: a MADLAD file model tokenized as NLLB
# receives no target-language prefix at all, and a transformers model would be
# sent down the CTranslate2 path. Callers with their own model pass these.
_ct2 = _live_translation_is_ct2 if is_ct2 is None else is_ct2
_model_id = _live_translation_model_id if model_id is None else model_id
# CTranslate2 backend uses a different (token-string) API — route to it and
# leave the transformers path below untouched.
if _ct2:
return _translate_text_ct2(text, source_lang, target_lang, model, tokenizer,
return_confidence, num_alternatives, generation_params)
# MADLAD encodes the target as a "<2xx>" prefix on the input (no source
# language, no forced_bos); NLLB sets tokenizer.src_lang + a forced target
# token. Everything after tokenization is shared.
_is_madlad = is_madlad_model(_model_id or "")
if _is_madlad:
inputs = tokenizer(build_madlad_input(text, target_lang),
return_tensors="pt", padding=True, truncation=True, max_length=1024)
else:
# Convert ISO codes to NLLB codes and set the source language
tokenizer.src_lang = NLLB_LANG_CODES.get(source_lang, "eng_Latn")
inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True, max_length=1024)
# Move inputs to same device as model (CUDA, MPS, or CPU)
_device = next(model.parameters()).device
if _device.type != "cpu":
inputs = {k: v.to(_device) for k, v in inputs.items()}
# Merge user generation params with defaults
gp = generation_params or {}
num_beams = gp.get("num_beams", 2)
length_penalty = gp.get("length_penalty", 1.0)
no_repeat_ngram_size = gp.get("no_repeat_ngram_size", 0)
repetition_penalty = gp.get("repetition_penalty", 1.0)
if _is_madlad:
# MADLAD can loop; apply safe anti-repetition defaults unless tuned.
repetition_penalty, no_repeat_ngram_size = madlad_anti_repetition_defaults(
repetition_penalty, no_repeat_ngram_size)
generate_kwargs = {
"max_length": 1024,
"num_beams": num_beams,
"length_penalty": length_penalty,
"early_stopping": True,
}
if not _is_madlad:
# NLLB forces the target-language BOS token — validate it is known.
tgt_nllb = NLLB_LANG_CODES.get(target_lang, "eng_Latn")
forced_bos_id = tokenizer.convert_tokens_to_ids(tgt_nllb)
if forced_bos_id == tokenizer.unk_token_id:
print(f"[LIVE-TRANSLATION WARNING] Unknown target language token: {tgt_nllb} for lang={target_lang}, falling back to eng_Latn")
forced_bos_id = tokenizer.convert_tokens_to_ids("eng_Latn")
generate_kwargs["forced_bos_token_id"] = forced_bos_id
# Only add these if non-default to avoid warnings
if no_repeat_ngram_size > 0:
generate_kwargs["no_repeat_ngram_size"] = no_repeat_ngram_size
if repetition_penalty != 1.0:
generate_kwargs["repetition_penalty"] = repetition_penalty
# Enable confidence scoring and/or alternatives
if return_confidence or num_alternatives > 0:
generate_kwargs["return_dict_in_generate"] = True
generate_kwargs["output_scores"] = True
if num_alternatives > 0:
generate_kwargs["num_return_sequences"] = min(num_alternatives + 1, 5)
generate_kwargs["num_beams"] = max(5, num_alternatives + 1)
# Generate translation
_tr_t0 = time.perf_counter()
translated = model.generate(**inputs, **generate_kwargs)
try:
_record_local_translate_ms((time.perf_counter() - _tr_t0) * 1000.0)
except Exception:
pass
if return_confidence or num_alternatives > 0:
# Extract sequences and scores
sequences = translated.sequences
all_decoded = tokenizer.batch_decode(sequences, skip_special_tokens=True)
# Compute confidence from sequence scores if available
confidence = None
if hasattr(translated, 'sequences_scores') and translated.sequences_scores is not None:
import torch
confidence = float(torch.exp(translated.sequences_scores[0]).item())
best_text = _apply_glossary(all_decoded[0], source_lang, target_lang)
alternatives = [_apply_glossary(t, source_lang, target_lang) for t in all_decoded[1:]] if len(all_decoded) > 1 else []
return {"text": best_text, "confidence": confidence, "alternatives": alternatives}
# Simple mode - decode and return
result_text = tokenizer.batch_decode(translated, skip_special_tokens=True)[0]
return _apply_glossary(result_text, source_lang, target_lang)
def translate_segments(segments, source_lang, target_lang, model, tokenizer, progress_callback=None, generation_params=None, context_window=1, model_id=None, is_ct2=None):
"""
Translate a list of transcription segments
Args:
segments: List of segment dicts with 'text', 'start', 'end' keys
source_lang: Source language ISO code
target_lang: Target language ISO code
model: Loaded NLLB model
tokenizer: Loaded NLLB tokenizer
progress_callback: Optional callback function(percent, status) for progress updates
generation_params: Dict of generation parameters (num_beams, length_penalty, etc.)
context_window: Number of segments to combine for context (1 = no batching)
Returns:
List of translated segment dicts with same structure
"""
translated_segments = []
total = len(segments)
context_window = max(1, context_window)
for i, seg in enumerate(segments):
translated_text = None
if context_window > 1 and i > 0:
# Translate (context + target) in one call, then extract the target's
# portion by sentence-count alignment
start_idx = max(0, i - (context_window - 1))
context_text = " ".join(segments[j]["text"] for j in range(start_idx, i)).strip()
if context_text:
num_ctx_sentences = count_sentence_units(context_text)
combined_source = context_text + " " + seg["text"]
ctx_char_ratio = (len(context_text) + 1) / max(1, len(combined_source))
combined_translated = translate_text(combined_source, source_lang, target_lang, model, tokenizer, generation_params=generation_params, model_id=model_id, is_ct2=is_ct2)
translated_text = extract_context_translation(combined_translated, num_ctx_sentences, ctx_char_ratio)
if not translated_text:
# No context, or alignment failed - translate the segment alone
translated_text = translate_text(seg["text"], source_lang, target_lang, model, tokenizer, generation_params=generation_params, model_id=model_id, is_ct2=is_ct2)
translated_segments.append({
"text": translated_text,
"start": seg["start"],
"end": seg["end"]
})
# Report progress
if progress_callback and total > 0:
percent = int(70 + (i + 1) / total * 20) # 70-90% range
progress_callback(percent, f"Translating... ({i + 1}/{total} segments)")
return translated_segments
def _empty_device_cache():
"""Release cached accelerator memory back to the OS (CUDA and MPS).
Call AFTER gc.collect(): the allocators can only release blocks whose
tensors have been collected. On MPS this is what actually returns model
memory to macOS — without it the process keeps the freed memory reserved
and a reload allocates a second full copy (swap death on ARM Macs)."""
if torch is None:
return
if torch.cuda.is_available():
torch.cuda.empty_cache()
_mps = getattr(torch, "mps", None)
if _mps is not None and hasattr(_mps, "empty_cache") and torch.backends.mps.is_available():
try:
_mps.empty_cache()
except Exception:
pass
def _preload_cudnn(use_gpu):
"""Make cuDNN 9 loadable before faster_whisper/ctranslate2 is imported.
ctranslate2 >= 4.5 dlopens cuDNN 9 itself and knows nothing about the
venv's nvidia-cudnn wheel or torch's bundled copy, so resolve the lib
directory from the running interpreter (works for any Python version/venv)
instead of trusting the loader search path. No-op on macOS or CPU-only."""
if not use_gpu or platform == "darwin":
return
import ctypes
import importlib.util
lib_dirs = []
for mod in ("nvidia.cudnn", "torch"):
try:
spec = importlib.util.find_spec(mod)
except (ImportError, AttributeError, ValueError):
spec = None
for loc in (spec.submodule_search_locations or []) if spec else []:
lib_dirs.append(os.path.join(loc, "lib"))
if platform.startswith("win"):
lib_dirs.append(os.path.join(loc, "bin"))
if platform.startswith("win"):
# cuDNN DLLs live in the wheel's bin/ (nvidia-cudnn) or torch/lib;
# register them explicitly instead of relying on PATH.
for d in lib_dirs:
if os.path.isdir(d):
try:
os.add_dll_directory(d)
except OSError:
pass
return
cudnn_libs = [
"libcudnn_ops.so.9",
"libcudnn_cnn.so.9",
"libcudnn_adv.so.9",
"libcudnn_graph.so.9",
"libcudnn_engines_precompiled.so.9",
"libcudnn_engines_runtime_compiled.so.9",
"libcudnn_heuristic.so.9",
"libcudnn.so.9",
]
lib_dir = next((d for d in lib_dirs
if os.path.exists(os.path.join(d, "libcudnn.so.9"))), None)
if not lib_dir:
return
for lib in cudnn_libs:
lib_full_path = os.path.join(lib_dir, lib)
if os.path.exists(lib_full_path):
try:
ctypes.CDLL(lib_full_path, mode=ctypes.RTLD_GLOBAL)
except OSError as e:
print(f"Warning: Could not preload {lib}: {e}")
def cleanup_translation_model(model, tokenizer):
"""Clean up translation model to free memory"""
import gc
del model
del tokenizer
gc.collect()
_empty_device_cache()
print("[CLEANUP] Translation model unloaded")
# ====================================================================================
# Live Translation - Singleton Model Management and Caching
# ====================================================================================
# Global translation model state (persistent while live translation is enabled)
_live_translation_model = None
_live_translation_tokenizer = None
_live_translation_lock = threading.Lock()
_live_translation_model_loaded = False
_live_translation_model_loading = False # Track when model is being loaded
_live_translation_model_id = None # Track which model is loaded to detect config changes
_live_translation_device = None # 'cuda' | 'mps' | 'cpu' once loaded — exposed in /api/translation/status
_live_translation_is_ct2 = False # True when the loaded model is a CTranslate2 Translator
_live_translation_target_lang = None
# Set True by load/preload, False by unload. A queued unload re-checks this under
# the lock and aborts if a preload re-requested the model in the meantime, so a
# quick stop->start doesn't ack "already loaded" and then lose the model.
_live_translation_model_wanted = False
# EMA of local NLLB translate time (ms), surfaced on the health dashboard. Set in
# translate_text() around model.generate; read in get_translation_status().
_local_translate_ms_ema = None
_local_translate_ms_lock = threading.Lock()
def _record_local_translate_ms(elapsed_ms, alpha=0.3):
"""Fold one local translation timing into the running EMA (thread-safe)."""
global _local_translate_ms_ema
with _local_translate_ms_lock:
prev = _local_translate_ms_ema
_local_translate_ms_ema = elapsed_ms if prev is None else alpha * elapsed_ms + (1 - alpha) * prev
# EMA of the round-trip time (ms) for an OFFLOADED translation as seen by this
# machine (Machine A): network + serialization + Machine B's inference. Compared
# against Machine B's own local_translate_ms_ema, it separates network overhead
# from remote inference. Set in _translate_via_remote on success.
_remote_translate_ms_ema = None
_remote_translate_ms_lock = threading.Lock()
def _record_remote_translate_ms(elapsed_ms, alpha=0.3):
"""Fold one remote-translation round-trip timing into the EMA (thread-safe)."""
global _remote_translate_ms_ema
with _remote_translate_ms_lock:
prev = _remote_translate_ms_ema
_remote_translate_ms_ema = elapsed_ms if prev is None else alpha * elapsed_ms + (1 - alpha) * prev
def is_live_translation_ready():
"""True when a live translation can actually be produced right now: a remote
endpoint is configured, the LLM is the engine and is up, or the local NLLB model
has finished loading. Used to avoid persisting a warmup echo (the source text
returned unchanged while the model is still loading)."""
live_cfg = config.get("live_translation", {})
remote_cfg = live_cfg.get("remote", {})
if remote_cfg.get("enabled") and remote_cfg.get("endpoint"):
# A machine configured to offload can produce translations for its own
# display regardless of whether it also hosts trusted clients. (Serving
# a paired machine's request always translates locally — see
# translate_remote's local_only=True — so this never causes chaining.)
return True
if live_cfg.get("translation_method") == "llm":
# The LLM is the engine here; the NMT model is only its fallback, and
# llm.fallback = "skip" means it is never loaded at all. Asking the NMT
# flag in that configuration would report "not ready" for the whole
# session, and every translated caption would be produced and then
# silently dropped instead of saved.
llm_cfg = live_cfg.get("llm") or {}
if (llm_cfg.get("provider") or "endpoint").strip().lower() == "local":
return is_local_llm_loaded()
return bool((llm_cfg.get("endpoint") or "").strip() and (llm_cfg.get("model") or "").strip())
return _live_translation_model_loaded
def _warmup_translation_model(model, tokenizer, device):
"""Run one throwaway translation so MPS/CUDA compile their generate kernels
here rather than on the first real request (a multi-second cold-start spike,
worst right after a restart / hourly auto-update). No-op on CPU (nothing to
pre-compile) and when disabled via config. Never raises — a warmup failure
must not block model availability."""
try:
if not config.get("live_translation", {}).get("warmup", True):
return
# CTranslate2 Translator has no .generate — warm it via its own path.
if _live_translation_is_ct2:
t0 = time.perf_counter()
_translate_text_ct2("Hello.", "en", "es", model, tokenizer)
print(f"[LIVE-TRANSLATION] Warmup (ct2/{device}) took {(time.perf_counter() - t0) * 1000:.0f}ms")
return
if device == "cpu":
return
num_beams = int((config.get("live_translation", {}).get("generation_params", {}) or {}).get("num_beams", 2))
t0 = time.perf_counter()
# Mirror the real path per engine so the same generate kernels compile.
_is_madlad = is_madlad_model(_live_translation_model_id or "")
if _is_madlad:
inputs = tokenizer(build_madlad_input("Hello.", "es"), return_tensors="pt")
_warm_kwargs = {}
else:
tokenizer.src_lang = "eng_Latn"
inputs = tokenizer("Hello.", return_tensors="pt")
_warm_kwargs = {"forced_bos_token_id": tokenizer.convert_tokens_to_ids("spa_Latn")}
_dev = next(model.parameters()).device
if _dev.type != "cpu":
inputs = {k: v.to(_dev) for k, v in inputs.items()}
model.generate(**inputs, max_length=8, num_beams=num_beams, **_warm_kwargs)
print(f"[LIVE-TRANSLATION] Warmup ({device}) took {(time.perf_counter() - t0) * 1000:.0f}ms")
except Exception as e:
print(f"[LIVE-TRANSLATION] Warmup failed (non-fatal): {e}")
# Latched so "no fallback model configured" is said once, not once per caption:
# this runs on the caption path. Cleared as soon as a model id appears, so a
# later change back is announced again.
_no_fallback_model_logged = False
def get_live_translation_model(use_gpu=True, model_id=None):
"""Get or load the live translation model (singleton pattern).
If model_id differs from the currently loaded model, unloads and reloads."""
global _live_translation_model, _live_translation_tokenizer, _live_translation_model_loaded, _live_translation_model_loading, _live_translation_model_id, _live_translation_model_wanted
global _no_fallback_model_logged
with _live_translation_lock:
# No model configured is a choice, not a failure: the operator selected
# "None" so this machine keeps no fallback weights resident — on a
# memory-bound box that is the room a larger LLM needs. Guarding here
# rather than at each caller covers all five of them at once, including
# /api/translate/preload, which an offload client calls on every service
# start and which would otherwise load the model the choice was avoiding.
#
# Without this, an empty id reached load_translation_model(), where ""
# joined to MODELS_DIR is a real directory — so it tried to load the
# models folder as a model and raised, once per caption.
if not (model_id or "").strip():
if not _no_fallback_model_logged:
print("[LIVE-TRANSLATION] no fallback model configured — captions this "
"engine declines will stay in the source language")
_no_fallback_model_logged = True
return None, None
_no_fallback_model_logged = False
# Don't load model if transcription is actively stopping (to prevent GPU memory leak)
status = _ts_get("status", "")
if _live_translation_model is None and status == "stopping":
print("[LIVE-TRANSLATION] Skipping model load - transcription is stopping")
return None, None
_live_translation_model_wanted = True
_lt_cfg = config.get("live_translation", {})
_want_ct2 = bool(_lt_cfg.get("use_ctranslate2", False))
# If model_id OR the inference backend changed, unload so it reloads correctly.
if _live_translation_model is not None and (
(model_id and _live_translation_model_id and model_id != _live_translation_model_id)
or (_want_ct2 != _live_translation_is_ct2)):
_why = "backend" if _want_ct2 != _live_translation_is_ct2 else "model"
print(f"[LIVE-TRANSLATION] {_why} changed ({_live_translation_model_id}, ct2={_live_translation_is_ct2}) -> "
f"({model_id}, ct2={_want_ct2}), reloading...")
import gc
del _live_translation_model
del _live_translation_tokenizer
_live_translation_model = None
_live_translation_tokenizer = None
_live_translation_model_loaded = False
_live_translation_model_id = None
gc.collect()
_empty_device_cache()
get_server_text_cache().clear() # results were from the old model
if _live_translation_model is None:
_live_translation_model_loading = True
try:
print(f"[LIVE-TRANSLATION] Loading live translation model: {model_id or 'default'} (ct2={_want_ct2})...")
_live_translation_model, _live_translation_tokenizer = load_translation_model(
use_gpu=use_gpu,
model_id=model_id,
use_fp16=_lt_cfg.get("use_fp16", False),
use_ct2=_want_ct2,
ct2_compute_type=_lt_cfg.get("ct2_compute_type", "auto"),
)
_live_translation_model_loaded = True
_live_translation_model_id = model_id
print(f"[LIVE-TRANSLATION] Live translation model loaded: {model_id or 'default'}")
# The weights that actually loaded, and the device they landed on.
# Only knowable here: loading is lazy, so the session db exists first.
_record_session_meta_change(**{
"mt.effective.model": model_id,
"mt.effective.device": _live_translation_device,
})
_warmup_translation_model(_live_translation_model, _live_translation_tokenizer, _live_translation_device)
finally:
_live_translation_model_loading = False
return _live_translation_model, _live_translation_tokenizer
def unload_live_translation_model():
"""Unload the live translation model to free GPU memory"""
global _live_translation_model, _live_translation_tokenizer, _live_translation_model_loaded, _live_translation_model_id, _live_translation_model_wanted, _live_translation_device
import gc
_live_translation_model_wanted = False
with _live_translation_lock:
if _live_translation_model_wanted:
# A preload re-requested the model while this unload waited on the
# lock (quick stop->start) — keep it loaded
print("[LIVE-TRANSLATION] Unload cancelled: model re-requested")
return
if _live_translation_model is not None:
print("[LIVE-TRANSLATION] Unloading live translation model...")
del _live_translation_model
del _live_translation_tokenizer
_live_translation_model = None
_live_translation_tokenizer = None
_live_translation_model_loaded = False
_live_translation_model_id = None
_live_translation_device = None
gc.collect()
_empty_device_cache()
# Drop server-side text cache: it may hold results from the model
# we're unloading (a reload could change model/precision/output).
get_server_text_cache().clear()
print("[LIVE-TRANSLATION] Live translation model unloaded")
def is_live_translation_model_loaded():
"""Check if the live translation model is currently loaded"""
return _live_translation_model_loaded
def is_live_translation_model_loading():
"""Check if the live translation model is currently being loaded"""
return _live_translation_model_loading
# ====================================================================================
# TTS (Text-to-Speech) - Multi-backend: edge-tts (cloud), piper and supertonic (local)
# ====================================================================================
# What differs between backends and is data (config keys, audio format, language
# coverage) lives in the registry, not in if/elif chains down here.
import importlib.util
from stt import tts_backends
# Loaded local models, keyed by backend id. Both piper and supertonic are local,
# so a single model global would make switching backends unload the wrong one.
_tts_local_models = {}
_tts_sample_rates = {}
_tts_lock = threading.Lock()
_tts_model_loaded = False
_tts_model_loading = False
# Edge-TTS voice cache (populated on first request)
_edge_tts_voices = None
_edge_tts_voices_lock = threading.Lock()
def _get_tts_backend():
"""Get the configured TTS backend id ('edge', 'piper' or 'supertonic')"""
return config.get("live_translation", {}).get("tts", {}).get("backend", tts_backends.DEFAULT_BACKEND)
def _tts_settings():
"""The live ``live_translation.tts`` config section."""
return config.get("live_translation", {}).get("tts", {})
def get_edge_tts_voices():
"""Get cached list of edge-tts voices. Returns list of dicts with Name, ShortName, Gender, Locale."""
if DEMO:
# A live call to Microsoft, reached on a language switch even when TTS is off.
# The demo answers /api/tts/voices from recorded fixtures instead.
return []
global _edge_tts_voices
with _edge_tts_voices_lock:
if _edge_tts_voices is not None:
return _edge_tts_voices
try:
import edge_tts
except ImportError:
print("[TTS] edge-tts not installed. Install with: pip install edge-tts")
return []
try:
import asyncio
loop = asyncio.new_event_loop()
voices = loop.run_until_complete(edge_tts.list_voices())
loop.close()
with _edge_tts_voices_lock:
_edge_tts_voices = voices
print(f"[TTS] Loaded {len(voices)} edge-tts voices")
return voices
except Exception as e:
print(f"[TTS] Failed to fetch edge-tts voices: {e}")
return []
# Piper catalog + pure voice-selection matching live in stt/tts_catalog.py; the
# shims below inject the IO (live edge-tts voices, the downloaded-model check).
from stt.tts_catalog import ( # noqa: F401
PIPER_MODELS_CATALOG as _PIPER_MODELS_CATALOG,
edge_voice_for_lang,
piper_model_for_lang,
)
def _pick_default_edge_voice(lang_code):
"""Default edge-tts voice ShortName for a language code, or None."""
return edge_voice_for_lang(lang_code, get_edge_tts_voices())
def _pick_default_piper_model(lang_code):
"""A downloaded piper model id matching a language code, or None."""
return piper_model_for_lang(lang_code, _PIPER_MODELS_CATALOG, _is_piper_model_downloaded)
def _pick_default_supertonic_voice(lang_code):
"""Supertonic preset for a language: the current one, since voices are
multilingual. None when the model can't speak the language at all."""
current = _tts_settings().get("supertonic_voice", "")
return tts_backends.supertonic_voice_for_lang(lang_code, current)
def _default_voice_for(backend):
"""Fallback shown in the UI when no voice has been chosen for a backend yet."""
if backend.id == tts_backends.EDGE:
return "en-US-AriaNeural"
if backend.id == tts_backends.SUPERTONIC:
return tts_backends.DEFAULT_SUPERTONIC_VOICE
return "" # piper: nothing sensible until a model is downloaded
def _pick_default_voice(backend_id, lang_code):
"""Default voice for a language on any backend."""
if backend_id == tts_backends.PIPER:
return _pick_default_piper_model(lang_code)
if backend_id == tts_backends.SUPERTONIC:
return _pick_default_supertonic_voice(lang_code)
return _pick_default_edge_voice(lang_code)
def _switch_tts_voice_for_language(tts_section, old_language, new_language):
"""Point TTS at the right voice after a target-language switch.
Stashes the outgoing voice as ``old_language``'s preference and reloads the
model when the new voice needs a different one. Returns the chosen voice, or
None when the backend has nothing for that language."""
backend_id = tts_section.get("backend", tts_backends.DEFAULT_BACKEND)
backend = tts_backends.get_backend(backend_id)
previous = tts_section.get(backend.voice_key, "")
new_voice = tts_backends.select_voice(
backend_id, old_language, new_language, tts_section,
lambda lang: _pick_default_voice(backend.id, lang),
)
if new_voice:
print(f"[TTS] Auto-switched {backend.id} voice: {previous or '(none)'} -> {new_voice}")
if backend.is_local and new_voice != previous:
def _reload():
unload_tts_model(backend.id)
get_tts_model(backend=backend.id, model_name=new_voice)
threading.Thread(target=_reload, daemon=True).start()
elif backend.id == tts_backends.SUPERTONIC:
# The only backend that can be loaded and still have nothing to say:
# its model covers 30 languages, and the target isn't one of them.
print(f"[TTS] Supertonic has no voice for '{new_language}' - speaking it language-agnostically")
else:
print(f"[TTS] No {backend.id} voice available for '{new_language}' - keeping {previous or '(none)'}")
return new_voice
def _load_piper_model(model_name):
"""Construct a PiperVoice from the named downloaded model. Caller holds _tts_lock."""
from piper import PiperVoice
if not model_name:
print("[TTS ERROR] No piper model configured")
return None, None
model_path = os.path.join(_tts_cache_dir, "piper", model_name)
onnx_files = [f for f in os.listdir(model_path) if f.endswith(".onnx")] if os.path.isdir(model_path) else []
if not onnx_files:
print(f"[TTS ERROR] No .onnx model found in {model_path}")
return None, None
onnx_path = os.path.join(model_path, onnx_files[0])
json_path = onnx_path + ".json"
print(f"[TTS] Loading piper model: {model_name}...")
model = PiperVoice.load(onnx_path, config_path=json_path if os.path.exists(json_path) else None)
sample_rate = model.config.sample_rate if hasattr(model, "config") else 22050
return model, sample_rate
def _load_supertonic_model(model_name):
"""Construct the supertonic engine from our own assets dir. Caller holds _tts_lock.
``model_name`` is the voice preset, not a model: one model serves every
voice and language, so it plays no part in loading."""
from supertonic import TTS as SupertonicTTS
model_dir = _get_supertonic_dir()
if not _is_supertonic_downloaded():
print(f"[TTS ERROR] Supertonic assets not downloaded (expected in {model_dir})")
return None, None
print("[TTS] Loading supertonic model...")
# auto_download=False: the Model Manager owns downloading, so a missing
# asset surfaces as an error instead of a silent 260MB pull mid-service.
model = SupertonicTTS(model_dir=model_dir, auto_download=False)
return model, model.sample_rate
_LOCAL_MODEL_LOADERS = {
tts_backends.PIPER: _load_piper_model,
tts_backends.SUPERTONIC: _load_supertonic_model,
}
def get_tts_model(use_gpu=False, model_name=None, backend=None):
"""Load a local TTS model (one singleton per backend). Cloud backends need none."""
global _tts_model_loaded, _tts_model_loading
backend_id = backend or _get_tts_backend()
descriptor = tts_backends.get_backend(backend_id)
if not descriptor.is_local:
_tts_model_loaded = True
return True # edge-tts needs no model
with _tts_lock:
status = _ts_get("status", "")
if _tts_local_models.get(descriptor.id) is None and status == "stopping":
print(f"[TTS] Skipping {descriptor.id} model load - transcription is stopping")
return None
if _tts_local_models.get(descriptor.id) is None:
_tts_model_loading = True
try:
if model_name is None:
model_name = _tts_settings().get(descriptor.voice_key, "")
model, sample_rate = _LOCAL_MODEL_LOADERS[descriptor.id](model_name)
if model is None:
return None
_tts_local_models[descriptor.id] = model
_tts_sample_rates[descriptor.id] = sample_rate
_tts_model_loaded = True
print(f"[TTS] {descriptor.id} model loaded (sample_rate={sample_rate})")
except Exception as e:
print(f"[TTS ERROR] Failed to load {descriptor.id} model: {e}")
_tts_local_models.pop(descriptor.id, None)
_tts_model_loaded = False
finally:
_tts_model_loading = False
return _tts_local_models.get(descriptor.id)
def unload_tts_model(backend=None):
"""Unload a local TTS model to free memory (all of them when backend is None).
Defaults to every local backend rather than the active one: the common
caller is "we're switching away from X", and X is by then no longer active.
"""
global _tts_model_loaded
import gc
targets = [backend] if backend else tts_backends.local_backend_ids()
with _tts_lock:
for backend_id in targets:
model = _tts_local_models.pop(backend_id, None)
if model is not None:
print(f"[TTS] Unloading {backend_id} model...")
del model
_tts_sample_rates.pop(backend_id, None)
print(f"[TTS] {backend_id} model unloaded")
if not _tts_local_models:
_tts_model_loaded = False
gc.collect()
def is_tts_model_loaded():
descriptor = tts_backends.get_backend(_get_tts_backend())
if not descriptor.is_local:
return True # edge-tts is always ready
return _tts_local_models.get(descriptor.id) is not None
def is_tts_model_loading():
return _tts_model_loading
def _synthesize_edge_tts(text, voice=None, speed=1.0):
"""Synthesize speech using edge-tts (Microsoft Edge cloud TTS). Returns (mp3_bytes, sample_rate, 'mp3') or (None, None, None)."""
try:
import asyncio
import edge_tts
import io
if voice is None:
voice = _tts_settings().get("edge_voice", "en-US-AriaNeural")
# edge-tts rate format: "+0%", "-50%", "+100%" etc
rate_str = "+0%"
if speed != 1.0:
pct = int((speed - 1.0) * 100)
rate_str = f"{pct:+d}%"
async def _do_synth():
communicate = edge_tts.Communicate(text, voice, rate=rate_str)
buf = io.BytesIO()
async for chunk in communicate.stream():
if chunk["type"] == "audio":
buf.write(chunk["data"])
return buf.getvalue()
loop = asyncio.new_event_loop()
try:
# Bound the network call: a hung connection would otherwise block
# the TTS emit loop indefinitely
mp3_bytes = loop.run_until_complete(asyncio.wait_for(_do_synth(), timeout=15))
finally:
loop.close()
if not mp3_bytes:
return None, None, None
return mp3_bytes, 24000, "mp3" # edge-tts outputs 24kHz mp3
except Exception as e:
print(f"[TTS ERROR] edge-tts synthesis failed: {e}")
return None, None, None
def _wav_container(pcm_writer, sample_rate):
"""A mono 16-bit WAV file in memory, written by ``pcm_writer(wave_file)``."""
import io
import wave
buf = io.BytesIO()
with wave.open(buf, 'wb') as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2) # 16-bit
wav_file.setframerate(sample_rate)
pcm_writer(wav_file)
return buf.getvalue()
def _synthesize_piper_tts(text, language="en", speed=1.0):
"""Synthesize speech using piper (local). Returns (wav_bytes, sample_rate, 'wav') or (None, None, None).
``speed`` is accepted for a uniform backend signature but ignored: piper's
length scale is baked into the downloaded voice."""
model = get_tts_model(backend=tts_backends.PIPER)
if model is None:
return None, None, None
try:
sample_rate = _tts_sample_rates.get(tts_backends.PIPER, 22050)
wav_bytes = _wav_container(lambda wav_file: model.synthesize(text, wav_file), sample_rate)
return wav_bytes, sample_rate, "wav"
except Exception as e:
print(f"[TTS ERROR] Piper synthesis failed: {e}")
return None, None, None
def _synthesize_supertonic_tts(text, language="en", speed=1.0):
"""Synthesize speech using supertonic (local). Returns (wav_bytes, sample_rate, 'wav') or (None, None, None).
One model covers 30 languages; anything outside them is synthesized
language-agnostically rather than refused."""
model = get_tts_model(backend=tts_backends.SUPERTONIC)
if model is None:
return None, None, None
try:
import numpy as np
settings = _tts_settings()
voice_name = settings.get("supertonic_voice") or tts_backends.DEFAULT_SUPERTONIC_VOICE
style = _get_supertonic_style(model, voice_name)
wav, _duration = model.synthesize(
text,
voice_style=style,
lang=tts_backends.supertonic_lang(language),
total_steps=tts_backends.clamp_supertonic_steps(settings.get("supertonic_steps")),
speed=tts_backends.clamp_supertonic_speed(speed),
)
# float32 (1, N) in [-1, 1] -> the 16-bit PCM the browser decodes
samples = np.clip(np.asarray(wav).reshape(-1), -1.0, 1.0)
pcm = (samples * 32767.0).astype("<i2").tobytes()
sample_rate = _tts_sample_rates.get(tts_backends.SUPERTONIC, model.sample_rate)
wav_bytes = _wav_container(lambda wav_file: wav_file.writeframes(pcm), sample_rate)
return wav_bytes, sample_rate, "wav"
except Exception as e:
print(f"[TTS ERROR] Supertonic synthesis failed: {e}")
return None, None, None
_supertonic_styles = {}
def _get_supertonic_style(model, voice_name):
"""Voice style vectors for a preset, loaded once per voice."""
style = _supertonic_styles.get(voice_name)
if style is None:
style = model.get_voice_style(voice_name=voice_name)
_supertonic_styles[voice_name] = style
return style
# EMA of TTS synthesis time (ms), surfaced on the health dashboard.
_tts_synth_ms_ema = None
_tts_synth_ms_lock = threading.Lock()
def _record_tts_ms(elapsed_ms, alpha=0.3):
"""Fold a TTS synth timing into the EMA (thread-safe)."""
global _tts_synth_ms_ema
with _tts_synth_ms_lock:
prev = _tts_synth_ms_ema
_tts_synth_ms_ema = elapsed_ms if prev is None else alpha * elapsed_ms + (1 - alpha) * prev
_TTS_SYNTHESIZERS = {
tts_backends.EDGE: lambda text, language, speed: _synthesize_edge_tts(text, speed=speed),
tts_backends.PIPER: _synthesize_piper_tts,
tts_backends.SUPERTONIC: _synthesize_supertonic_tts,
}
def synthesize_tts(text, language="en"):
"""Synthesize speech from text. Returns (audio_bytes, sample_rate, format) or (None, None, None).
The format is reported by the backend that produced the audio rather than
inferred by the caller.
"""
speed = _tts_settings().get("speed", 1.0)
backend = tts_backends.get_backend(_get_tts_backend())
_t0 = time.perf_counter()
result = _TTS_SYNTHESIZERS[backend.id](text, language, speed)
try:
if result and result[0] is not None:
_record_tts_ms((time.perf_counter() - _t0) * 1000.0)
except Exception:
pass
return result
# Global translation cache instance
_translation_cache = TranslationCache()
# Server-side cache for offloaded /api/translate requests: keyed by
# (text, langs, num_beams) so repeated phrases skip model.generate. Distinct
# from _translation_cache (segment-id keyed). Sized from config on first use.
_server_text_cache = TextTranslationCache(
max_size=int((config.get("live_translation", {}).get("remote", {}) or {}).get("server_cache_size", 512) or 512)
)
def get_translation_cache():
"""Get the global translation cache"""
return _translation_cache
def get_server_text_cache():
"""Get the server-side text translation cache (for offloaded requests)."""
return _server_text_cache
# ====================================================================================
# Model Factory - Supports multiple STT model types
# ====================================================================================
class ModelFactory:
"""Factory class for loading different types of speech-to-text models"""
_model_cache: ClassVar[dict] = {} # shared across instances by design
_cache_lock = threading.Lock()
@staticmethod
def load_model(model_config, use_gpu=True):
"""
Load a speech-to-text model based on configuration with caching
Args:
model_config: Dictionary with model configuration
use_gpu: Whether to use GPU acceleration
Returns:
Tuple of (model, processor/tokenizer, model_type)
"""
# Import ML libraries before using them
_lazy_import_ml_libraries()
model_type = model_config.get("type", "whisper")
if use_gpu and torch.cuda.is_available():
device = "cuda"
elif use_gpu and hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
device = "mps"
else:
device = "cpu"
# Create cache key
cache_key = f"{model_type}_{model_config!s}_{device}"
# Check cache first
with ModelFactory._cache_lock:
if cache_key in ModelFactory._model_cache:
print(f"Using cached {model_type} model on {device}")
return ModelFactory._model_cache[cache_key]
print(f"Loading {model_type} model on {device}...")
try:
if model_type == "whisper":
# Check for backend preference (faster-whisper or standard whisper)
backend = model_config.get("backend")
if backend == "faster-whisper":
print("Using faster-whisper backend (4-10x faster)")
model, processor, model_type_return = ModelFactory._load_faster_whisper(
model_config["whisper"], use_gpu
)
else:
print("Using standard OpenAI Whisper backend")
model, processor, model_type_return = ModelFactory._load_whisper(
model_config["whisper"], use_gpu
)
elif model_type == "huggingface":
model, processor, model_type_return = ModelFactory._load_huggingface(
model_config["huggingface"], device
)
elif model_type == "custom":
model, processor, model_type_return = ModelFactory._load_custom(
model_config["custom"], device
)
else:
raise ValueError(f"Unknown model type: {model_type}")
# Cache the loaded model
with ModelFactory._cache_lock:
ModelFactory._model_cache[cache_key] = (
model,
processor,
model_type_return,
)
return model, processor, model_type_return
except Exception as e:
print(f"[ERROR] Failed to load model: {e}")
raise
@staticmethod
def cleanup_models():
"""Clean up all cached models to free memory"""
import gc
with ModelFactory._cache_lock:
# First, copy items and clear cache to remove references
cache_items = list(ModelFactory._model_cache.items())
ModelFactory._model_cache.clear()
# Now delete model objects from the copied list
for cache_key, (model, processor, _) in cache_items:
try:
# Move to CPU first if possible (frees GPU memory faster)
if hasattr(model, "cpu"):
model.cpu()
# Delete the actual model object
del model
if processor:
del processor
except Exception as e:
print(f"[WARNING] Error cleaning up model {cache_key}: {e}")
# Delete the list too
del cache_items
# Force garbage collection OUTSIDE the lock
# This is CRITICAL for ctranslate2/faster-whisper to release GPU memory
gc.collect()
# Now clear the accelerator cache (CUDA or MPS)
try:
_empty_device_cache()
print("[OK] All models cleaned up from memory, GPU cache cleared")
except Exception as e:
print(f"[OK] All models cleaned up from memory (GPU cleanup: {e})")
@staticmethod
def _load_whisper(whisper_config, device):
"""Load OpenAI Whisper model"""
_lazy_import_ml_libraries()
model_name = whisper_config.get("model", "base")
print(f"Loading Whisper model: {model_name}")
# Check if model exists in ./models directory first
models_dir = MODELS_DIR
whisper_model_dir = os.path.join(models_dir, f"whisper-{model_name}")
# Determine download_root based on where model is located
download_root = None
if os.path.exists(whisper_model_dir):
# Model exists in new ./models location
download_root = whisper_model_dir
print(f"Using Whisper model from: {whisper_model_dir}")
else:
# Fall back to checking old cache location
whisper_cache_old = os.path.expanduser("~/.cache/whisper")
model_file = f"{model_name}.pt"
old_model_path = os.path.join(whisper_cache_old, model_file)
if os.path.exists(old_model_path):
print(f"Using Whisper model from cache: {whisper_cache_old}")
else:
raise FileNotFoundError(
f"Whisper model '{model_name}' is not downloaded. "
f"Please download it first from the Model Manager (Settings → Model Manager)."
)
# Load the model (model_name can include .en suffix for English-only variants)
model = whisper.load_model(model_name, download_root=download_root)
if device == "cuda":
model = model.cuda()
return model, None, "whisper"
@staticmethod
def _load_faster_whisper(whisper_config, use_gpu=True):
"""Load faster-whisper model (CTranslate2-based, 4-10x faster)"""
# Must happen BEFORE importing faster_whisper/ctranslate2
_preload_cudnn(use_gpu)
from faster_whisper import WhisperModel
model_name = whisper_config.get("model", "small")
compute_type = whisper_config.get("compute_type", "auto")
print(f"Loading faster-whisper model: {model_name}")
# Auto-detect best compute type based on hardware
if compute_type == "auto":
if use_gpu and torch.cuda.is_available():
# Check GPU compute capability for float16 support
gpu_props = torch.cuda.get_device_properties(0)
if gpu_props.major >= 7:
compute_type = "float16"
else:
compute_type = "float32"
else:
compute_type = "int8" # CPU optimized
device = "cuda" if use_gpu and torch.cuda.is_available() else "cpu"
# Check for local model first
models_dir = MODELS_DIR
local_model_path = os.path.join(models_dir, f"faster-whisper-{model_name}")
# Directory existence is not enough. A partial download (or a partial
# delete) leaves a directory that passes os.path.exists and then fails
# deep inside the loader — or worse, does not fail at all: with
# tokenizer.json absent, faster-whisper falls back to
# tokenizers.Tokenizer.from_pretrained("openai/whisper-tiny"), a Rust
# builtin with its own HTTP client that honours neither HF_HUB_OFFLINE nor
# any timeout we can pass. On a connection that stalls rather than
# refusing, that blocks for ever, and because the worker only reports
# "running" once the load returns, the UI sits on STARTING with no error.
# (local_files_only=True does not help: faster-whisper only forwards it to
# download_model(), which is skipped entirely for a local directory.)
status = _model_files.faster_whisper_status(local_model_path)
if status.state == "absent":
raise FileNotFoundError(
f"Faster-whisper model '{model_name}' is not downloaded. "
f"Please download it first from the Model Manager (Settings → Model Manager)."
)
if not status.complete:
raise FileNotFoundError(
f"Faster-whisper model '{model_name}' is incomplete — "
f"{_model_files.describe_missing(status.missing)} "
f"{'are' if len(status.missing) > 1 else 'is'} missing or truncated. "
f"Open the Model Manager and press Repair on this model."
)
model_path = local_model_path
print(f"Using faster-whisper model from: {local_model_path}")
print(f"Device: {device}, Compute type: {compute_type}")
model = WhisperModel(model_path, device=device, compute_type=compute_type)
return model, None, "faster_whisper"
@staticmethod
def _load_huggingface(hf_config, device):
"""Load Hugging Face transformers model"""
_lazy_import_ml_libraries()
model_id = hf_config.get("model_id", "openai/whisper-tiny")
use_flash_attention = hf_config.get("use_flash_attention", False)
# Check if model exists locally in ./models directory
models_dir = MODELS_DIR
model_dir_name = model_id.replace("/", "--")
local_model_path = os.path.join(models_dir, model_dir_name)
# Use local path if it exists, otherwise tell user to download first
if os.path.exists(local_model_path):
model_path = local_model_path
print(f"Loading Hugging Face model from local path: {local_model_path}")
else:
raise FileNotFoundError(
f"HuggingFace model '{model_id}' is not downloaded. "
f"Please download it first from the Model Manager (Settings → Model Manager)."
)
try:
# Determine model architecture. Ask the directory we already have
# before asking the internet: model_info() defaults to timeout=None,
# and this ran on every load of a fully-local model — so a stalled
# connection hung the worker with nothing wrong on disk at all.
pipeline_tag = None
try:
with open(os.path.join(model_path, "config.json"), encoding="utf-8") as _cf:
pipeline_tag = _model_files.pipeline_tag_from_config(json.load(_cf))
except (OSError, ValueError) as _cfg_err:
print(f"[WARN] Could not read local config.json ({_cfg_err}); "
f"falling back to the model card")
if pipeline_tag is None:
# Only when the local files could not answer. Bounded, and failing
# open: an unreachable Hub must not stop a downloaded model loading.
try:
pipeline_tag = model_info(model_id, timeout=15).pipeline_tag
except Exception as _info_err:
print(f"[WARN] Could not reach the model card ({_info_err}); "
f"inferring from the model id")
# Load based on architecture
if (
"whisper" in model_id.lower()
or pipeline_tag == "automatic-speech-recognition"
):
# Whisper-based models (including Distil-Whisper)
torch_dtype = torch.float16 if device == "cuda" else torch.float32
model_kwargs = {
"torch_dtype": torch_dtype,
"low_cpu_mem_usage": True,
}
if use_flash_attention and device == "cuda":
model_kwargs["attn_implementation"] = "flash_attention_2"
model = AutoModelForSpeechSeq2Seq.from_pretrained(
model_path, **model_kwargs
)
model.to(device)
processor = AutoProcessor.from_pretrained(model_path)
# Create pipeline for easier inference
pipe = pipeline(
"automatic-speech-recognition",
model=model,
tokenizer=processor.tokenizer,
feature_extractor=processor.feature_extractor,
max_new_tokens=128,
chunk_length_s=30,
batch_size=16,
torch_dtype=torch_dtype,
device=device,
)
return pipe, processor, "huggingface_whisper"
elif "wav2vec2" in model_id.lower():
# Wav2Vec2 models
model = AutoModelForCTC.from_pretrained(model_path)
model.to(device)
processor = Wav2Vec2Processor.from_pretrained(model_path)
return model, processor, "huggingface_wav2vec2"
else:
# Generic ASR model
pipe = pipeline(
"automatic-speech-recognition", model=model_path, device=device
)
return pipe, None, "huggingface_generic"
except Exception as e:
print(f"Error loading Hugging Face model: {e}")
raise
@staticmethod
def _load_custom(custom_config, device):
"""Load custom model from local path"""
_lazy_import_ml_libraries()
model_path = custom_config.get("model_path", "")
model_type = custom_config.get("model_type", "whisper")
if not model_path or not os.path.exists(model_path):
raise ValueError(f"Custom model path not found: {model_path}")
print(f"Loading custom {model_type} model from: {model_path}")
if model_type == "whisper":
# If model_path is a directory (e.g., ./models/whisper-base),
# check if it contains a .pt file
if os.path.isdir(model_path):
pt_files = [f for f in os.listdir(model_path) if f.endswith(".pt")]
if pt_files:
# Use the directory as download_root and extract model name
model_file = pt_files[0]
model_name = model_file.replace(".pt", "").replace(".en", "")
model = whisper.load_model(model_name, download_root=model_path)
else:
raise ValueError(f"No .pt files found in directory: {model_path}")
else:
# model_path points directly to a .pt file
model = whisper.load_model(model_path)
if device == "cuda":
model = model.cuda()
return model, None, "whisper"
else:
# Try loading as Hugging Face model
pipe = pipeline(
"automatic-speech-recognition", model=model_path, device=device
)
return pipe, None, "huggingface_generic"
@staticmethod
def transcribe(model, processor, model_type, audio_data, language="auto", whisper_params=None, return_segments=False):
"""
Transcribe audio using the loaded model
Args:
model: The loaded model
processor: The processor/tokenizer (if applicable)
model_type: Type of model ('whisper', 'huggingface_whisper', etc.)
audio_data: Audio data as numpy array
language: Language code (default: 'en', 'auto' for auto-detection)
whisper_params: Dict of Whisper decoding parameters (optional)
For Whisper models: beam_size, temperature, condition_on_previous_text, etc.
See LIVE_TRANSCRIPTION_PARAMS and FILE_TRANSCRIPTION_PARAMS constants
return_segments: If True, return list of segments with timestamps instead of just text
Returns:
If return_segments=False: Transcription text (str)
If return_segments=True: List of segment dicts with 'text', 'start', 'end' keys
"""
try:
if model_type == "whisper":
# Original Whisper model (OpenAI whisper)
# Build params dict with language and whisper_params
params = {}
if language != "auto":
params["language"] = language
# OpenAI whisper supported parameters (transcribe-level + DecodingOptions)
whisper_transcribe_params = {
"verbose", "temperature", "compression_ratio_threshold",
"logprob_threshold", "no_speech_threshold",
"condition_on_previous_text", "initial_prompt",
"word_timestamps", "prepend_punctuations", "append_punctuations",
"clip_timestamps", "hallucination_silence_threshold",
"carry_initial_prompt",
# DecodingOptions params
"task", "language", "sample_len", "best_of", "beam_size",
"patience", "length_penalty", "prefix", "suppress_tokens",
"suppress_blank", "without_timestamps", "max_initial_timestamp",
"fp16",
}
if whisper_params:
for k, v in whisper_params.items():
if k.startswith("_"):
continue
if k in whisper_transcribe_params:
params[k] = v
result = model.transcribe(audio_data, **params)
if return_segments:
# Return Whisper's native segments with timestamps
segments = result.get("segments", [])
return [{"text": seg["text"].strip(), "start": seg["start"], "end": seg["end"]} for seg in segments if seg["text"].strip()]
return result["text"].strip()
elif model_type == "faster_whisper":
# faster-whisper model (CTranslate2-based)
# Build params dict with language and whisper_params
params = {
"vad_filter": False, # External VAD already screens; internal VAD over-chunks long buffers
}
if language != "auto":
params["language"] = language
# faster-whisper supported parameters (different from standard whisper)
# NOTE: initial_prompt and hotwords are intentionally excluded.
# Both are tokenized into the decoder prefix and consume decoder positions
# (max 448 total). The full Bible-book hotwords list is ~227 tokens alone,
# leaving only ~220 positions for actual transcription — not enough for
# dense speech. Whisper's language setting is sufficient for recognition.
faster_whisper_params = {
"beam_size", "best_of", "patience", "length_penalty",
"repetition_penalty", "no_repeat_ngram_size",
"temperature", "compression_ratio_threshold",
"log_prob_threshold", "no_speech_threshold",
"condition_on_previous_text",
"prefix", "suppress_blank", "suppress_tokens",
"without_timestamps", "max_initial_timestamp",
"word_timestamps", "prepend_punctuations",
"append_punctuations", "vad_filter", "vad_parameters",
"task",
}
# Parameter name mapping (whisper -> faster-whisper)
param_mapping = {
"logprob_threshold": "log_prob_threshold", # Different naming
}
if whisper_params:
for k, v in whisper_params.items():
if k.startswith("_"):
continue
# Map parameter name if needed
mapped_key = param_mapping.get(k, k)
# Only include if faster-whisper supports it
if mapped_key in faster_whisper_params:
params[mapped_key] = v
# faster-whisper returns (segments_iterator, info)
segments_iter, info = model.transcribe(audio_data, **params)
# Convert iterator to list with standard format
segments = []
full_text = []
for seg in segments_iter:
text = seg.text.strip()
if text:
seg_dict = {
"text": text,
"start": seg.start,
"end": seg.end,
"no_speech_prob": getattr(seg, "no_speech_prob", 0),
"avg_logprob": getattr(seg, "avg_logprob", 0),
# Detected source language (ISO code) so rows can record source_language
# even when audio.language is 'auto'. Same for every seg in the chunk.
"language": getattr(info, "language", None),
}
# Extract word-level confidence if available
if hasattr(seg, 'words') and seg.words:
seg_dict["words"] = [
{
"word": w.word,
"probability": getattr(w, "probability", None),
"start": w.start,
"end": w.end,
}
for w in seg.words
]
segments.append(seg_dict)
full_text.append(text)
if return_segments:
return segments
return " ".join(full_text)
elif model_type == "huggingface_whisper":
# Hugging Face Whisper pipeline
generate_kwargs = {}
if language != "auto":
generate_kwargs["language"] = language
# Map Whisper params to HuggingFace generate_kwargs
if whisper_params:
if "beam_size" in whisper_params:
generate_kwargs["num_beams"] = whisper_params["beam_size"]
if "temperature" in whisper_params:
_hf_temp = whisper_params["temperature"]
if isinstance(_hf_temp, (list, tuple)):
# HF supports a fallback tuple in long-form generation
generate_kwargs["temperature"] = tuple(float(t) for t in _hf_temp)
elif isinstance(_hf_temp, (int, float)):
generate_kwargs["temperature"] = _hf_temp
# Quality thresholds: honored by HF Whisper long-form generation;
# stripped via the retry below when the installed transformers
# version doesn't accept them.
if "compression_ratio_threshold" in whisper_params:
generate_kwargs["compression_ratio_threshold"] = whisper_params["compression_ratio_threshold"]
if "logprob_threshold" in whisper_params:
generate_kwargs["logprob_threshold"] = whisper_params["logprob_threshold"]
if "no_speech_threshold" in whisper_params:
generate_kwargs["no_speech_threshold"] = whisper_params["no_speech_threshold"]
if "condition_on_previous_text" in whisper_params:
# HF names this condition_on_prev_tokens
generate_kwargs["condition_on_prev_tokens"] = whisper_params["condition_on_previous_text"]
_hf_optional_keys = (
"compression_ratio_threshold", "logprob_threshold",
"no_speech_threshold", "condition_on_prev_tokens",
)
def _hf_pipeline_call(**call_kwargs):
try:
return model(audio_data, generate_kwargs=generate_kwargs, **call_kwargs) if generate_kwargs else model(audio_data, **call_kwargs)
except (TypeError, ValueError) as hf_err:
stripped = {k: v for k, v in generate_kwargs.items() if k not in _hf_optional_keys}
if len(stripped) == len(generate_kwargs):
raise
print(f"[WARNING] HF pipeline rejected quality thresholds ({hf_err}); retrying without them")
return model(audio_data, generate_kwargs=stripped, **call_kwargs) if stripped else model(audio_data, **call_kwargs)
if return_segments:
# Request timestamps from HuggingFace pipeline
result = _hf_pipeline_call(return_timestamps=True)
chunks = result.get("chunks", [])
segments = []
for chunk in chunks:
text = chunk.get("text", "").strip()
timestamp = chunk.get("timestamp", (0, 0))
if text and timestamp:
start = timestamp[0] if timestamp[0] is not None else 0
end = timestamp[1] if timestamp[1] is not None else start
segments.append({"text": text, "start": start, "end": end})
return segments
result = _hf_pipeline_call()
return result["text"].strip()
elif model_type == "huggingface_wav2vec2":
# Wav2Vec2 model (doesn't support language parameter or timestamps)
import torch
inputs = processor(
audio_data, sampling_rate=16000, return_tensors="pt", padding=True
)
with torch.no_grad():
logits = model(inputs.input_values.to(model.device)).logits
predicted_ids = torch.argmax(logits, dim=-1)
transcription = processor.batch_decode(predicted_ids)
text = transcription[0].strip()
if return_segments:
# Wav2Vec2 doesn't provide timestamps, return as single segment
return [{"text": text, "start": 0, "end": 0}] if text else []
return text
elif model_type == "huggingface_generic":
# Generic Hugging Face pipeline
# Try to pass language if supported, otherwise just use audio
try:
if return_segments:
if language == "auto":
result = model(audio_data, return_timestamps=True)
else:
result = model(audio_data, return_timestamps=True, generate_kwargs={"language": language})
chunks = result.get("chunks", [])
segments = []
for chunk in chunks:
text = chunk.get("text", "").strip()
timestamp = chunk.get("timestamp", (0, 0))
if text and timestamp:
start = timestamp[0] if timestamp[0] is not None else 0
end = timestamp[1] if timestamp[1] is not None else start
segments.append({"text": text, "start": start, "end": end})
return segments
if language == "auto":
result = model(audio_data)
else:
result = model(
audio_data, generate_kwargs={"language": language}
)
except (TypeError, ValueError, RuntimeError) as e:
# Fallback if language parameter not supported
print(f"[WARNING] Model language parameter failed ({e}), falling back to auto-detect")
result = model(audio_data)
if return_segments:
text = result["text"].strip()
return [{"text": text, "start": 0, "end": 0}] if text else []
return result["text"].strip()
else:
raise ValueError(f"Unknown model type: {model_type}")
except Exception as e:
print(f"Transcription error: {e}")
return [] if return_segments else ""
class WhisperLiveTranscriber:
"""
Streaming transcription using whisper-live approach.
Uses a rolling numpy buffer instead of dual confirmed/active buffers.
Segments are finalized when the same output is repeated N times (same_output_threshold).
Ported from: Whisper-Live-main/whisper_live/backend/base.py
"""
RATE = 16000 # Sample rate in Hz
def __init__(
self,
sample_rate=16000,
same_output_threshold=7,
no_speech_thresh=0.45,
send_last_n_segments=10,
):
"""
Initialize the transcriber.
Args:
sample_rate: Audio sample rate (default 16000)
same_output_threshold: Number of repeated outputs before finalizing (default 7)
no_speech_thresh: Threshold for filtering no-speech segments (default 0.45)
send_last_n_segments: Number of recent segments to keep (default 10)
"""
self.RATE = sample_rate
self.same_output_threshold = same_output_threshold
self.no_speech_thresh = no_speech_thresh
self.send_last_n_segments = send_last_n_segments
# Frame buffer (numpy array of float32 audio samples)
self.frames_np = None
self.frames_offset = 0.0 # Time offset when buffer was clipped
self.timestamp_offset = 0.0 # Current transcription position
# Segment tracking
self.transcript = [] # List of completed segments
self.current_out = "" # Current incomplete output
self.prev_out = "" # Previous output for comparison
self.same_output_count = 0
self.end_time_for_same_output = None
self._last_seg_confidence = {} # Word-level confidence from last segment
# Threading lock for buffer access
self.lock = threading.Lock()
def _is_similar_output(self, text1, text2, threshold=0.85):
"""
Check if two outputs are similar enough to count as 'same'.
Uses fuzzy matching because Whisper often returns slightly different text
each iteration (e.g., "I'm going to" vs "I'm gonna").
Args:
text1: First text to compare
text2: Second text to compare
threshold: Similarity threshold (0.0-1.0), default 0.85
Returns:
bool: True if texts are similar enough
"""
if not text1 or not text2:
return False
from difflib import SequenceMatcher
ratio = SequenceMatcher(None, text1.lower().strip(), text2.lower().strip()).ratio()
return ratio >= threshold
def add_frames(self, audio_bytes, sample_width=2):
"""
Add audio frames to the buffer.
Converts bytes to float32 numpy array and appends to rolling buffer.
If buffer exceeds 45 seconds, clips oldest 30 seconds.
Args:
audio_bytes: Raw audio bytes (int16 PCM)
sample_width: Bytes per sample (default 2 for int16)
"""
# Convert bytes to float32 numpy array normalized to [-1, 1]
frame_np = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32) / 32768.0
with self.lock:
# Clip buffer if it exceeds 45 seconds
if self.frames_np is not None and self.frames_np.shape[0] > 45 * self.RATE:
self.frames_offset += 30.0
self.frames_np = self.frames_np[int(30 * self.RATE):]
# Ensure timestamp_offset doesn't fall behind frames_offset
if self.timestamp_offset < self.frames_offset:
self.timestamp_offset = self.frames_offset
# Append frames
if self.frames_np is None:
self.frames_np = frame_np.copy()
else:
self.frames_np = np.concatenate((self.frames_np, frame_np), axis=0)
def get_audio_chunk_for_processing(self):
"""
Get the next audio chunk to transcribe.
Returns audio from timestamp_offset to end of buffer.
Returns:
tuple: (audio_np, duration) - audio as float32 numpy array and duration in seconds
"""
with self.lock:
if self.frames_np is None:
return np.array([], dtype=np.float32), 0.0
samples_to_skip = max(0, int((self.timestamp_offset - self.frames_offset) * self.RATE))
audio_chunk = self.frames_np[samples_to_skip:].copy()
duration = audio_chunk.shape[0] / self.RATE if audio_chunk.shape[0] > 0 else 0.0
return audio_chunk, duration
def get_buffer_duration(self):
"""Get total duration of audio in buffer in seconds."""
with self.lock:
if self.frames_np is None:
return 0.0
return self.frames_np.shape[0] / self.RATE
def update_segments(self, segments, duration):
"""
Process segments from Whisper transcription using Whisper-Live's approach.
Key insight: Whisper returns segments with timestamps. We finalize all
segments except the last one immediately. The last segment stays as
"in-progress" and only finalizes when it repeats (same_output_threshold).
This prevents overlapping text because we use Whisper's segment boundaries
instead of guessing where phrases end.
Args:
segments: List of dicts with 'text', 'start', 'end' keys
duration: Duration of the audio chunk that was transcribed
Returns:
dict: {
'completed_segments': list of newly completed segments,
'current_text': current incomplete text (last segment),
'is_finalized': whether last segment was just finalized
}
"""
result = {
'completed_segments': [],
'current_text': '',
'is_finalized': False
}
if not segments:
return result
# FIX: Detect garbage output from Whisper (overwhelmed by too much audio)
# When Whisper gets 30+ seconds, it often returns just '...' or empty text
all_text = ' '.join(seg.get('text', '').strip() for seg in segments)
if len(all_text) < 5 or all_text == '...' or all_text.strip() == '':
# Whisper is overwhelmed - force buffer trim and skip garbage
with self.lock:
if self.frames_np is not None:
buffer_duration = self.frames_np.shape[0] / self.RATE
current_pos = self.timestamp_offset - self.frames_offset
chunk_to_process = buffer_duration - current_pos
if chunk_to_process > 10:
# Force advance to keep only 10 seconds
extra_advance = chunk_to_process - 10
self.timestamp_offset += extra_advance
return result # Skip processing garbage segments
offset = None
# Process all segments except the last one (finalize them immediately)
# This is the key difference from our previous approach
if len(segments) > 1:
for seg in segments[:-1]:
text = seg.get('text', '').strip()
if not text:
continue
start = self.timestamp_offset + seg.get('start', 0)
end = self.timestamp_offset + min(duration, seg.get('end', duration))
if start >= end:
continue
completed = {
'start': start,
'end': end,
'text': text,
'completed': True,
}
# Pass through word-level confidence data if available
if 'words' in seg:
completed['words'] = seg['words']
if 'avg_logprob' in seg:
completed['avg_logprob'] = seg['avg_logprob']
if 'no_speech_prob' in seg:
completed['no_speech_prob'] = seg['no_speech_prob']
if seg.get('language'):
completed['language'] = seg['language']
self.transcript.append(completed)
result['completed_segments'].append(completed)
# print(f"[SEGMENT] Finalized: '{text[:50]}...' ({seg.get('start', 0):.1f}s-{seg.get('end', 0):.1f}s)" if len(text) > 50 else f"[SEGMENT] Finalized: '{text}'", flush=True)
offset = min(duration, seg.get('end', duration))
# Handle the last segment (in-progress until repeated)
last_seg = segments[-1]
self.current_out = last_seg.get('text', '').strip()
# Store last segment's confidence data for finalization
self._last_seg_confidence = {
k: last_seg[k] for k in ('words', 'avg_logprob', 'no_speech_prob', 'language') if k in last_seg
}
result['current_text'] = self.current_out
# Check if last segment is repeating (same_output_threshold logic)
if self._is_similar_output(self.current_out, self.prev_out) and self.current_out:
self.same_output_count += 1
if self.end_time_for_same_output is None:
self.end_time_for_same_output = last_seg.get('end', duration)
# Debug logging for same_output tracking
else:
self.same_output_count = 0
self.end_time_for_same_output = None
# Finalize last segment if repeated enough times
if self.same_output_count >= self.same_output_threshold:
if self.current_out:
completed = {
'start': self.timestamp_offset,
'end': self.timestamp_offset + min(duration, self.end_time_for_same_output or duration),
'text': self.current_out,
'completed': True,
}
# Pass through word-level confidence from last segment
if 'words' in last_seg:
completed['words'] = last_seg['words']
if 'avg_logprob' in last_seg:
completed['avg_logprob'] = last_seg['avg_logprob']
if 'no_speech_prob' in last_seg:
completed['no_speech_prob'] = last_seg['no_speech_prob']
if last_seg.get('language'):
completed['language'] = last_seg['language']
self.transcript.append(completed)
result['completed_segments'].append(completed)
result['is_finalized'] = True
# FIX: Save the text that was just finalized so phrase_complete knows not to re-process
result['just_finalized_text'] = self.current_out
print(f"[SAME_OUTPUT] Finalized last segment: '{self.current_out[:50]}...'" if len(self.current_out) > 50 else f"[SAME_OUTPUT] Finalized: '{self.current_out}'", flush=True)
offset = min(duration, self.end_time_for_same_output or duration)
# Reset
self.current_out = ''
self.prev_out = ''
self.same_output_count = 0
self.end_time_for_same_output = None
result['current_text'] = ''
else:
self.prev_out = self.current_out
# Advance timestamp_offset by the end of finalized segments
if offset is not None:
with self.lock:
self.timestamp_offset += offset
# PROACTIVE FIX: ALWAYS check buffer size and limit it, not just when offset is None
# This prevents the buffer from slowly growing over time even when segments are finalizing
with self.lock:
if self.frames_np is not None:
buffer_duration = self.frames_np.shape[0] / self.RATE
current_pos = self.timestamp_offset - self.frames_offset
chunk_to_process = buffer_duration - current_pos
# If chunk exceeds 20 seconds, force advance to keep ~15 seconds
# This is more aggressive than before to prevent Whisper from being overwhelmed
if chunk_to_process > 20:
extra_advance = chunk_to_process - 15 # Keep 15 seconds
self.timestamp_offset += extra_advance
return result
def force_finalize(self):
"""
Force finalize current text (e.g., on phrase timeout / silence detection).
Returns:
dict: Segment if there was text to finalize, None otherwise
"""
if not self.current_out:
return None
# Get current duration from buffer
_, duration = self.get_audio_chunk_for_processing()
segment = {
'start': self.timestamp_offset,
'end': self.timestamp_offset + duration,
'text': self.current_out,
'completed': True,
}
# Include stored confidence data from last segment
if hasattr(self, '_last_seg_confidence') and self._last_seg_confidence:
segment.update(self._last_seg_confidence)
self.transcript.append(segment)
# Update timestamp offset
with self.lock:
self.timestamp_offset += duration
# Reset
self.current_out = ""
self.prev_out = ""
self.same_output_count = 0
self.end_time_for_same_output = None
return segment
def get_recent_segments(self):
"""Get the most recent completed segments."""
return self.transcript[-self.send_last_n_segments:] if self.transcript else []
def get_all_text(self):
"""Get all transcribed text concatenated."""
return " ".join(seg['text'] for seg in self.transcript if seg.get('text'))
def audio_end_time(self):
"""Seconds of audio heard so far, on the clock segment start/end times use."""
with self.lock:
buffered = 0 if self.frames_np is None else self.frames_np.shape[0]
return self.frames_offset + buffered / float(self.RATE)
def reset(self):
"""Reset transcriber state for new session."""
with self.lock:
self.frames_np = None
self.frames_offset = 0.0
self.timestamp_offset = 0.0
self.transcript = []
self.current_out = ""
self.prev_out = ""
self.same_output_count = 0
self.end_time_for_same_output = None
# Use 'spawn' everywhere, not just where it is the platform default.
#
# The parent process loads the live-translation model, and when that runs on a local
# GPU it initialises a CUDA context. A *forked* child inherits that context, and CUDA
# forbids re-initialising it — so the transcription worker dies the moment it touches
# the GPU:
#
# RuntimeError: Cannot re-initialize CUDA in forked subprocess.
# To use CUDA with multiprocessing, you must use the 'spawn' start method
#
# That makes "Restart Transcription" impossible on Linux+CUDA once translation has
# loaded, which is exactly the configuration that runs translation on the same box.
# macOS has always defaulted to spawn and this module is written for that path (see
# the note below about the child re-importing and receiving its state as arguments),
# so forcing it on Linux aligns the two rather than introducing an untried mode.
#
# Must happen before the Manager and Queues below: they inherit the active context,
# and mixing contexts is unsupported.
if not DEMO and multiprocessing.current_process().name == "MainProcess":
try:
if multiprocessing.get_start_method(allow_none=True) != "spawn":
multiprocessing.set_start_method("spawn", force=True)
print("[INIT] multiprocessing start method set to 'spawn' (CUDA-safe)")
except RuntimeError as e:
# Already started something; leave the existing method rather than crash.
print(f"[INIT] WARNING: could not set spawn start method: {e}")
# Create shared state only in the main process.
# On macOS, 'spawn' is the default start method (safe — avoids ObjC/fork crashes after
# PyTorch/Whisper initialize the Objective-C runtime with background threads).
# On Linux, 'fork' was the default; the block above now forces spawn there too.
# With spawn (macOS), the child re-imports this module and must NOT recreate the Manager
# (it would fail before bootstrap completes). Instead, the child receives these objects
# as pickled arguments to thread1_function and assigns them to module globals there.
#
# A demo has no worker process — playback and the web server are threads in one
# process — so it uses plain dicts and queues instead. That keeps a spawned Manager
# child out of the frozen bundle (the part of multiprocessing least happy under
# PyInstaller) while leaving every transcription_state[...] call site unchanged.
if DEMO:
mp_manager = _demo_mode.LocalManager()
config_queue = _demo_mode.local_queue()
control_queue = _demo_mode.local_queue()
audio_stream_queue = _demo_mode.local_queue(maxsize=10)
hq_stream_queue = _demo_mode.local_queue(maxsize=20)
elif multiprocessing.current_process().name == 'MainProcess':
# multiprocessing.Manager() with an initializer: the Manager's own process watches for
# this one to disappear and exits with it, instead of outliving a killed server
# (stt/parent_watch.py). Manager() is exactly SyncManager(ctx=...).start().
from multiprocessing.managers import SyncManager as _SyncManager
from stt.parent_watch import watch_parent as _watch_parent
mp_manager = _SyncManager(ctx=multiprocessing.get_context())
mp_manager.start(initializer=_watch_parent)
# Create multiprocessing Queue for config updates (hot-reload)
config_queue = MPQueue()
# Create multiprocessing Queue for control commands (start/stop)
control_queue = MPQueue()
# Create multiprocessing Queue for streaming audio to web clients
audio_stream_queue = MPQueue(maxsize=10)
# Full-quality listening chunks (100 ms each) from the same capture; dropped when full
hq_stream_queue = MPQueue(maxsize=20)
else:
# Spawned worker process: shared objects will be received as function arguments
# and assigned to these globals at the top of thread1_function.
mp_manager = None
config_queue = None
control_queue = None
audio_stream_queue = None
hq_stream_queue = None
if mp_manager is not None:
# Global transcription state - use Manager.dict() for cross-process sharing
transcription_state = mp_manager.dict(
{
"running": False,
"status": "stopped",
"message": "Transcription not started",
"error": None, # Error message if status == "error"
"db_name": None, # Shared database name for cross-process access
"session_id": None, # Stable per-session id (.db filename stem); cross-process
"audio_level": 0, # Audio level for histogram (0-100)
"audio_db": -60, # Audio level in decibels
"audio_stalled": False, # True when ffmpeg has produced no audio for 10s+ (mic muted/unplugged) — mirrors FFmpegAudioCapture.is_stalled
"audio_silent": False, # True when audio arrives but every sample has been exactly 0 for 10s+ (macOS mic permission denied, device muted) — stt/digital_silence.py
"audio_energy": 0, # Raw audio energy (RMS)
"start_time": 0, # epoch seconds when transcription became active; 0 = not running
"live_text": "", # Live preview text (not yet saved to DB)
"live_start": 0, # Start time of the live preview within the session
"live_end": 0, # End time of the live preview within the session
"live_word_confidences": [], # Word-level confidence for the live preview
"loaded_model": "", # Name of the actual model that was loaded
"audio_stream_enabled": False, # Whether to stream audio to web clients
"audio_hq_enabled": False, # Whether anyone is in the full-quality listening room
"audio_device": None, # Device the worker's capture opened; the HQ listening capture opens the same one
"audio_type": None, # "Speaking", "Music", or "Quiet" — PANNs detection (no_speech_prob fallback)
"detection_mode": None, # "panns" (tagger live) or "energy" (fallback) — which detector is actually running
"loaded_model_device": None, # "cuda" / "mps" / "cpu" the ASR model landed on
"model_load_ms": None, # How long the ASR model took to load (ms)
"infer_ms_ema": None, # EMA of per-chunk transcribe time (ms) — health dashboard
"rtf_ema": None, # EMA real-time factor (transcribe_s / audio_s); <1 keeps up
"segments_total": 0, # Chunks transcribed this session (throughput numerator)
"segments_per_min": None, # Throughput over the session window
"rows_saved": 0, # Finalized transcript lines saved to the session DB
"queue_depth": None, # audio_stream_queue depth, when readable
"filter_stats": None, # Sentence-filter counts + alert status (stt/disposition_stats.py)
# Which init step the worker is on, and when it got there. Only
# meaningful while status == "starting"; see stt/start_watch.py.
"init_stage": "",
"init_stage_at": None,
}
)
# Shared calibration state for cross-process communication
calibration_state = mp_manager.dict(
{
"active": False,
"step": 1, # 1 = noise floor calibration, 2 = speech calibration
"step1_complete": False,
"start_time": 0,
"duration": 15, # 15 seconds per step (30 total)
"speech_samples": 0,
"noise_samples": 0,
"silence_samples": 0,
}
)
# Shared calibration data storage (Manager lists for cross-process)
calibration_data_shared = mp_manager.dict(
{
"speech_samples": mp_manager.list(),
"noise_samples": mp_manager.list(),
"silence_durations": mp_manager.list(),
"energy_levels": mp_manager.list(),
"vad_probabilities": mp_manager.list(),
}
)
# Step 1 calibration data (noise floor only)
calibration_step1_data = mp_manager.dict(
{
"noise_energies": mp_manager.list(),
"avg_noise": 0.0,
"max_noise": 0.0,
}
)
else:
transcription_state = None
calibration_state = None
calibration_data_shared = None
calibration_step1_data = None
# Global reference to transcription process for restart functionality
transcription_process = None
def _spawn_worker():
"""Build and start a transcription worker, returning it only once started.
Callers assign the result to ``transcription_process``, which the main loop
joins once a second. Assigning the bare Process and calling ``start()`` on
the next line left a window — seconds long under Windows' spawn — in which
the loop joined an unstarted process, hit "can only join a started process"
and lost the main thread (STT-2J).
"""
proc = multiprocessing.Process(
target=thread1_function,
args=(transcription_state, control_queue, config_queue,
calibration_state, calibration_data_shared, calibration_step1_data,
audio_stream_queue, hq_stream_queue)
)
proc.start()
return proc
# Set once the server begins shutting down / restarting. The transcription_state
# Manager proxy dies when the Manager process is torn down (execv restart, signal,
# auto-update), after which any proxy access raises BrokenPipeError/EOFError/
# ConnectionError. Background emit threads check this and exit cleanly instead of
# crashing with an unhandled error.
_server_shutting_down = threading.Event()
# Every way a dead Manager proxy can fail: broken pipe/EOF mid-call,
# ConnectionRefusedError to the AF_UNIX listener, FileNotFoundError when the
# socket file is gone, AttributeError/TypeError when the proxy itself is None.
_TS_PROXY_ERRORS = (BrokenPipeError, EOFError, ConnectionError, FileNotFoundError, AttributeError, TypeError)
def _ts_get(key, default=None):
"""Read from the transcription_state Manager proxy, tolerating a torn-down
Manager during shutdown. On a proxy disconnect, flag shutdown and return the
default instead of raising an unhandled BrokenPipeError."""
try:
return transcription_state.get(key, default)
except _TS_PROXY_ERRORS:
_server_shutting_down.set()
return default
def _ts_snapshot():
"""dict(transcription_state), tolerating a torn-down Manager during a
shutdown/restart (e.g. the auto-update window). Returns a benign
'restarting' state on proxy disconnect instead of raising an unhandled 500
from the status routes."""
try:
return dict(transcription_state)
except _TS_PROXY_ERRORS:
_server_shutting_down.set()
return {"running": False, "status": "restarting", "message": "Server is restarting"}
# Database cache for performance
_db_cache = {
"last_entries": [],
"last_fetch_time": 0,
"cache_duration": 1.0, # Cache for 1 second
}
# Thread locks for synchronization
_db_lock = threading.Lock()
_cache_lock = threading.Lock()
_transcription_state_lock = threading.Lock()
_transcription_start_lock = threading.Lock() # Guards worker-process creation in the start route
_audio_queue_lock = threading.Lock()
# Generate the current date and time as a string
current_datetime = datetime.now().strftime("%Y-%m-%d_%H-%M_%A")
current_year = datetime.now().strftime("%Y")
current_month = datetime.now().strftime("%Y-%m")
# Database will be created lazily when transcription starts
db_name = None # Will be set when database is initialized
db_initialized = False
live_session_id = None # Stable per-session id (the .db filename stem); set in initialize_database
# ============== PANNs audio tagger (music / speech detection) ==============
# Default checkpoint location (kept under APP_DIR so compiled builds stay self-contained)
PANNS_CHECKPOINT = os.path.join(APP_DIR, "panns_data", "Cnn14_mAP=0.431.pth")
PANNS_CHECKPOINT_URL = "https://zenodo.org/record/3987831/files/Cnn14_mAP%3D0.431.pth?download=1"
# AudioSet class labels. panns_inference loads these at IMPORT TIME from a hardcoded
# ~/panns_data/class_labels_indices.csv and (over plain http) tries to wget them if
# absent. When that fetch is blocked/offline it leaves an empty file, which is never
# retried -> labels=[] -> classes_num=0 -> the 527-class CNN14 checkpoint fails to load
# -> music detection silently falls back to energy-only ("Speaking"/"Quiet", never
# "Music"). We ship the CSV under APP_DIR and place a valid copy before importing the
# library (with an https fallback) so detection works self-contained / offline.
PANNS_LABELS_FILENAME = "class_labels_indices.csv"
PANNS_LABELS_BUNDLED = os.path.join(APP_DIR, "panns_data", PANNS_LABELS_FILENAME)
PANNS_LABELS_URL = "https://storage.googleapis.com/us_audioset/youtube_corpus/v1/csv/class_labels_indices.csv"
_PANNS_LABELS_MIN_BYTES = 1024 # a valid 527-row CSV is ~14 KB; smaller == missing/poisoned
#: Below this, the CNN14 checkpoint is a truncated download rather than a model.
#: A floor rather than an exact size: PANNS_CKPT_SIZE is approximate and only ever
#: drove a progress bar, so asserting it as an expectation would fail a good file.
#: Same shape as the labels floor above, for the same reason — it separates
#: "interrupted" from "fine" without claiming precision we do not have.
#:
#: Without it a dropped transfer left a short .pth that every check called
#: present, the download endpoint refused to re-fetch because the path existed,
#: and the detector met it much later.
_PANNS_CKPT_MIN_BYTES = 250_000_000
_panns_labels_ready = False # only need to repair the on-disk CSV once per process
_audio_tagger = None
_audio_tagger_failed_key = None # (device, ckpt) that hit a real load error — don't retry it
_audio_tagger_key = None # (device, ckpt) currently loaded
_panns_label_idx = None # (music_idx_list, speech_idx_list)
_panns_missing_logged = False # avoid spamming the "checkpoint missing" log
def panns_checkpoint_path(cfg=None):
"""Resolve the CNN14 checkpoint path (config override or the default location)."""
cfg = cfg if cfg is not None else config.get("speech_type_detection", {})
custom = (cfg.get("checkpoint_path", "") or "").strip()
return custom or PANNS_CHECKPOINT
def panns_checkpoint_ok(path=None):
"""True when the checkpoint is present and big enough to be real."""
ckpt = path or panns_checkpoint_path()
try:
return os.path.getsize(ckpt) >= _PANNS_CKPT_MIN_BYTES
except OSError:
return False
def panns_labels_home_path():
"""The path panns_inference.config hardcodes for the AudioSet label CSV."""
return os.path.join(os.path.expanduser("~"), "panns_data", PANNS_LABELS_FILENAME)
def ensure_panns_labels_csv():
"""Guarantee a valid class_labels_indices.csv exists where panns_inference expects
it, BEFORE the library is imported (it hardcodes ~/panns_data at import time).
All PANNs data lives in APP_DIR/panns_data; ~/panns_data is kept as a symlink
into it so nothing real is stored outside the app folder. Best-effort: never
raises."""
global _panns_labels_ready
if _panns_labels_ready:
return
try:
home_csv = panns_labels_home_path()
home_dir = os.path.dirname(home_csv)
app_dir = os.path.dirname(PANNS_LABELS_BUNDLED)
# The app-folder copy is the real storage — make sure it's valid first
# (https download; the library itself only tries plain http, often blocked).
if not (os.path.exists(PANNS_LABELS_BUNDLED) and os.path.getsize(PANNS_LABELS_BUNDLED) >= _PANNS_LABELS_MIN_BYTES):
os.makedirs(app_dir, exist_ok=True)
import urllib.request
urllib.request.urlretrieve(PANNS_LABELS_URL, PANNS_LABELS_BUNDLED)
print(f"[PANNS] Downloaded AudioSet labels -> {PANNS_LABELS_BUNDLED}")
if not (os.path.exists(PANNS_LABELS_BUNDLED) and os.path.getsize(PANNS_LABELS_BUNDLED) >= _PANNS_LABELS_MIN_BYTES):
print("[PANNS] AudioSet labels unavailable; music detection will be unavailable")
return
# Point ~/panns_data at the app folder. Migrate a stale real directory
# only when it holds nothing but the label CSV (never delete unknown data).
if os.path.islink(home_dir):
if os.path.realpath(home_dir) != os.path.realpath(app_dir):
os.unlink(home_dir)
elif os.path.isdir(home_dir):
if all(name == PANNS_LABELS_FILENAME for name in os.listdir(home_dir)):
shutil.rmtree(home_dir)
if not os.path.lexists(home_dir):
try:
os.symlink(app_dir, home_dir)
print(f"[PANNS] Linked {home_dir} -> {app_dir}")
except OSError:
pass
# Fallback when the symlink couldn't be made (or a real dir with other
# content remains): copy the CSV like before.
if not (os.path.exists(home_csv) and os.path.getsize(home_csv) >= _PANNS_LABELS_MIN_BYTES):
os.makedirs(home_dir, exist_ok=True)
shutil.copyfile(PANNS_LABELS_BUNDLED, home_csv)
print(f"[PANNS] Installed AudioSet labels from bundled copy -> {home_csv}")
if os.path.exists(home_csv) and os.path.getsize(home_csv) >= _PANNS_LABELS_MIN_BYTES:
_panns_labels_ready = True
else:
print("[PANNS] AudioSet labels still missing after repair attempt; music detection will be unavailable")
except Exception as e:
print(f"[PANNS] Could not install AudioSet labels: {e}")
def panns_package_installed():
try:
import importlib.util
return importlib.util.find_spec("panns_inference") is not None
except Exception:
return False
def get_audio_tagger(cfg):
"""Lazy-load the PANNs CNN14 tagger for the given speech_type_detection cfg.
Reloads if device/checkpoint changed. Returns None if unavailable (missing
package or checkpoint). MUST be called off the audio-drain path (it can block
for seconds on first load)."""
global _audio_tagger, _audio_tagger_failed_key, _audio_tagger_key, _panns_label_idx, _panns_missing_logged
# Make sure the AudioSet label CSV is valid before panns_inference is imported
# below (it loads labels once at import time). Without this the checkpoint fails
# to load and detection silently degrades to energy-only.
ensure_panns_labels_csv()
ckpt = panns_checkpoint_path(cfg)
device = cfg.get("device", "cpu") or "cpu"
key = (device, ckpt)
if _audio_tagger is not None and _audio_tagger_key == key:
return _audio_tagger
if _audio_tagger_failed_key == key:
return None
# Missing checkpoint is *transient* (a download in the main process may produce
# it later) — recheck cheaply each call instead of failing permanently.
# A short file counts as missing: it is an interrupted download, and handing
# it to AudioTagging buys a confusing failure instead of the honest fallback.
if not panns_checkpoint_ok(ckpt):
if not _panns_missing_logged:
_why = "incomplete" if os.path.exists(ckpt) else "not found"
print(f"[PANNS] Checkpoint {_why} at {ckpt}; music detection falls back "
f"to energy-based until it's downloaded")
_panns_missing_logged = True
return None
# Config changed (device/checkpoint) or first load: drop any stale model.
_audio_tagger = None
_audio_tagger_key = None
try:
from panns_inference import AudioTagging, labels
tagger = AudioTagging(checkpoint_path=ckpt, device=device)
music_idx = [i for i, l in enumerate(labels) if l in ("Music", "Singing")]
_panns_label_idx = (music_idx, None)
_audio_tagger = tagger
_audio_tagger_key = key
_audio_tagger_failed_key = None
_panns_missing_logged = False
print(f"[PANNS] Audio tagger loaded on {device} from {ckpt}")
return _audio_tagger
except Exception as e:
_audio_tagger_failed_key = key
print(f"[PANNS] Could not load audio tagger: {e}; falling back to energy-based")
return None
def unload_audio_tagger():
"""Release the tagger (called when transcription stops/unloads to free VRAM).
The detector reloads it lazily if detection is used again."""
global _audio_tagger, _audio_tagger_key, _audio_tagger_failed_key, _panns_missing_logged
_audio_tagger = None
_audio_tagger_key = None
_audio_tagger_failed_key = None
_panns_missing_logged = False
def compute_music_prob(audio_np, sr, cfg):
"""Return (music_prob, dominant_tag) for an audio buffer using PANNs, or
(None, None) when the tagger is unavailable. audio_np: float32 mono in [-1, 1]."""
tagger = get_audio_tagger(cfg)
if tagger is None or audio_np is None or len(audio_np) == 0:
return None, None
try:
from panns_inference import labels
wav = np.asarray(audio_np, dtype=np.float32)
if sr != 32000:
import librosa
wav = librosa.resample(wav, orig_sr=sr, target_sr=32000)
clipwise, _ = tagger.inference(wav[None, :])
clip = clipwise[0]
music_idx, _ = _panns_label_idx
music_prob = float(max((clip[i] for i in music_idx), default=0.0))
return music_prob, labels[int(np.argmax(clip))]
except Exception as e:
print(f"[PANNS] inference error: {e}")
return None, None
# Audio-type labels and word-attribution logic live in stt/segments.py
# (importable, unit-tested); names are re-imported so call sites stay unchanged.
from stt import segments as _segments
from stt import session_cleanup as _session_cleanup
from stt import wav_edit as _wav_edit
# Waveform envelopes for the trim editor, kept so reopening a recording (or a
# second operator opening the same one) does not re-read the file.
_wav_peaks_cache = _wav_edit.PeaksCache()
from stt.segments import ( # noqa: E402, F401
attribute_words_to_sentences,
effective_music_prob,
panns_label_from_prob,
words_json_or_none,
words_to_session_ms,
)
def classify_audio_type(audio_db, cfg=None):
"""Energy-based fallback label (no PANNs): audible => Speaking, else Quiet.
We never claim Music without the PANNs detector."""
return _segments.classify_audio_type(audio_db, cfg if cfg is not None else config.get("speech_type_detection", {}))
class MusicDetector:
"""Runs PANNs inference on a dedicated daemon thread so the audio-drain loop
never blocks. Latest-buffer / drop-stale: submit() just hands off the newest
buffer; the thread loads the model (off the hot path), throttles, smooths, and
writes music_prob / audio_tag / audio_type into the shared transcription state.
Reads speech_type_detection live from process_config each iteration -> hot-reload."""
def __init__(self):
self.cfg_root = {} # live process_config (refreshed on each submit)
self.state = None # shared transcription_state
self._buf = None
self._sr = 16000
self._lock = threading.Lock()
self._event = threading.Event()
self._history = [] # instance-local smoothing buffer (no cross-thread race)
self._last_ts = 0.0
self._thread = threading.Thread(target=self._run, daemon=True, name="panns-detector")
self._thread.start()
def submit(self, process_config, state, audio_np, sr):
"""Non-blocking hand-off of the latest raw (pre-VAD) buffer."""
self.cfg_root = process_config
self.state = state
with self._lock:
self._buf = audio_np
self._sr = sr
self._event.set()
def _run(self):
while True:
self._event.wait(timeout=1.0)
self._event.clear()
cfg = (self.cfg_root or {}).get("speech_type_detection", {})
if not cfg.get("enabled", True) or cfg.get("method", "panns") != "panns":
self._history.clear()
# Clear stale PANNs state so the live monitor doesn't keep showing
# the last music label after detection is disabled mid-session.
st = self.state
if st is not None:
st["detection_mode"] = "energy"
if st.get("music_prob") is not None:
st["music_prob"] = None
st["audio_tag"] = None
st["audio_type"] = None
continue
now = time.time()
if now - self._last_ts < 0.4: # throttle to ~2-3 runs/sec
continue
with self._lock:
buf = self._buf
sr = self._sr
self._buf = None # consume: don't re-run on a stale buffer once audio stops
if buf is None:
continue
self._last_ts = now
try:
music_prob, tag = compute_music_prob(buf, sr, cfg)
except Exception as e:
print(f"[PANNS] detector error: {e}")
continue
if music_prob is None:
# PANNs enabled but the tagger is unavailable (missing/failed load):
# finalized_audio_type falls back to the energy-based label.
st = self.state
if st is not None:
st["detection_mode"] = "energy"
continue
window = max(1, int(cfg.get("smoothing_window", 4) or 1))
self._history.append(effective_music_prob(float(music_prob), tag, cfg))
del self._history[:-window]
smoothed = sum(self._history) / len(self._history)
st = self.state
if st is not None:
st["detection_mode"] = "panns"
st["music_prob"] = music_prob
st["audio_tag"] = tag
# Live TYPE for the monitor, even when this audio isn't transcribed.
st["audio_type"] = panns_label_from_prob(smoothed, st.get("audio_db"), cfg)
_music_detector = None
def submit_music_detection(process_config, state, audio_np, sr):
"""Hand the latest pre-VAD buffer to the background detector (non-blocking).
Creates/starts the detector thread on first use."""
global _music_detector
if _music_detector is None:
_music_detector = MusicDetector()
_music_detector.submit(process_config, state, audio_np, sr)
def finalized_audio_type(process_config, state):
"""Label for a finalized (transcribed) segment: the detector's live PANNs
label when active, else the energy-based fallback."""
cfg = process_config.get("speech_type_detection", {})
if (cfg.get("enabled", True) and cfg.get("method", "panns") == "panns"
and state.get("music_prob") is not None):
return state.get("audio_type") or "Speaking"
return classify_audio_type(state.get("audio_db"), cfg)
def _set_asr_row_stamp(db_connection, label):
"""Make new rows carry ``label`` as asr_model, or nothing when it is None.
A trigger rather than an addition to the INSERT statements: there are nineteen
of those with nineteen different column lists — the segment batch, the phrase
timeout, the stop flush, and a denied variant of each — and an approach that has
to touch all nineteen is one that silently misses the twentieth when it is
added. A trigger cannot be bypassed by a writer that does not know about it.
Called with None at session start, so rows stay NULL while the session's own
model is transcribing and the value lives once in session_meta. Called with a
label after a hot reload, so every row from that point says what changed. The
boundary between the two is then visible in the rows themselves.
"""
cursor = db_connection.cursor()
cursor.execute("DROP TRIGGER IF EXISTS stamp_asr_model")
if label:
cursor.execute(
"CREATE TRIGGER stamp_asr_model AFTER INSERT ON transcriptions"
" WHEN NEW.asr_model IS NULL"
" BEGIN UPDATE transcriptions SET asr_model = '%s' WHERE id = NEW.id; END"
% label.replace("'", "''")
)
db_connection.commit()
def initialize_database(session_config=None):
"""Initialize database only when transcription starts (lazy loading)
``session_config`` is the config this session actually runs on. The worker
process is reused across Start/Stop cycles and reloads config from disk at
each session start, so its module-level `config` is whatever was on disk when
the process spawned — every setting read here must come from the session's
own config or the database is created (and described) from a stale one.
"""
global db_name, db_initialized, live_session_id
if db_initialized:
return db_name
cfg = session_config or config
# Get custom database path from config or use default
custom_db_path = cfg.get("database", {}).get("path", "").strip()
path_format = cfg.get("database", {}).get("path_format", "").strip() or "%Y/%m"
now = datetime.now()
formatted_path = now.strftime(path_format)
if custom_db_path:
# Use custom base path + path_format subdirectory
folder_name = os.path.join(custom_db_path, formatted_path)
print(f"[OK] Using custom database path: {folder_name}")
else:
# Use default base path (under APP_DIR) + path_format subdirectory so compiled
# builds keep the DB in ~/.stt instead of the launch directory.
folder_name = os.path.join(BACKUP_DIR, formatted_path)
print(f"[OK] Using default database path: {folder_name}")
# Create the folder if it doesn't exist
os.makedirs(folder_name, exist_ok=True)
# Make the DB directory tree readable/traversable by all users (consumers read these)
make_dirs_world_readable(folder_name, custom_db_path or BACKUP_DIR)
# Create database file path with configurable format (using Python strftime format)
filename_format = cfg.get("database", {}).get(
"filename_format", ""
).strip() or "%Y-%m-%d_%H%M%S"
# Validate that format includes time component for unique per-session databases
if not any(time_fmt in filename_format for time_fmt in ["%H", "%M", "%S"]):
print(f"[WARNING] Database filename_format '{filename_format}' does not include time component.")
print("[WARNING] This may cause sessions on the same day to share a database.")
print("[WARNING] Using default format: %Y-%m-%d_%H%M%S for unique per-session databases.")
filename_format = "%Y-%m-%d_%H%M%S"
now = datetime.now()
# Use strftime directly with user's format
formatted_filename = now.strftime(filename_format)
# Get custom filename prefix or use default
filename_prefix = cfg.get("database", {}).get("filename_prefix", "").strip()
if filename_prefix:
db_name = os.path.join(
folder_name, f"{formatted_filename}_{filename_prefix}.db"
)
print(
f"[OK] Using custom database filename: {formatted_filename}_{filename_prefix}.db"
)
else:
db_name = os.path.join(folder_name, f"{formatted_filename}.db")
print(
f"[OK] Using default database filename: {formatted_filename}.db"
)
print(f"[OK] Initializing database: {db_name}")
try:
# Use context manager for database connection
with sqlite3.connect(db_name) as db_connection:
db_cursor = db_connection.cursor()
# Enable WAL mode for better concurrent read/write performance
# WAL (Write-Ahead Logging) allows simultaneous reads while writing
db_cursor.execute("PRAGMA journal_mode=WAL")
db_cursor.execute("PRAGMA synchronous=NORMAL") # Faster writes, still safe
# Create the table if it doesn't exist (with start_time, end_time for temporal ordering)
db_cursor.execute(
"""CREATE TABLE IF NOT EXISTS transcriptions (id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, text TEXT, start_time REAL DEFAULT 0, end_time REAL DEFAULT 0)"""
)
# Create index on timestamp for faster ORDER BY queries
db_cursor.execute(
"""CREATE INDEX IF NOT EXISTS idx_timestamp ON transcriptions(timestamp DESC)"""
)
# Migration: Check if id column exists, add it if missing
db_cursor.execute("PRAGMA table_info(transcriptions)")
columns = [row[1] for row in db_cursor.fetchall()]
if "id" not in columns:
print("[DB] Migrating database: adding id column...")
# SQLite doesn't support ALTER TABLE ADD COLUMN with PRIMARY KEY
# So we need to recreate the table (wrapped in transaction for safety)
try:
db_cursor.execute("BEGIN")
db_cursor.execute(
"""CREATE TABLE transcriptions_new (id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, text TEXT)"""
)
db_cursor.execute(
"""INSERT INTO transcriptions_new (timestamp, text) SELECT timestamp, text FROM transcriptions"""
)
db_cursor.execute("""DROP TABLE transcriptions""")
db_cursor.execute(
"""ALTER TABLE transcriptions_new RENAME TO transcriptions"""
)
# Recreate index
db_cursor.execute(
"""CREATE INDEX IF NOT EXISTS idx_timestamp ON transcriptions(timestamp DESC)"""
)
db_connection.commit()
print("[DB] OK: Migration complete")
except Exception:
db_connection.rollback()
raise
# Migration: Check if start_time/end_time columns exist, add them if missing
db_cursor.execute("PRAGMA table_info(transcriptions)")
columns = [row[1] for row in db_cursor.fetchall()]
if "start_time" not in columns:
print("[DB] Migrating database: adding start_time and end_time columns...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN start_time REAL DEFAULT 0")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN end_time REAL DEFAULT 0")
db_connection.commit()
print("[DB] OK: Migration complete (added temporal columns)")
# Migration: Add corrections-related columns
db_cursor.execute("PRAGMA table_info(transcriptions)")
columns = [row[1] for row in db_cursor.fetchall()]
if "confidence" not in columns:
print("[DB] Migrating database: adding corrections columns (confidence, original_text, corrected_by, needs_review)...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN confidence REAL DEFAULT NULL")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN original_text TEXT DEFAULT NULL")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN corrected_by TEXT DEFAULT NULL")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN needs_review INTEGER DEFAULT 0")
db_cursor.execute("CREATE INDEX IF NOT EXISTS idx_needs_review ON transcriptions(needs_review) WHERE needs_review = 1")
db_connection.commit()
print("[DB] OK: Migration complete (added corrections columns)")
if "translated_text" not in columns:
print("[DB] Migrating database: adding translation columns (translated_text, translation_language)...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN translated_text TEXT DEFAULT NULL")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN translation_language TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added translation columns)")
if "speech_type" not in columns:
print("[DB] Migrating database: adding speech_type column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN speech_type TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added speech_type column)")
if "audio_tag" not in columns:
print("[DB] Migrating database: adding audio_tag column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN audio_tag TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added audio_tag column)")
if "music_prob" not in columns:
print("[DB] Migrating database: adding music_prob column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN music_prob REAL DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added music_prob column)")
if "denied" not in columns:
# `denied` is a UI visibility/hide flag (0 = visible, 1 = hidden from
# the transcript view); toggled by handle_set_segment_denied.
print("[DB] Migrating database: adding denied column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN denied INTEGER DEFAULT 0")
db_connection.commit()
print("[DB] OK: Migration complete (added denied column)")
# Schema v2: additive columns for the downstream consumer (epoch-ms ordering,
# per-word timing/confidence, partial/final flag, source language, segment pairing).
# All nullable / defaulted so old .db files and readers keep working unchanged.
db_cursor.execute("PRAGMA table_info(transcriptions)")
columns = [row[1] for row in db_cursor.fetchall()]
if "ts_ms" not in columns:
print("[DB] Migrating database: adding ts_ms column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN ts_ms INTEGER DEFAULT NULL")
db_cursor.execute("CREATE INDEX IF NOT EXISTS idx_ts_ms ON transcriptions(ts_ms)")
db_connection.commit()
print("[DB] OK: Migration complete (added ts_ms column)")
if "words_json" not in columns:
print("[DB] Migrating database: adding words_json column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN words_json TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added words_json column)")
if "is_final" not in columns:
print("[DB] Migrating database: adding is_final column...")
# 1 = finalized (all existing rows are finals); 0 = partial hypothesis
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN is_final INTEGER DEFAULT 1")
db_connection.commit()
print("[DB] OK: Migration complete (added is_final column)")
if "partial_seq" not in columns:
print("[DB] Migrating database: adding partial_seq column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN partial_seq INTEGER DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added partial_seq column)")
if "source_language" not in columns:
print("[DB] Migrating database: adding source_language column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN source_language TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added source_language column)")
if "segment_id" not in columns:
print("[DB] Migrating database: adding segment_id column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN segment_id TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added segment_id column)")
if "words_source" not in columns:
# ASR backend that produced the row (faster_whisper/whisper/...). Makes a
# NULL words_json interpretable: 'whisper' emits no per-word data, so NULL
# is expected there, vs 'faster_whisper' where NULL would be unexpected.
print("[DB] Migrating database: adding words_source column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN words_source TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added words_source column)")
if "session_id" not in columns:
# Stable per-session id (the .db filename stem) on every row, so the
# consumer can anchor socket<->db and group rows by session.
print("[DB] Migrating database: adding session_id column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN session_id TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added session_id column)")
if "denied_reason" not in columns:
# Why the row was denied: 'hallucination', 'cjk', 'cjk_shadow', 'short', 'dup'.
# NULL means the row is a normal visible segment.
print("[DB] Migrating database: adding denied_reason column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN denied_reason TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added denied_reason column)")
if "marked" not in columns:
# Manual operator bookmark (0 = normal, 1 = marked). Set from the
# corrections page so a segment can be found again later; unrelated
# to needs_review (the low-confidence review queue).
print("[DB] Migrating database: adding marked column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN marked INTEGER DEFAULT 0")
db_cursor.execute("CREATE INDEX IF NOT EXISTS idx_marked ON transcriptions(marked) WHERE marked = 1")
db_connection.commit()
print("[DB] OK: Migration complete (added marked column)")
if "translation_ts_ms" not in columns:
# Epoch-ms when translated_text was written. Translation arrives via a
# later async UPDATE, so the row's ts_ms alone can't reproduce the
# source-vs-translation timing a live consumer experienced.
print("[DB] Migrating database: adding translation_ts_ms column...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN translation_ts_ms INTEGER DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added translation_ts_ms column)")
if "asr_model" not in columns:
# What produced this row, per row rather than per session. The session
# records what was *configured*; a caption records what actually ran,
# and the two part company exactly where it matters. A session set to
# translate with the LLM still contains NMT rows wherever the LLM
# declined a caption, and nothing distinguished them — so measuring an
# LLM change against a past service compared it against the other
# model's output on precisely the rows the two disagree about.
print("[DB] Migrating database: adding model provenance columns...")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN asr_model TEXT DEFAULT NULL")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN mt_engine TEXT DEFAULT NULL")
db_cursor.execute("ALTER TABLE transcriptions ADD COLUMN mt_model TEXT DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added asr_model, mt_engine, mt_model columns)")
_attempt_cols = _missing_attempt_columns(set(columns))
if _attempt_cols:
# How a caption got its translation: how many tries, and whether the
# live loop or the backfill finally managed it. See
# stt.translation_attempts.
print("[DB] Migrating database: adding translation attempt columns...")
for _col, _kind in _attempt_cols:
db_cursor.execute(f"ALTER TABLE transcriptions ADD COLUMN {_col} {_kind} DEFAULT NULL")
db_connection.commit()
print("[DB] OK: Migration complete (added mt_attempts, mt_via columns)")
# A row's asr_model is left NULL while the session's own model is the one
# transcribing, because session_meta already records that and repeating it
# per row is the same string written a thousand times (160 KB on a 3,500-row
# service). It is stamped only after a hot reload changes the model, so a
# value means "not what the session started with" — see
# _set_asr_row_stamp(), which installs the trigger at that point.
try:
_set_asr_row_stamp(db_connection, None)
except Exception as _asr_err:
print(f"[DB] WARNING: could not prepare the ASR row stamp ({_asr_err})")
# Service-phase tables. Created here, inside the init transaction, so the
# detector's tick (which runs in the web process, on its own connection)
# never races a CREATE against the writer. Empty tables are harmless on a
# session where the feature is disabled.
try:
_service_phase_ensure_tables(db_connection)
except Exception as _sp_err:
print(f"[DB] WARNING: service phase tables unavailable ({_sp_err})")
# Insert a blank first entry with default values
default_timestamp = " "
default_text = " "
db_cursor.execute(
"INSERT INTO transcriptions (timestamp, text) VALUES (?, ?)",
(default_timestamp, default_text),
)
db_connection.commit()
db_initialized = True
# Make the DB file (and any WAL/SHM sidecars) readable by all users
make_db_world_readable(db_name)
# Record which models and decode settings produced this session, so a
# transcript reviewed weeks later can be attributed to the transcription
# or the translation stage instead of being unattributable once the server
# is restarted or retuned. Written after the init transaction closes so
# there is only ever one writer, and non-fatal by contract: a session must
# start even when provenance can't be recorded.
if _session_meta_enabled(cfg):
_session_provenance = _current_session_meta(cfg)
if _write_session_meta(db_name, _session_provenance):
print(f"[DB] OK: Recorded session provenance ({len(_session_provenance)} settings)")
# When translation is offloaded, the remote's model is the one that
# translates — fetch it over the network off-thread so a slow or
# unreachable remote can't delay the start of transcription.
if _translation_is_offloaded(cfg.get("live_translation", {})):
_record_remote_provenance_async(db_name)
# Stable per-session id = the .db filename stem (e.g. 2026-06-22_183007).
# Stored on every row and emitted top-level on every socket payload so the
# consumer can anchor socket<->db by exact match; re-derived each session.
live_session_id = os.path.splitext(os.path.basename(db_name))[0]
# Store database name + session id in shared state for web server access
transcription_state["db_name"] = db_name
transcription_state["session_id"] = live_session_id
print("[OK] Database initialized successfully")
return db_name
except Exception as e:
print(f"[ERROR] Failed to initialize database: {e}")
# Clean up database file if initialization failed
if db_name and os.path.exists(db_name):
try:
os.unlink(db_name)
except OSError:
pass
raise
# File Transcription Helper Functions
def extract_audio_from_file(file_path):
"""
Extract audio from video/audio file using pydub.
Converts to WAV 16kHz mono for transcription.
Args:
file_path: Path to audio/video file
Returns:
Path to converted WAV file
"""
temp_wav = None
try:
# Get file extension
ext = os.path.splitext(file_path)[1].lower().replace(".", "")
# Load file
_lazy_import_audio()
if ext in SUPPORTED_VIDEO_FORMATS + SUPPORTED_AUDIO_FORMATS:
audio = AudioSegment.from_file(file_path, format=ext)
else:
audio = AudioSegment.from_file(file_path)
# Convert to WAV 16kHz mono
audio = audio.set_frame_rate(16000)
audio = audio.set_channels(1)
# Save as WAV with proper cleanup
temp_wav = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
temp_wav.close() # Close the file handle before pydub writes to it
audio.export(temp_wav.name, format="wav")
return temp_wav.name
except Exception as e:
# Clean up temp file if it exists
if temp_wav and os.path.exists(temp_wav.name):
try:
os.unlink(temp_wav.name)
except OSError:
pass
raise Exception(f"Failed to extract audio: {e!s}") from e
# Formatting/export helpers live in stt/formatting.py (importable, unit-tested);
# names are re-imported here so existing call sites stay unchanged.
from stt import formatting as _formatting
from stt.formatting import ( # noqa: F401
apply_word_highlighting_server,
convert_db_to_translation_srt,
format_file_size,
format_timestamp_srt,
format_timestamp_vtt,
format_transcription,
)
def _word_highlighting_config_path():
return os.path.join(CONFIG_DIR, "word_highlighting.json")
def convert_db_to_srt(db_path):
"""Convert a Transcriptions.db to SRT (plus HTML export if enabled in settings)."""
# Reload config to get fresh settings (global config may be stale)
html_enabled = load_config().get("database", {}).get("html_enabled", True)
return _formatting.convert_db_to_srt(db_path, html_enabled=html_enabled, highlight_config_path=_word_highlighting_config_path())
def convert_db_to_html(db_path):
"""Convert a Transcriptions.db to HTML with word highlighting."""
return _formatting.convert_db_to_html(db_path, highlight_config_path=_word_highlighting_config_path())
app = Flask(__name__,
template_folder=os.path.join(BUNDLE_DIR, "templates"),
static_folder=os.path.join(BUNDLE_DIR, "static"))
app.config["SECRET_KEY"] = os.environ.get("STT_SECRET_KEY") or secrets.token_urlsafe(32)
app.config["MAX_CONTENT_LENGTH"] = 4 * 1024 * 1024 * 1024 # 4GB cap on uploads (media files are large)
app.config["TEMPLATES_AUTO_RELOAD"] = True # Auto-reload templates when they change
app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 0 # Disable caching for static files
socketio = SocketIO(app, async_mode="threading", static_url_path="/static", static_folder=os.path.join(BUNDLE_DIR, "static"), ping_timeout=120, ping_interval=25)
# A binary broadcast (audio_chunk_hq) shares one packet between websocket and
# polling listeners; engineio caches its first encoding for both. STT-2R.
import engineio.packet as _engineio_packet # noqa: E402
_engineio_compat.install(_engineio_packet.Packet)
# --- Access log ------------------------------------------------------------
# Records every HTTP request (tagged web vs api), WebSocket connect/disconnect,
# and SocketIO action, into a size-capped SQLite table under APP_DIR/logs.
# Viewed at /logs (page) and /api/logs (JSON). Thin wrapper over stt.request_log.
from stt import request_log as _request_log # noqa: E402
from stt import repeat_filter as _repeat_filter # noqa: E402
from stt import file_backup as _file_backup # noqa: E402
from stt import metrics as _metrics # noqa: E402
from stt import audio_file as _audio_file # noqa: E402
from stt.hypothesis_buffer import LocalAgreementBuffer # noqa: E402
try:
os.makedirs(os.path.join(APP_DIR, "logs"), exist_ok=True)
request_logger = _request_log.RequestLog(
os.path.join(APP_DIR, "logs", "access_log.db"),
max_rows=int((config.get("access_log", {}) or {}).get("max_rows", 50000) or 50000),
)
except Exception:
# A logging store failure must never stop the server from booting.
request_logger = None
# Reverse-DNS names for the IPs in the log ("who is 192.168.2.62?"). Lookups run
# on a background thread and are cached, so the logs endpoints never block on a
# LAN address with no PTR record.
from stt.hostname_lookup import HostnameCache # noqa: E402
hostname_cache = HostnameCache()
def _access_log_enabled():
"""Whether request logging is currently on (default True). Read fresh so a
config change via /api/config takes effect without a restart."""
try:
return bool((config.get("access_log", {}) or {}).get("enabled", True))
except Exception:
return False
def _access_log_skip_polling():
"""Whether to drop dashboard-polling requests before they are written.
The log already knows these paths are noise — /api/logs filters them out via
POLLING_LOG_PATHS whenever hide_polling is set, which the logs page sends by default.
Writing and fsyncing a row only to hide it again is pure cost: an operator with the
health and live-settings pages open generates around 150 rows a minute, and they
displace the real requests inside the 50,000-row cap. Set false to log them anyway
when diagnosing the polling itself. Read fresh so a config change takes effect
without a restart."""
try:
return bool((config.get("access_log", {}) or {}).get("skip_polling_paths", True))
except Exception:
return True
def _record_socket_event(event_name, kind, refused=False):
"""Log a SocketIO connect/disconnect or action. Best-effort, never raises."""
if request_logger is None or not _access_log_enabled():
return
try:
sid = getattr(request, "sid", None)
origin = request_origin.classify(
getattr(request, "remote_addr", None),
getattr(request, "headers", None),
)
bits = [f"sid={sid}"] if sid else []
if origin.ray_id:
bits.append(f"ray={origin.ray_id}")
if refused:
bits.append("refused: tunnel connection not authenticated")
request_logger.log(
source=_request_log.SOURCE_SOCKET,
kind=kind,
path=event_name,
# The real visitor, not our loopback: every tunnelled socket would
# otherwise log as 127.0.0.1 and be indistinguishable.
ip=origin.client_ip or None,
origin=origin.kind,
status=403 if refused else None,
user_agent=request.headers.get("User-Agent") if request else None,
detail=("; ".join(bits) if bits else None),
)
except Exception:
pass
# Origin of each live socket connection, by sid. The websocket has no auth of
# its own, so a tunnelled connection would otherwise be able to rewrite
# transcripts and retime live output — gating HTTP alone would be theatre.
# Recorded at connect (the handshake carries the CF headers and the auth cookie)
# because later events cannot be re-classified: engineio may deliver them over a
# polling request whose headers differ from the upgrade's.
_socket_origins = {}
_socket_origins_lock = threading.Lock()
def _remember_socket_origin():
"""Classify a connecting socket and note whether it authenticated."""
sid = getattr(request, "sid", None)
if sid is None:
return
origin = request_origin.classify(request.remote_addr, request.headers)
# check_ip_whitelist reads the same cookie/Bearer session the HTTP side uses,
# so a visitor who logged in over the tunnel keeps their socket privileges.
authed = check_ip_whitelist()
with _socket_origins_lock:
_socket_origins[sid] = (origin, authed)
if origin.is_tunnelled:
print(f"[SOCKET] Tunnelled connection {sid} from {origin.client_ip} (authenticated={authed})")
def _forget_socket_origin():
sid = getattr(request, "sid", None)
if sid is not None:
with _socket_origins_lock:
_socket_origins.pop(sid, None)
def _socket_event_allowed(event):
"""Whether the current socket connection may run this event.
Unknown sids fail closed only for gated events — a reconnect that raced the
connect handler still gets the display feed.
"""
sid = getattr(request, "sid", None)
with _socket_origins_lock:
entry = _socket_origins.get(sid)
if entry is None:
return True
origin, authed = entry
if authed:
return True
return not request_origin.socket_event_requires_auth(event, origin.kind)
def _handler_accepts_args(handler):
"""Whether ``handler`` can take a positional argument.
A SocketIO event carries whatever payload the client chose to attach, and
most of our handlers declare no parameters because the display emits those
events bare. Nothing stops another client attaching data, and the library
passes it straight through — which raised TypeError out of a socket thread
for an event that is otherwise perfectly valid. Asking the signature once is
cheaper and safer than catching TypeError around the call, which would also
swallow a genuine TypeError raised inside the handler itself.
"""
try:
params = inspect.signature(handler).parameters.values()
except (TypeError, ValueError): # pragma: no cover - builtins/C callables
return True # unknown shape: pass the payload through unchanged
return any(
p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD, p.VAR_POSITIONAL)
for p in params
)
# Wrap socketio.on so every @socketio.on(...) handler registered below is
# transparently logged — connect/disconnect as connections, everything else as
# actions — and so state-changing events are gated for unauthenticated tunnel
# connections. Installed before any handler is defined so all of them are covered.
_socketio_on_orig = socketio.on
def _logging_socketio_on(message, namespace=None):
decorator = _socketio_on_orig(message, namespace=namespace)
_kind = _request_log.KIND_CONNECTION if message in ("connect", "disconnect") else _request_log.KIND_ACTION
def register(handler):
# Whether this handler can accept the payload a client may attach to the
# event. Most take none — the display emits request_all_entries with no
# data — but nothing stops any client sending some, and passing it to a
# zero-arg handler raised an unhandled TypeError out of a socket thread.
# Computed once at registration, so there is no per-event cost.
_accepts_payload = _handler_accepts_args(handler)
@functools.wraps(handler)
def instrumented(*args, **kwargs):
if not _accepts_payload:
# Drop what the handler cannot take rather than crashing on it.
args = ()
if message == "connect":
_remember_socket_origin()
if not _socket_event_allowed(message):
_record_socket_event(message, _kind, refused=True)
# Tell the caller rather than dropping it silently: a display
# that quietly stopped saving corrections would look like a bug.
emit("auth_required", {
"event": message,
"error": "Log in with the server password to make changes over the tunnel.",
})
return None
_record_socket_event(message, _kind)
try:
return handler(*args, **kwargs)
finally:
if message == "disconnect":
_forget_socket_origin()
return decorator(instrumented)
return register
socketio.on = _logging_socketio_on
# ---------------------------------------------------------------------------
# When this (web) process started — used for the Server Settings uptime display.
# Resets on restart/redeploy/crash-recovery, reflecting the live server-process lifetime.
SERVER_START_TIME = time.time()
# Running version (read once at startup; a self-update restarts the process, so a
# fresh run picks up the new VERSION). Same source as the Sentry release tag.
try:
with open(os.path.join(BUNDLE_DIR, "VERSION"), encoding="utf-8") as _vf:
SERVER_VERSION = _vf.read().strip()
except OSError:
SERVER_VERSION = "unknown"
# Exact git commit of this checkout (empty for frozen/installer builds with no .git).
# Recorded in the live-map ping alongside SERVER_VERSION; recomputed on self-update re-exec.
try:
from stt.self_update import git_commit as _git_commit
SERVER_COMMIT = _git_commit(BUNDLE_DIR)
except Exception:
SERVER_COMMIT = ""
# Human version of the running checkout, e.g. '26.1.2-9-gc588d29' (tag + commits
# since + commit). The VERSION file only changes on releases, so on a git deploy
# it understates what's actually running; empty for frozen builds.
try:
from stt.self_update import git_describe as _git_describe
SERVER_DESCRIBE = _git_describe(BUNDLE_DIR)
except Exception:
SERVER_DESCRIBE = ""
# Whether this checkout has uncommitted work, i.e. someone is developing on it.
# Sent as src=dev so the live map can hide maintainer machines from a map that is
# meant to show real installs; until this existed STT sent no src at all and every
# dev box counted as a church.
try:
from stt.self_update import is_dev_checkout as _is_dev_checkout
SERVER_IS_DEV = _is_dev_checkout(BUNDLE_DIR)
except Exception:
SERVER_IS_DEV = False
def _compute_display_version():
"""Single monotonic version string for scripts and the UI (see stt/config_utils.py)."""
return _config_utils.compute_display_version(SERVER_DESCRIBE, SERVER_COMMIT, SERVER_VERSION)
SERVER_DISPLAY_VERSION = _compute_display_version()
def _os_version_string():
"""The specific OS release for the live-map ping ("15.5", "Ubuntu 24.04.1 LTS").
Reading it is the IO half of stt.livemap.os_version_label: platform's own accessors
plus, on Linux, /etc/os-release — the kernel version says nothing useful about a
distro, so PRETTY_NAME is preferred and the kernel is only the fallback.
"""
import platform as _platform
distro = ""
try:
with open("/etc/os-release", encoding="utf-8") as f:
for line in f:
if line.startswith("PRETTY_NAME="):
distro = line.partition("=")[2].strip().strip('"')
break
except OSError:
pass # not Linux, or an image without it — the kernel fallback covers it
return _livemap.os_version_label(
sys.platform,
mac_ver=_platform.mac_ver()[0],
win_ver=_platform.win32_ver()[1],
distro=distro,
kernel=_platform.release())
# Static for the life of the process, and both ping events carry them.
try:
import platform as _platform_mod
SERVER_OS_VERSION = _os_version_string()
SERVER_ARCH = _livemap.arch_label(_platform_mod.machine())
except Exception:
SERVER_OS_VERSION = ""
SERVER_ARCH = ""
def _remote_ping_provenance():
"""``(model, method)`` of the paired machine, or ("", "") when not offloading.
Only those two of the provenance keys: the same probe returns the remote's LLM
endpoint and system prompt, which are the operator's own infrastructure and must
never leave the building.
The cache is preferred and a probe is the fallback, because it is empty until
something has asked — which on a boot-time app_start ping is always. Affording the
probe is what makes an idle offloading box report its peer at all; it costs a
5-second timeout on a thread whose whole job is fire-and-forget.
"""
try:
remote = (config.get("live_translation") or {}).get("remote") or {}
if not (remote.get("enabled") and remote.get("endpoint")):
return "", ""
provenance = _remote_effective or _fetch_remote_provenance()
return (provenance.get("mt.remote.effective.model", ""),
provenance.get("mt.remote.effective.method", ""))
except Exception:
return "", "" # an unreachable peer still leaves an install worth counting
def _send_livemap_ping(event, **fields):
"""Send one anonymous live-map ping. Returns whether the collector was reached.
Call this on a daemon thread, never inline: it makes a network request, and neither
a server boot nor a transcription start may wait on a collector that is slow, down,
or unreachable because the box has no route out at all. Every failure is one printed
line — the ping is the least important thing this process does.
The URL and the shape of the two events live in stt/livemap.py; what stays here is
the part that cannot be unit-tested without a network: the config read, the install
id, the hardware probe, and the request itself.
"""
try:
endpoint = (config.get("analytics", {}) or {}).get("endpoint")
# No collector, no probe: an opted-out install does no analytics work at all,
# not even the LAN round trip to its own peer.
_remote_model, _remote_method = (_remote_ping_provenance()
if str(endpoint or "").strip() else ("", ""))
url = _livemap.build_ping_url(
endpoint,
event=event,
os_name=_livemap.os_name_for_platform(sys.platform),
version=_livemap.numeric_version(SERVER_DISPLAY_VERSION, SERVER_VERSION),
commit=SERVER_COMMIT,
os_version=SERVER_OS_VERSION,
arch=SERVER_ARCH,
# Cached after the first probe; on a CPU-only box this is blank, which is
# itself the answer to half the "why is it slow" questions.
gpu=_livemap.gpu_label(_probe_hardware().get("gpu_name") or ""),
# A trial is not an install. The collector filters on this, so a demo
# that failed to say so would be counted as a church running the app.
src="demo" if DEMO else ("dev" if SERVER_IS_DEV else ""),
as_root=getattr(os, "geteuid", None) is not None and os.geteuid() == 0,
**_livemap.install_fields_from_config(config, remote_model=_remote_model,
remote_method=_remote_method),
**fields)
if url is None:
return False # blank endpoint: the operator has opted out
import requests as _req
_req.get(url, headers={"X-Install-Id": _get_install_id()}, timeout=10)
return True
except Exception as e:
print(f"[LIVEMAP] Ping failed ({event}): {e}")
return False
def _ping_transcription_started():
"""Tell the live map a session has begun, from whichever path began it.
Four places put {"command": "start"} on the control queue, and for a long time
only the /api/transcription/start route pinged. An install running with
audio.autostart therefore reported itself alive at every boot and never once
reported a service — precisely inverted for the unattended installs that caption
every service they are used for. Calibration is deliberately excluded: it starts a
real session, but to set levels, and the map counts services.
"""
threading.Thread(
target=_send_livemap_ping,
args=(_livemap.EVENT_TRANSCRIPTION_START,),
kwargs=_livemap.ping_fields_from_config(config),
daemon=True, name="livemap-transcription-start").start()
def _livemap_app_start_worker():
"""The app-start ping, with one retry for a box that boots faster than its network.
.62 starts under systemd at boot, where DNS is routinely not up yet when this
thread runs. A single attempt would fail there nearly every time and the app-start
signal would be missing from exactly the installs that run unattended, so a failure
is retried once a minute later. One retry, not a schedule: if the network is still
down after a minute the machine has a real problem, and the next restart pings.
"""
if _send_livemap_ping(_livemap.EVENT_APP_START):
return
# Event rather than sleep: a shutdown during the wait should not be held up by it.
threading.Event().wait(60)
_send_livemap_ping(_livemap.EVENT_APP_START)
# --- System requirements (informational warning shown in the web UI header) ---
# Requirements are estimated from the configured models (live Whisper, file
# transcription Whisper, local NLLB) rather than static tiers, and compared
# against detected hardware: RAM, CUDA VRAM, and Apple Silicon unified memory
# (where CPU and GPU share one pool). Every probe and lookup fails open
# (unknown -> no warning) and must never block startup.
# Model-memory tables + pure estimation/requirements logic live in
# stt/model_memory.py (importable, tested). Hardware probing (below) stays here
# and is injected via the shims further down.
from stt.model_memory import ( # noqa: F401
MODEL_MEMORY_ESTIMATES,
estimate_memory_requirements,
check_system_requirements,
)
def _get_total_ram_bytes():
"""Total physical RAM in bytes, stdlib only; 0 when unknown (no false warning)."""
try:
if sys.platform == "win32":
import ctypes
class _MemStatus(ctypes.Structure):
_fields_ = [
("dwLength", ctypes.c_uint32),
("dwMemoryLoad", ctypes.c_uint32),
("ullTotalPhys", ctypes.c_uint64),
("ullAvailPhys", ctypes.c_uint64),
("ullTotalPageFile", ctypes.c_uint64),
("ullAvailPageFile", ctypes.c_uint64),
("ullTotalVirtual", ctypes.c_uint64),
("ullAvailVirtual", ctypes.c_uint64),
("ullAvailExtendedVirtual", ctypes.c_uint64),
]
status = _MemStatus()
status.dwLength = ctypes.sizeof(_MemStatus)
if ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(status)):
return int(status.ullTotalPhys)
return 0
if sys.platform == "darwin":
import subprocess
r = subprocess.run(["sysctl", "-n", "hw.memsize"],
capture_output=True, text=True, timeout=5)
return int(r.stdout.strip()) if r.returncode == 0 else 0
return os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE")
except Exception:
return 0
def _probe_gpu():
"""``(name, vram_bytes)`` for CUDA device 0; either half is None when unknown.
Name and VRAM come from the same source in one shot: the requirements check needs
the size, the live-map ping needs the model, and asking nvidia-smi twice would pay
the subprocess twice for one answer.
Never triggers the heavy torch import itself: uses torch only when it is
already in sys.modules, otherwise falls back to nvidia-smi (stdlib-only,
works at startup before the lazy ML import runs).
"""
try:
if "torch" in sys.modules:
import torch
if torch.cuda.is_available():
props = torch.cuda.get_device_properties(0)
return getattr(props, "name", None), int(props.total_memory)
return None, None # torch is loaded and says no CUDA — trust it
except Exception:
pass
try:
import subprocess
r = subprocess.run(["nvidia-smi", "--query-gpu=name,memory.total", "--format=csv,noheader,nounits"],
capture_output=True, text=True, timeout=5,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
if r.returncode == 0 and r.stdout.strip():
name, _, total = r.stdout.strip().splitlines()[0].partition(",")
return name.strip() or None, int(float(total)) * 1024**2
except Exception:
pass
return None, None
def _probe_apple_gpu_name():
"""The Apple Silicon chip ("Apple M2 Pro"), or None. The GPU is part of the SoC."""
try:
import subprocess
r = subprocess.run(["sysctl", "-n", "machdep.cpu.brand_string"],
capture_output=True, text=True, timeout=5)
if r.returncode == 0:
return r.stdout.strip() or None
except Exception:
pass
return None
_HW_PROBE_CACHE = None
_HW_PROBE_LOCK = threading.Lock()
def _probe_hardware():
"""Cached hardware snapshot; every field fails open (0 / None / False).
VRAM is re-probed on later calls if it was unknown and torch has since
been imported (the lazy ML import can happen after startup).
"""
global _HW_PROBE_CACHE
with _HW_PROBE_LOCK:
hw = _HW_PROBE_CACHE
if hw is None:
try:
import platform as _platform # module-level `platform` is sys.platform
apple_silicon = sys.platform == "darwin" and _platform.machine() == "arm64"
except Exception:
apple_silicon = False
gpu_name, vram = _probe_gpu()
hw = {
"ram_bytes": _get_total_ram_bytes(),
"cpu_cores": os.cpu_count() or 0,
"vram_bytes": vram,
"has_cuda": vram is not None,
"apple_silicon": apple_silicon,
# Reported to the live map, not used for the requirements check. On a
# Mac the accelerator is the SoC, so the chip name is the GPU name.
"gpu_name": gpu_name or (_probe_apple_gpu_name() if apple_silicon else None),
}
_HW_PROBE_CACHE = hw
elif hw["vram_bytes"] is None and "torch" in sys.modules:
gpu_name, vram = _probe_gpu()
hw["vram_bytes"] = vram
hw["has_cuda"] = vram is not None
hw["gpu_name"] = gpu_name or hw.get("gpu_name")
try:
import shutil
hw["disk_free_bytes"] = shutil.disk_usage(APP_DIR).free # cheap; keep current
except Exception:
hw["disk_free_bytes"] = 0
return hw
def _estimate_memory_requirements(cfg, gpu_available=False, unified=False):
"""Thin shim: pure math lives in stt.model_memory; inject MODELS_DIR."""
return estimate_memory_requirements(cfg, MODELS_DIR, gpu_available=gpu_available, unified=unified)
def _check_system_requirements(cfg=None, hw=None):
"""Thin shim: default cfg/hw to the live config + cached probe (both IO),
then delegate to the pure stt.model_memory.check_system_requirements."""
if cfg is None:
cfg = config
if hw is None:
hw = _probe_hardware()
return check_system_requirements(cfg, hw, MODELS_DIR)
SYSTEM_REQUIREMENTS_WARNINGS = _check_system_requirements()
if SYSTEM_REQUIREMENTS_WARNINGS:
print("[SYSREQ] Below minimum system requirements: "
+ "; ".join(SYSTEM_REQUIREMENTS_WARNINGS))
app_logger = logging.getLogger(__name__) # Use your module name here
socket_io_logger = logging.getLogger("socketio")
# Polled endpoints that log an unchanging value would otherwise ship the same
# record every minute forever (Sentry Logs are unsampled). Gate them on the
# value actually changing.
from stt.log_gate import ChangeGate, is_benign_wsgi_message, is_stale_session_message # noqa: E402
_poll_log_gate = ChangeGate()
# Set log levels as needed
app_logger.setLevel(logging.DEBUG)
socket_io_logger.setLevel(logging.WARNING)
# Disable Flask's built-in logging
log = logging.getLogger("werkzeug")
log.setLevel(logging.ERROR)
class _SuppressBenignWSGINoise(logging.Filter):
"""Drop werkzeug records that report no fault of ours.
Which messages, and why a logger filter is the right place, are documented
in stt/log_gate.py — the short version is that this runs before the hook
sentry-sdk uses to turn ERROR records into issues, so it stops the crash
report too. All other werkzeug errors pass through unchanged."""
def filter(self, record):
try:
message = record.getMessage()
except Exception:
return True
return not is_benign_wsgi_message(message)
log.addFilter(_SuppressBenignWSGINoise())
class _StaleSessionAtInfo(logging.Filter):
"""Log a browser reusing a closed Socket.IO session at INFO, not ERROR.
engine.io already does that for every occurrence after the first; the first
carries the sid in its message, so each server start filed a fresh Sentry
issue for a client doing the right thing. See stt/log_gate.py."""
def filter(self, record):
try:
message = record.getMessage()
except Exception:
return True
if record.levelno > logging.INFO and is_stale_session_message(message):
record.levelno, record.levelname = logging.INFO, "INFO"
return True
logging.getLogger("engineio.server").addFilter(_StaleSessionAtInfo())
# Password-based authentication sessions
# Format: {session_token: {"ip": client_ip, "expires": datetime}}
auth_sessions = {}
auth_sessions_lock = threading.Lock()
# Per-IP login throttle: after LOGIN_MAX_FAILURES wrong passwords, that IP is
# locked out for LOGIN_LOCKOUT_SECONDS. Bounds online brute-forcing of the
# configured password. Format: {ip: {"failures": int, "locked_until": datetime}}
_login_attempts = {}
_login_attempts_lock = threading.Lock()
LOGIN_MAX_FAILURES = 5
LOGIN_LOCKOUT_SECONDS = 30
def generate_session_token():
"""Generate a secure random session token"""
import secrets
return secrets.token_urlsafe(32)
def cleanup_expired_sessions():
"""Remove expired sessions from the auth_sessions dict"""
now = datetime.now()
with auth_sessions_lock:
expired = [token for token, data in auth_sessions.items() if data["expires"] < now]
for token in expired:
del auth_sessions[token]
if expired:
print(f"[AUTH] Cleaned up {len(expired)} expired sessions")
@app.route("/")
def index():
# Check if URL has any parameters
if not request.args:
# No parameters provided: redirect to the active URL-builder profile (if any)
profiles, active = get_url_builder_profiles()
params = next((p["params"] for p in profiles if p["name"] == active), None)
if params:
from flask import redirect, url_for
return redirect(url_for('index', **params))
response = make_response(render_template("index.html"))
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
response.headers['Pragma'] = 'no-cache'
return response
@app.route("/profile/<name>")
def profile_view(name):
"""Display the page using a named profile's settings, e.g. /profile/lower3rd.
Case-insensitive; unknown names fall back to the root view."""
from flask import redirect, url_for
profiles, _ = get_url_builder_profiles()
target = (name or "").strip().lower()
params = next((p["params"] for p in profiles if p["name"].strip().lower() == target), None)
if params is None:
return redirect("/")
return redirect(url_for("index", **params))
@app.route("/favicon.ico")
def favicon():
# Return a custom response or an empty response
return "", 204
@app.errorhandler(404)
def page_not_found(e):
"""Handle 404 errors by redirecting to appropriate page"""
if check_ip_whitelist():
return redirect("/live-settings")
else:
return redirect("/")
def _control_params(keep_blank=False):
"""Request parameters for an endpoint, from wherever the client sent them.
Thin wrapper over stt.http_params.merge_request_params: a show-control system
may send a JSON body, a form body, or a query string, and rejecting two of the
three looks to the operator like a button that did nothing. Precedence is
JSON > form > query.
``keep_blank=False`` (switch endpoints) treats a blank value as "not sent", so
a surface posting a fixed field set on every press can't blank a live setting.
``keep_blank=True`` (settings endpoints) keeps blanks, because an empty string
is how those clear a field.
Do NOT use this on a route that merges the body wholesale into config — it
unions the query string in, and a URL can carry ?key=<access_token>, which
would be persisted as a config key. /api/config reads its JSON directly for
exactly that reason.
"""
body = request.get_json(silent=True)
if body is None and not request.form:
# Well-formed JSON under the wrong Content-Type: Flask's get_json returns
# None and request.form stays empty (it only parses form content types),
# so the body is invisible and the route would 400 as though nothing was
# sent. Read it off the raw bytes as a last resort.
try:
body = _parse_json_body(request.get_data(as_text=True))
except Exception:
body = None
return merge_request_params(body, request.form, request.args, keep_blank=keep_blank)
def _request_origin():
"""Where this request came from (local / lan / tunnel) and its real client IP.
Cached per request: check_ip_whitelist, the access log and the after_request
hook all want it, and re-reading headers three times invites them to drift.
"""
origin = getattr(g, "_request_origin", None)
if origin is None:
origin = request_origin.classify(request.remote_addr, request.headers)
try:
g._request_origin = origin
except Exception: # pragma: no cover - outside an app context
pass
return origin
def check_ip_whitelist():
"""Check if the client IP is in the whitelist or has a valid password session"""
# A demo binds to loopback and holds no real data — every page is meant to be
# walked through. Granting here also covers Socket.IO, which authorises through
# this same function.
if DEMO:
return True
import ipaddress
# First check if password authentication is enabled
password_auth_config = config.get("web_server", {}).get("password_auth", {})
password_auth_enabled = password_auth_config.get("enabled", False)
access_token = str(config.get("web_server", {}).get("access_token", "") or "")
# A tunnelled request reaches us from our own loopback, so neither the
# localhost shortcut nor an empty whitelist may speak for it. Only a password
# session does. See stt/request_origin for why the CF headers can be believed.
origin = _request_origin()
transport_trusted = request_origin.is_trusted_transport(origin)
# A request carrying ?key=<access_token> is granted directly, so a
# non-whitelisted device (or an OBS/display source that can't use the login
# page) can embed auth in the URL. _mint_access_token_cookie (after_request)
# then sets a session cookie so the rest of that browser session works
# without re-passing the key. Empty token = disabled.
#
# Not honoured over the tunnel: a URL key would travel in links, browser
# history and screenshots, and the public surface is exactly where that
# leaks. Password login only from there.
if access_token and transport_trusted:
provided_key = request.args.get("key", "")
if provided_key and secrets.compare_digest(provided_key, access_token):
return True
# Check for a valid session token (from password login OR a minted
# access-token session) via cookie or Authorization header. A tunnelled
# caller must get here — it is the only door left open to it.
if password_auth_enabled or access_token or origin.is_tunnelled:
# Check cookie first
session_token = request.cookies.get("auth_session")
# Fallback to Authorization header
if not session_token:
auth_header = request.headers.get("Authorization")
if auth_header and auth_header.startswith("Bearer "):
session_token = auth_header[7:] # Remove "Bearer " prefix
# Validate session token
if session_token:
cleanup_expired_sessions() # Clean up expired sessions
with auth_sessions_lock:
session_data = auth_sessions.get(session_token)
if session_data:
# Check if session is still valid
if session_data["expires"] > datetime.now():
# Check if IP matches (prevent session hijacking).
# origin.client_ip, not remote_addr: every tunnelled
# session would otherwise bind to 127.0.0.1, which is
# the same value for every visitor — no protection at all.
if session_data["ip"] == origin.client_ip:
return True
else:
print(f"[AUTH WARNING] Session token used from different IP: {origin.client_ip} != {session_data['ip']}")
else:
# Session expired, remove it
del auth_sessions[session_token]
# Past this point access is granted on the caller's address alone, which a
# tunnelled request must never get: its address is our loopback.
if not transport_trusted:
return False
whitelist = config.get("web_server", {}).get("settings_ip_whitelist", [])
# If whitelist is empty, allow all
if not whitelist:
return True
client_ip = request.remote_addr
# Always allow localhost variations
localhost_ips = ["127.0.0.1", "::1", "localhost"]
if client_ip in localhost_ips:
return True
# Check whitelist (supports both exact IPs and CIDR ranges)
try:
client_ip_obj = ipaddress.ip_address(client_ip)
for entry in whitelist:
entry = entry.strip()
if not entry or entry.startswith("#"): # Skip comments
continue
try:
# Check if entry is a CIDR range (contains /)
if "/" in entry:
network = ipaddress.ip_network(entry, strict=False)
if client_ip_obj in network:
return True
else:
# Exact IP match
if client_ip == entry:
return True
except ValueError:
# Invalid entry, skip it
print(f"[WARNING] Invalid IP whitelist entry: {entry}")
continue
return False
except ValueError:
# Invalid client IP format
print(f"[WARNING] Invalid client IP format: {client_ip}")
return False
@app.before_request
def _access_log_start_timer():
"""Stamp a monotonic start time so the after_request hook can report the
request duration. Best-effort — never breaks a request."""
try:
g._access_log_t0 = time.perf_counter()
except Exception:
pass
if DEMO:
from stt import demo_api as _demo_api
_demo_api_state = _demo_api.State()
def _demo_player_ref():
return globals().get("_demo_player")
# Registered after the access-log timer so a demo request is still logged, and
# before dispatch so the production view is never entered: several of those views
# import huggingface_hub or shell out to uv/git, neither of which exists in a demo
# build. See stt/demo_api.py for what is answered and what is passed through.
@app.before_request
def _demo_api_intercept():
try:
canned = _demo_api.intercept(
request.method, request.path, request.args.to_dict(),
request.get_json(silent=True) or {},
_demo_api_state, player=_demo_player_ref())
except Exception as exc: # pragma: no cover - a demo must not 500
print(f"[DEMO] intercept failed for {request.path}: {exc}")
return None
if canned is None:
return None
payload, status = canned
return jsonify(payload), status
def _note_access_detail(detail):
"""Attach a short summary to this request's access-log row.
Routes opt in and pass only a curated description of what they did — never the
raw parameter map, which can contain ?key=<access_token> picked up from the
query string. That is the same reason the hook below drops query strings.
Exists so a control-surface press is self-explaining afterwards: the row alone
shows whether the request arrived, what it asked for, and whether anything
actually changed. Best-effort — never breaks a request."""
try:
g._access_log_detail = str(detail)[:300]
except Exception:
pass
# Statuses worth thinning when one client repeats them. A refusal repeats
# identically for as long as whatever is being refused keeps asking: an
# unauthorised page polling (403), or a peer polling an endpoint this build does
# not have (404). Everything else stays fully logged — a 400 differs run to run
# and is what someone reads to work out why.
_REFUSAL_STATUSES = (401, 403, 404)
_refusal_log_filter = _repeat_filter.RepeatSuppressor()
#: Why a paired-peer request was refused — see _log_pair_refusal. Its own
#: suppressor, because a peer retrying every 20s is a different rhythm from a
#: browser polling, and one must not use up the other's burst allowance.
_pair_refusal_log_filter = _repeat_filter.RepeatSuppressor()
def _refusal_log_verdict(path, status, ip):
"""Whether to write this row, given how often the same client has seen it.
A page open on a machine that is not whitelisted is refused every time it
polls, forever — 60 rows an hour, each one displacing a real request inside
the log's row cap. The first few are kept, then one per cooldown carrying the
count of what it stands for, so the flood is still visible in a line rather
than in six hundred.
"""
if status not in _REFUSAL_STATUSES:
return _repeat_filter.Verdict(True)
return _refusal_log_filter.decide((ip, request.method, path, status))
@app.after_request
def _access_log_record(response):
"""Log the request (tagged web vs api). Skips static assets and socket.io
transport frames — WebSocket traffic is logged via the socket handlers.
Best-effort — never breaks a response.
The query string is deliberately dropped: it can carry ?key=<access_token>,
which must not be written to the log. Routes that want context in the row
call _note_access_detail() with a curated summary instead.
A refusal repeated by the same client is thinned to a heartbeat row — see
_refusal_log_verdict."""
try:
if request_logger is not None and _access_log_enabled():
path = request.path or ""
if not (path.startswith("/static/") or path.startswith("/socket.io/")
or (_access_log_skip_polling() and path in POLLING_LOG_PATHS)):
origin = _request_origin()
# Thin repeats per real visitor: over the tunnel every caller
# shares remote_addr 127.0.0.1, so one noisy client would
# suppress everyone else's rows.
verdict = _refusal_log_verdict(path, response.status_code, origin.client_ip)
if not verdict.log:
return response
t0 = getattr(g, "_access_log_t0", None)
duration_ms = round((time.perf_counter() - t0) * 1000.0, 2) if t0 is not None else None
detail = getattr(g, "_access_log_detail", None)
if verdict.suppressed:
repeats = f"+{verdict.suppressed} identical since the last row"
detail = f"{detail} ({repeats})" if detail else repeats
if origin.ray_id:
# Cloudflare's own request id — what makes a row traceable
# back to their logs if something needs chasing.
detail = f"{detail}; ray={origin.ray_id}" if detail else f"ray={origin.ray_id}"
request_logger.log(
source=_request_log.classify_source(path),
kind=_request_log.KIND_HTTP,
method=request.method,
path=path,
status=response.status_code,
ip=origin.client_ip,
origin=origin.kind,
user_agent=request.headers.get("User-Agent"),
duration_ms=duration_ms,
detail=detail,
)
except Exception:
pass # request logging must never break a response
return response
@app.after_request
def _mint_access_token_cookie(response):
"""Turn a valid ?key=<access_token> request into a browser session.
A display/OBS source (or any device) can embed ?key=<token> in the URL to
get in; this mints an IP-bound auth_session cookie on that response so the
page's subsequent requests (assets, socket.io, navigation) authenticate
without re-passing the key. Best-effort — never breaks a response."""
try:
access_token = str(config.get("web_server", {}).get("access_token", "") or "")
if not access_token or request.cookies.get("auth_session"):
return response
# check_ip_whitelist refuses ?key= over the tunnel, so minting a session
# from one here would reopen exactly the door it closes.
if not request_origin.is_trusted_transport(_request_origin()):
return response
provided_key = request.args.get("key", "")
if not (provided_key and secrets.compare_digest(provided_key, access_token)):
return response
timeout_minutes = int(config.get("web_server", {}).get("password_auth", {})
.get("session_timeout_minutes", 60) or 60)
token = generate_session_token()
with auth_sessions_lock:
auth_sessions[token] = {
"ip": _request_origin().client_ip,
"expires": datetime.now() + timedelta(minutes=timeout_minutes),
"created": datetime.now(),
}
response.set_cookie("auth_session", token, max_age=timeout_minutes * 60,
httponly=True, samesite="Strict")
except Exception:
pass # auth-cookie minting must never break a response
return response
@app.route("/api/auth/login", methods=["POST"])
def password_login():
"""Authenticate with password and create a temporary session"""
try:
password_auth_config = config.get("web_server", {}).get("password_auth", {})
# Check if password auth is enabled
if not password_auth_config.get("enabled", False):
return jsonify({"success": False, "error": "Password authentication is disabled"}), 403
# Reject early if this IP is currently locked out
client_ip = _request_origin().client_ip
with _login_attempts_lock:
entry = _login_attempts.get(client_ip)
if entry and entry.get("locked_until") and entry["locked_until"] > datetime.now():
retry_after = int((entry["locked_until"] - datetime.now()).total_seconds()) + 1
return jsonify({
"success": False,
"error": "Too many failed attempts. Try again later.",
}), 429, {"Retry-After": str(retry_after)}
# Get password from request
data = request.get_json()
provided_password = data.get("password", "")
if not provided_password:
return jsonify({"success": False, "error": "Password required"}), 400
# Get configured password
configured_password = password_auth_config.get("password", "")
# If no password configured, reject
if not configured_password:
return jsonify({"success": False, "error": "No password configured on server"}), 500
# Verify password (constant-time to avoid leaking length/prefix via timing)
if not secrets.compare_digest(str(provided_password), str(configured_password)):
print(f"[AUTH] Failed login attempt from {client_ip}")
with _login_attempts_lock:
rec = _login_attempts.setdefault(client_ip, {"failures": 0, "locked_until": None})
rec["failures"] += 1
if rec["failures"] >= LOGIN_MAX_FAILURES:
rec["locked_until"] = datetime.now() + timedelta(seconds=LOGIN_LOCKOUT_SECONDS)
rec["failures"] = 0
return jsonify({"success": False, "error": "Invalid password"}), 401
# Password correct — clear any failure record for this IP
with _login_attempts_lock:
_login_attempts.pop(client_ip, None)
# Password correct - create session
session_token = generate_session_token()
timeout_minutes = password_auth_config.get("session_timeout_minutes", 60)
expires = datetime.now() + timedelta(minutes=timeout_minutes)
with auth_sessions_lock:
auth_sessions[session_token] = {
"ip": client_ip,
"expires": expires,
"created": datetime.now()
}
print(f"[AUTH] Successful login from {client_ip} ({_request_origin().kind}), session expires in {timeout_minutes} minutes")
# Return session token
response = jsonify({
"success": True,
"session_token": session_token,
"expires": expires.isoformat(),
"timeout_minutes": timeout_minutes
})
# Set cookie for browser-based access
response.set_cookie(
"auth_session",
session_token,
max_age=timeout_minutes * 60, # in seconds
httponly=True, # Prevent JavaScript access
samesite="Strict" # CSRF protection
)
return response
except Exception as e:
print(f"[AUTH ERROR] {e}")
import traceback
traceback.print_exc()
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/api/auth/logout", methods=["POST"])
def password_logout():
"""Logout and invalidate the current session"""
try:
# Get session token
session_token = request.cookies.get("auth_session")
if not session_token:
auth_header = request.headers.get("Authorization")
if auth_header and auth_header.startswith("Bearer "):
session_token = auth_header[7:]
if session_token:
with auth_sessions_lock:
if session_token in auth_sessions:
del auth_sessions[session_token]
print(f"[AUTH] Logged out session from {request.remote_addr}")
response = jsonify({"success": True, "message": "Logged out successfully"})
# Clear cookie
response.set_cookie("auth_session", "", max_age=0)
return response
except Exception as e:
print(f"[AUTH ERROR] {e}")
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/api/auth/status", methods=["GET"])
def auth_status():
"""Check if the current session is authenticated"""
try:
is_authenticated = check_ip_whitelist()
# Get session info if authenticated via password
session_info = None
if is_authenticated:
session_token = request.cookies.get("auth_session")
if not session_token:
auth_header = request.headers.get("Authorization")
if auth_header and auth_header.startswith("Bearer "):
session_token = auth_header[7:]
if session_token:
with auth_sessions_lock:
session_data = auth_sessions.get(session_token)
if session_data:
session_info = {
"authenticated_via": "password",
"expires": session_data["expires"].isoformat(),
"created": session_data["created"].isoformat()
}
if not session_info and is_authenticated:
session_info = {
"authenticated_via": "ip_whitelist"
}
# Determine redirect URL based on authentication
redirect_url = "/live-settings" if is_authenticated else "/url-builder"
origin = _request_origin()
return jsonify({
"success": True,
"authenticated": is_authenticated,
"session": session_info,
# The real caller, so a tunnel visitor doesn't see our loopback.
"ip": origin.client_ip,
"origin": origin.kind,
"redirect_url": redirect_url
})
except Exception as e:
print(f"[AUTH ERROR] {e}")
return jsonify({"success": False, "error": str(e)}), 500
@app.route("/live-settings")
def settings_page():
"""Render the live settings page"""
if not check_ip_whitelist():
return render_template("auth-required.html"), 403
return render_template("live-settings.html")
@app.route("/service-phase")
def service_phase_page():
"""Render the service phase detection / review page"""
if not check_ip_whitelist():
return render_template("auth-required.html"), 403
return render_template("service-phase.html")
def _service_phase_resolve_db():
"""(path, error_response) for the running session.
Live only, deliberately. These routes take no path from the caller, so there is no
path to confine and no way to reach a file outside the running session.
"""
live = _service_phase_session_db()
if not live or not os.path.exists(live):
return None, (jsonify({"success": False, "error": "No session is running"}), 404)
return live, None
@app.route("/api/service-phase")
def get_service_phase():
"""Detected phases for a session — the running one, or ``?session=`` from the archive.
``?recompute=1`` re-runs the detector over the session's rows instead of reading the
output the tick saved, without writing anything. That is what makes the logic
improvable without waiting: change a threshold, reload, compare against what the
operator is watching happen. POST /api/service-phase/rerun is the writing version.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
db_path, is_live, err = _archive_resolve_db(request.args.get("session"))
if err:
return err
cfg = _service_phase_config()
recompute = request.args.get("recompute") in ("1", "true", "yes")
try:
with _archive_open_ro(db_path, is_live) as conn:
if recompute:
result = _service_phase_analyze(
_service_phase_rows(conn), cfg,
first_sunday=_service_phase_first_sunday(db_path),
rules=_service_phase_rules(), live=is_live)
else:
result = _service_phase_load(conn)
corrections = _service_phase_corrections(conn)
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
# Resolved here rather than in the page: the rule for where a marked phase ends is the
# one the summariser applies, and two implementations of it would eventually disagree
# about which stretch a summary was written from.
now_ms = int(time.time() * 1000)
mark_spans = _phase_marks_resolve(corrections, result.get("blocks", []), now_ms=now_ms)
return jsonify({
"success": True,
"session_id": os.path.basename(db_path),
"live": is_live,
"recomputed": recompute,
"enabled": bool(cfg.get("enabled", True)),
# The page renumbers the songs itself when a correction moves the opening, so it
# needs the same threshold the detector used to decide what counts as a song.
"songs_min_minutes": int(cfg.get("songs_min_minutes", 3)),
"first_sunday": _service_phase_first_sunday(db_path),
"current": result.get("current"),
"blocks": result.get("blocks", []),
"bins": result.get("bins", []),
"classes": result.get("classes", ""),
"spans": result.get("spans", []),
"corrections": corrections,
"mark_spans": mark_spans,
"marked": _phase_marks_describe(corrections, now_ms=now_ms),
})
@app.route("/sermon-summary")
def sermon_summary_page():
"""Render the sermon summary review page (live session and the archive)."""
if not check_ip_whitelist():
return render_template("auth-required.html"), 403
return render_template("sermon-summary.html")
def _archive_session_paths(limit=None):
"""Every recorded session database, newest first.
The same sweep _service_phase_learn_scan uses. It is also the whitelist: a caller names
a session by basename and stt.session_index.resolve_session only ever compares that name
against this enumeration, so no caller-supplied string is joined onto a directory and
none of it reaches sqlite3.connect.
"""
if limit is None:
limit = coerce_int(_sermon_summary_config().get("archive_scan_limit"), 200, lo=1, hi=2000)
return sorted(_db_iter_databases(_sidecar_sweep_dirs()), reverse=True)[:limit]
def _archive_resolve_db(session):
"""(path, is_live, error_response) for a named session, or the running one.
``is_live`` is what decides whether a write may retire the database's WAL afterwards —
see _archive_write_done.
The archive is swept only when the caller actually named a session. Sweeping it for the
live case as well walked the whole backup tree on every poll, every correction and every
re-run, to build an enumeration that resolve_session then never looks at — and on an
archive of any size that walk was the request.
"""
named = (session or "").strip()
resolved = _session_resolve(_archive_session_paths() if named else (), session,
_service_phase_session_db())
if resolved is None:
if (session or "").strip():
return None, False, (jsonify({"success": False, "error": "Unknown session"}), 404)
return None, False, (jsonify({"success": False, "error": "No session is running"}), 404)
return resolved[0], resolved[1], None
def _archive_open_ro(db_path, is_live=False):
"""A read-only connection to a session database.
Thin wrapper over db_maintenance.open_readonly, which carries the reasoning: a session
the process died during is still in WAL mode, and a plain ``mode=ro`` open cannot read
it. Recovery is refused for the live session — its writer owns that WAL.
"""
return _db_open_readonly(db_path, recover=not is_live)
# What each session last failed with, so a broken archive is reported once rather than
# once per poll: the review pages ask for this listing every minute.
_archive_unreadable_seen = {}
def _archive_unreadable(db_path, error, stage="open"):
"""Record a session the sweep could not read, and say so in the log.
Both halves matter. The listing carries it so the page can tell "nothing recorded" from
"could not be read", and the log carries the exception text, which until now was
discarded at the ``except`` and existed nowhere afterwards — a server that had stopped
being able to open its own archive looked, from every side, like a server with an empty
one. That is not hypothetical: a checkpoint that emptied a WAL underneath this process
left every open of that session failing with "disk I/O error" for the rest of the
process's life, while the same file read perfectly from anywhere else.
The log line is printed when the failure is new or has changed, not on every listing:
a page polling once a minute would otherwise turn one broken session into a thousand
identical lines a day, which is its own way of hiding something.
"""
entry = _session_unreadable(db_path, error, stage)
key = f"{entry['stage']}: {entry['error']}"
if _archive_unreadable_seen.get(entry["session_id"]) != key:
_archive_unreadable_seen[entry["session_id"]] = key
print(f"[ARCHIVE] {entry['session_id']}: cannot {stage} ({entry['error']})")
return entry
def _archive_write_done(db_path, is_live):
"""Retire the sidecars after writing to a finished session. No-op on the live one.
A finished session may already have been delivered to the NAS by stt/file_mover.py, and
writing to it in WAL mode leaves -wal/-shm beside a file this process no longer owns —
the litter sweep_db_sidecars exists to clean. checkpoint_and_release folds the WAL in and
switches the database to journal_mode=DELETE, which is persistent, so a service that has
been reviewed never grows sidecars again.
The live session is excluded deliberately: its writer is another thread that owns that
WAL, and retiring it from a web request would be reaching into a file still in use.
sweep_db_sidecars takes the same stance.
"""
if is_live or not db_path:
return False
try:
return _db_checkpoint_and_release(db_path)
except Exception as e:
print(f"[ARCHIVE] could not retire sidecars for {os.path.basename(db_path)} "
f"({type(e).__name__}: {e})")
return False
@app.route("/api/service-phase/settings", methods=["POST"])
def save_service_phase_settings():
"""Turn phase detection on or off.
Until now this flag could only be changed by editing config.json on the machine, which
for a detector an operator is meant to be reviewing and correcting is the wrong place to
keep its only switch.
Takes effect on the next tick — _service_phase_tick re-reads the config every run — so
there is nothing to restart.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access Denied"}), 403
global config
data = _control_params(keep_blank=True)
if "service_phase" not in config:
config["service_phase"] = {}
raw = data.get("enabled")
enabled = raw if isinstance(raw, bool) else str(raw).strip().lower() in ("1", "true", "yes", "on")
config["service_phase"]["enabled"] = enabled
save_config(config)
return jsonify({"success": True, "enabled": enabled})
@app.route("/api/service-phase/sessions")
def list_service_phase_sessions():
"""Every recorded service, newest first — the picker behind both review pages.
Databases are opened read-only and one at a time: the archive shares a disk with the
session a service may be recording to, and a listing is never worth competing with the
writer for it. One unreadable session is reported rather than failing the request, the
same way _service_phase_learn_scan sweeps.
A session that cannot be opened is named in ``unreadable`` instead of being dropped
silently. An archive the server cannot *read* used to render exactly like an archive
with nothing in it — the page said "No recorded service" over a complete service sitting
on disk, and the reason (a sqlite error) existed nowhere at all. Listing continues past
the failure: one bad database must not empty the picker.
Sessions with no finalized rows are dropped by session_index.index — a start/stop leaves
a database behind, and listing those buries the real services.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
live = _service_phase_session_db()
described = []
unreadable = []
for path in _archive_session_paths():
try:
conn = _archive_open_ro(path, is_live=bool(live and os.path.basename(path) == os.path.basename(live)))
except Exception as e:
unreadable.append(_archive_unreadable(path, e, "open"))
continue
try:
described.append((path, _session_describe(conn)))
except Exception as e:
unreadable.append(_archive_unreadable(path, e, "read"))
continue
finally:
conn.close()
sessions = _session_index(described, live_path=live)
# A session that has just started has no rows yet, so the sweep drops it — but it is the
# one an operator mid-service is looking for. Put it back at the top.
if live and not any(x["live"] for x in sessions):
sessions.insert(0, {"session_id": os.path.basename(live), "date": os.path.basename(live)[:10],
"live": True, "rows": 0, "start_ms": 0, "end_ms": 0, "minutes": 0,
"has_phase": False, "has_summaries": False})
return jsonify({"success": True, "sessions": sessions, "unreadable": unreadable,
"live_session_id": os.path.basename(live) if live else None})
@app.route("/api/service-phase/transcript")
def get_service_phase_transcript():
"""The captions under a stretch of a service, for the operator moving its edges.
Deliberately the summariser's own query — read_sermon_rows, is_final and not denied — so
the panel shows the material a summary would actually be written from. Anything close but
different would let the page and the summary disagree about the same sermon, which is
worse than showing nothing.
``start_ms``/``end_ms`` are the window to read, and the caller widens them past the phase
on purpose: moving a start *earlier* means reading captions that are currently outside it.
Deciding which rows are inside the phase is the caller's job; this returns what is there.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
db_path, is_live, err = _archive_resolve_db(request.args.get("session"))
if err:
return err
start_ms = coerce_int(request.args.get("start_ms"), 0, lo=0, hi=2 ** 62)
end_ms = coerce_int(request.args.get("end_ms"), 0, lo=0, hi=2 ** 62)
if end_ms <= start_ms:
return jsonify({"success": False, "error": "A window needs an end after its start"}), 400
# A whole service is a few thousand captions; the cap is against a runaway range rather
# than a real one, and the caller is told when it bit so it can say so.
limit = coerce_int(request.args.get("limit"), 1500, lo=1, hi=5000)
try:
with _archive_open_ro(db_path, is_live) as conn:
rows = _sermon_read_rows(conn, start_ms, end_ms)
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
return jsonify({
"success": True,
"session_id": os.path.basename(db_path),
"live": is_live,
"truncated": len(rows) > limit,
"rows": [{"id": r.id, "ts_ms": r.ts_ms, "text": r.text} for r in rows[:limit]],
})
@app.route("/api/service-phase/rerun", methods=["POST"])
def rerun_service_phase():
"""Re-run the detector over a session and save the result.
Deliberately a POST rather than a flag on GET ?recompute=1, which re-derives and writes
nothing on purpose — that is what lets a threshold change be compared against what is
stored. This is the other thing an operator wants: a past service whose phases were
detected by older logic, brought up to date.
Corrections are untouched; they live in their own table precisely so a re-run cannot
destroy them.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
data = _control_params(keep_blank=True)
db_path, is_live, err = _archive_resolve_db(data.get("session"))
if err:
return err
cfg = _service_phase_config()
started = time.perf_counter()
try:
# Short timeout on the live session: the transcription worker owns that database and
# the phase tick is already re-running the detector over it every interval, so a
# re-run here has nothing to add and must not sit in a lock queue behind a service
# that is recording. An archived session has no competing writer.
conn = sqlite3.connect(db_path, timeout=3 if is_live else 15)
try:
result = _service_phase_analyze(
_service_phase_rows(conn), cfg,
first_sunday=_service_phase_first_sunday(db_path),
rules=_service_phase_rules(), live=is_live)
written = _service_phase_save(conn, result)
finally:
conn.close()
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
_archive_write_done(db_path, is_live)
return jsonify({"success": True, "session_id": os.path.basename(db_path),
"live": is_live, "written": written,
"elapsed_ms": round((time.perf_counter() - started) * 1000),
"blocks": len(result.get("blocks", []))})
@app.route("/api/sermon-summary")
def get_sermon_summary():
"""Stored sermon summaries — the running session, or ``?session=`` from the archive."""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
db_path, is_live, err = _archive_resolve_db(request.args.get("session"))
if err:
return err
cfg = _sermon_summary_config()
target = ((config.get("live_translation", {}) or {}).get("target_language") or "").strip()
try:
with _archive_open_ro(db_path, is_live) as conn:
summaries = _sermon_load_all(conn)
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
return jsonify({
"success": True,
"session_id": os.path.basename(db_path),
"live": is_live,
"enabled": bool(cfg.get("enabled", False)),
# Named so the page can label the second column and its copy button with the
# language rather than "translated", which tells a media team nothing.
"target_language": target,
"target_language_name": TRANSLATION_LANGUAGES.get(target, target),
"summaries": summaries,
})
@app.route("/api/sermon-summary/settings", methods=["POST"])
def save_sermon_summary_settings():
"""Turn the automatic summariser on or off.
Only the automatic trigger. The buttons stay live either way, because asking for a
summary is asking for one and an operator who has just pressed Summarise is not asking
to be told the feature is disabled.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access Denied"}), 403
global config
data = _control_params(keep_blank=True)
if "sermon_summary" not in config:
config["sermon_summary"] = {}
raw = data.get("enabled")
enabled = raw if isinstance(raw, bool) else str(raw).strip().lower() in ("1", "true", "yes", "on")
config["sermon_summary"]["enabled"] = enabled
save_config(config)
return jsonify({"success": True, "enabled": enabled})
@app.route("/api/sermon-summary/generate", methods=["POST"])
def generate_sermon_summary():
"""Summarise a sermon now, or regenerate one — live session or an archived one.
With no ``fingerprint`` it scans the session's blocks ignoring the settle window and
including a still-ongoing one, which is the operator saying the sermon is over
regardless of what the detector still thinks. On a finished service every block is
closed already, so the same call simply summarises whatever sermons it holds.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
data = _control_params(keep_blank=True)
db_path, is_live, err = _archive_resolve_db(data.get("session"))
if err:
return err
# Before either branch: regenerating is as impossible as generating when there is no
# model, and queueing it only moves the failure somewhere the operator will not look.
blocked = _sermon_llm_unavailable((config.get("live_translation", {}) or {}).get("llm") or {})
if blocked:
return jsonify({"success": False, "error": blocked}), 400
fingerprint_value = (data.get("fingerprint") or "").strip()
if fingerprint_value:
try:
with sqlite3.connect(db_path, timeout=5) as conn:
entry = _sermon_load(conn, fingerprint_value)
if not entry:
return jsonify({"success": False, "error": "Unknown summary"}), 404
_sermon_save(conn, fingerprint=fingerprint_value, label=entry["label"],
start_ms=entry["start_ms"], end_ms=entry["end_ms"],
status=_SERMON_PENDING, transcript=entry.get("transcript") or "")
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
_sermon_queue.put((db_path, fingerprint_value, is_live))
return jsonify({"success": True, "queued": 1, "live": is_live})
# No fingerprint: queue whatever is summarisable right now.
try:
with _archive_open_ro(db_path, is_live) as conn:
blocks = _service_phase_load(conn).get("blocks", [])
corrections = _service_phase_corrections(conn)
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
# manual=True: the operator is asking, so a sermon that already has a summary may be
# summarised again. This is the only caller that passes it — the end-of-session catch-up
# also sets ignore_settle, and it must stay bound by the one-summary rule.
queued = _sermon_summary_scan(blocks, db_path, ignore_settle=True, is_live=is_live,
corrections=corrections, manual=True)
cfg = _sermon_summary_config()
return jsonify({
"success": True, "live": is_live, "queued": queued,
"reason": None if queued else _sermon_explain(
blocks, min_minutes=coerce_int(cfg.get("min_minutes"), 8, lo=1, hi=240)),
})
@app.route("/api/service-phase/correct", methods=["POST"])
def save_service_phase_correction():
"""Record an operator correction. Never touched by the detector's own rewrites."""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
data = _control_params(keep_blank=True)
db_path, is_live, err = _archive_resolve_db(data.get("session"))
if err:
return err
block_index = data.get("block_index")
block_index = None if block_index in (None, "", "null") else coerce_int(block_index, 0, lo=0, hi=10000)
label = (data.get("label") or "").strip()[:120]
kind = (data.get("kind") or "").strip()[:8]
note = (data.get("note") or "").strip()[:500]
if not label and not kind:
return jsonify({"success": False, "error": "A correction needs a label or a kind."}), 400
try:
conn = sqlite3.connect(db_path, timeout=5)
try:
row_id = _service_phase_save_correction(
conn, block_index, kind=kind or None, label=label or None,
start_ms=coerce_int(data.get("start_ms"), 0, lo=0, hi=2 ** 62) or None,
end_ms=coerce_int(data.get("end_ms"), 0, lo=0, hi=2 ** 62) or None,
note=note, corrected_at=datetime.now().isoformat(timespec="seconds"))
corrections = _service_phase_corrections(conn)
finally:
conn.close()
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
_archive_write_done(db_path, is_live)
return jsonify({"success": True, "id": row_id, "corrections": corrections})
@app.route("/api/service-phase/uncorrect", methods=["POST"])
def delete_service_phase_correction():
"""Withdraw an operator correction, handing the block back to the detector."""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
data = _control_params(keep_blank=True)
db_path, is_live, err = _archive_resolve_db(data.get("session"))
if err:
return err
# A group spans several blocks and so has no block_index; the page undoes it by the row
# id it already holds. One or the other, never both.
row_id = data.get("id")
block_index = data.get("block_index")
if row_id not in (None, "", "null"):
row_id = coerce_int(row_id, 0, lo=1, hi=2 ** 62)
block_index = None
elif block_index not in (None, "", "null"):
row_id = None
block_index = coerce_int(block_index, 0, lo=0, hi=10000)
else:
return jsonify({"success": False, "error": "An undo needs a block index or a row id."}), 400
try:
conn = sqlite3.connect(db_path, timeout=5)
try:
if row_id is not None:
removed = _service_phase_delete_correction_by_id(conn, row_id)
else:
removed = _service_phase_delete_correction(conn, block_index)
corrections = _service_phase_corrections(conn)
finally:
conn.close()
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
_archive_write_done(db_path, is_live)
return jsonify({"success": True, "removed": removed, "corrections": corrections})
@app.route("/api/service-phase/mark", methods=["POST"])
def mark_service_phase():
"""Record the moment the operator says a phase begins — or ends, or began earlier.
The row is a correction with a start and no end, which is what makes it a mark: every
consumer written before marks existed tests both edges, so it is invisible to them
rather than half-understood. What it means is resolved in stt/phase_marks.
Deliberately not a span. An operator in the room knows the exact moment and can name a
stretch before it is over, which the detector cannot; an operator watching a service
will also mark a start and never mark the end, which is why the end stays the
detector's job unless they say otherwise.
Adjusting the start of a phase that is still running arrives here too, as ``move_id``,
for the same reason: writing a span with both edges would close a sermon that is still
being preached, which is what moving its start used to do.
"""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
return _phase_mark_apply(_control_params(keep_blank=True))
@app.route("/api/control/phase-mark", methods=["GET", "POST"])
def control_phase_mark():
"""The same mark, reachable from a button that can only send a GET.
Bitfocus Companion's simplest action is an HTTP GET, and every route that changes
something here is POST-only, so that button gets a 405. Rather than relax the method on
the route the browser uses — a GET that mutates is one image tag away from being fired
by any page the operator's browser loads — this is a separate door that will not accept
the ambient credentials a drive-by would ride on. The access token has to be in the URL,
which a button has forever and a hostile page cannot know. See stt/control_auth.
A mark outranks the detector everywhere downstream, so one press at the start of the
preaching is worth more than any amount of tuning — and it is also evidence the learner
reads back afterwards.
"""
data = _control_params(keep_blank=True)
refused = _control_refuse_reason(
data, token=str(config.get("web_server", {}).get("access_token") or ""),
method=request.method)
if refused:
return jsonify({"success": False, "error": refused}), 403
result = _phase_mark_apply(data)
# A stream deck shows a line of text, not a JSON body.
if isinstance(result, tuple):
return result
payload = result.get_json() if hasattr(result, "get_json") else result
if isinstance(payload, dict) and payload.get("success"):
payload = dict(payload)
payload["text"] = _phase_mark_text(data, payload)
return jsonify(payload)
return result
def _phase_mark_text(data, payload):
"""One line a control surface can display back to whoever pressed the button."""
if str(data.get("undo") or "").lower() in ("1", "true", "yes"):
return "Undid the last mark" if payload.get("removed") else "Nothing to undo"
if str(data.get("end") or "").lower() in ("1", "true", "yes"):
return "Marked the end"
return "Marked %s" % ((data.get("label") or "").strip()[:120] or "a phase")
def _phase_mark_apply(data):
"""Record, move or undo a mark. Shared by the page's route and the control link."""
db_path, is_live, err = _archive_resolve_db(data.get("session"))
if err:
return err
undo = str(data.get("undo") or "").lower() in ("1", "true", "yes")
ends = str(data.get("end") or "").lower() in ("1", "true", "yes")
# Moving a start rather than adding one. A phase has one beginning however many times
# the operator adjusts it, and an adjustment that left the old mark behind would open a
# second phase of the same name ending where the first press was.
move_id = coerce_int(data.get("move_id"), 0, lo=0, hi=2 ** 62)
label = (data.get("label") or "").strip()[:120]
kind = (data.get("kind") or "").strip()[:8]
if ends:
# The sentinel is the server's business, not the page's: "the operator said
# something stopped" is one fact, and only one place should know how it is spelled.
label, kind = _MARK_END, "_"
# Its own moment, or the one the page pressed: a press travelling over a slow link
# should land where the operator was, not where the server got to.
at_ms = coerce_int(data.get("at_ms"), 0, lo=0, hi=2 ** 62) or int(time.time() * 1000)
if not undo and not label:
return jsonify({"success": False, "error": "A mark needs a phase name."}), 400
try:
conn = sqlite3.connect(db_path, timeout=5)
try:
removed = 0
row_id = None
if undo:
# The newest mark only. Undo is for the press that just happened — a mark
# from earlier in the service is corrected by marking again, where the
# operator can see what they are changing.
latest = [c for c in _service_phase_corrections(conn) if _phase_is_mark(c)]
if latest:
removed = _service_phase_delete_correction_by_id(
conn, int(latest[-1].get("id") or 0))
else:
if move_id:
# Only a mark moves. A miss — an id undone from another tab, or a row
# that was never a mark — falls through to a plain insert rather than
# an error: the operator said where the phase starts, and mid-service is
# the worst moment to refuse that over bookkeeping.
prior = next((c for c in _service_phase_corrections(conn)
if int(c.get("id") or 0) == move_id and _phase_is_mark(c)), None)
if prior is not None:
removed = _service_phase_delete_correction_by_id(conn, move_id)
row_id = _service_phase_save_correction(
conn, None, kind=kind or None, label=label, start_ms=at_ms,
end_ms=None, note="live mark",
corrected_at=datetime.now().isoformat(timespec="seconds"))
corrections = _service_phase_corrections(conn)
blocks = _service_phase_load(conn).get("blocks", [])
finally:
conn.close()
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
_archive_write_done(db_path, is_live)
now_ms = int(time.time() * 1000)
return jsonify({
"success": True, "id": row_id, "removed": removed, "at_ms": at_ms,
"corrections": corrections,
"mark_spans": _phase_marks_resolve(corrections, blocks, now_ms=now_ms),
"marked": _phase_marks_describe(corrections, now_ms=now_ms),
})
@app.route("/api/service-phase/group", methods=["POST"])
def group_service_phase_blocks():
"""Record several detected blocks as one phase, spanning their whole time range."""
if not check_ip_whitelist():
return jsonify({"success": False, "error": "Access denied"}), 403
data = _control_params(keep_blank=True)
db_path, is_live, err = _archive_resolve_db(data.get("session"))
if err:
return err
start_ms = coerce_int(data.get("start_ms"), 0, lo=0, hi=2 ** 62)
end_ms = coerce_int(data.get("end_ms"), 0, lo=0, hi=2 ** 62)
label = (data.get("label") or "").strip()[:120]
kind = (data.get("kind") or "").strip()[:8]
note = (data.get("note") or "").strip()[:500]
if not label:
return jsonify({"success": False, "error": "A group needs a name."}), 400
if not start_ms or end_ms <= start_ms:
return jsonify({"success": False, "error": "A group needs a time range."}), 400
try:
conn = sqlite3.connect(db_path, timeout=5)
try:
row_id = _service_phase_save_group(
conn, start_ms, end_ms, kind=kind or None, label=label,
note=note, corrected_at=datetime.now().isoformat(timespec="seconds"))
corrections = _service_phase_corrections(conn)
finally:
conn.close()
except Exception as e:
return jsonify({"success": False, "error": f"{type(e).__name__}: {e}"}), 500
_archive_write_done(db_path, is_live)
return jsonify({"success": True, "id": row_id, "corrections": corrections})
def _service_phase_learn_scan(limit=200):
"""Corrected phases across the archive, newest sessions first.
Read-only and opened one at a time: the archive is on the same disk a service may be
recording to, and a learner is never worth competing with the writer for it.
"""
paths = sorted(_db_iter_databases(_sidecar_sweep_dirs()), reverse=True)[:limit]
phases = []
profiles = {}
for path in paths:
try:
# Same recovery as the review pages: a session the process died during is
# still in WAL mode, and it is the corrections in exactly those that the
# learner most wants.
conn = _archive_open_ro(path)
except Exception as e:
_archive_unreadable(path, e, "open")
continue
try:
name = os.path.basename(path)
found = _phase_learn_collect([(name, conn)], with_text=True)
# A live mark is the strongest boundary evidence there is, and it was being
# discarded: it carries no end and no block index, so the corrections reader
# rejects it. Only stretches nothing else already names are taken, so a mark the
# operator later corrected does not count twice.
claimed = [(p.start_ms, p.end_ms) for p in found]
for marked in _phase_learn_marks(conn, n