diff --git a/docs/HTTP_REMOTE.md b/docs/HTTP_REMOTE.md new file mode 100644 index 0000000..9479448 --- /dev/null +++ b/docs/HTTP_REMOTE.md @@ -0,0 +1,212 @@ +# Remote HTTP-Based Transcription + +Vocalinux can offload speech recognition to a remote HTTP server instead of running a local model. This is useful when you want faster transcription on a more powerful machine, want to share one model across several clients, or want to keep your laptop free of GPU/model overhead. + +Vocalinux speaks two wire formats out of the box: + +- **OpenAI-compatible** — `POST /v1/audio/transcriptions` (e.g. OpenAI, [Speaches](https://github.com/speaches-ai/speaches), LocalAI, vLLM with Whisper) +- **whisper.cpp server** — `POST /inference` (the binary shipped with [whisper.cpp](https://github.com/ggerganov/whisper.cpp)) + +Pick whichever your server exposes — the rest of this guide applies to both. + +> **Tip — share the server with your phone.** Both formats are open standards, so the same self-hosted server can also back mobile dictation apps. On Android, apps like *Dictate* and *Transcribro* speak OpenAI-compatible Whisper; on iOS, Shortcuts-based dictation clients can hit the same endpoint. Run one Whisper server, point your laptop and phone at it, and you get consistent dictation everywhere without uploading audio to a third party. + +## How It Works + +When the **Remote API** engine is active, Vocalinux: + +1. Captures microphone audio locally (16 kHz, mono, 16-bit PCM). +2. On end-of-utterance, packages the buffer as a WAV file in memory. +3. Uploads it via `multipart/form-data` to the configured server. +4. Reads the transcribed text from the JSON response (`{"text": "..."}`) and types it into the focused window. + +No audio is ever written to disk. The local machine still runs the VAD/segmentation logic — only the heavy ASR step is remote. + +## Server-Side Setup + +You need a server that accepts audio uploads and returns `{"text": "..."}`. Two common options: + +### Option A: whisper.cpp server + +Build and run the bundled HTTP server from the [whisper.cpp](https://github.com/ggerganov/whisper.cpp) repo: + +```bash +git clone https://github.com/ggerganov/whisper.cpp +cd whisper.cpp +make server +./models/download-ggml-model.sh medium.en +./build/bin/whisper-server \ + --model models/ggml-medium.en.bin \ + --host 0.0.0.0 \ + --port 8080 +``` + +The server now listens on `http://:8080`. The endpoint is `/inference`. + +### Option B: OpenAI-compatible server (Speaches, LocalAI, etc.) + +Run any server that implements `POST /v1/audio/transcriptions`. Example with [Speaches](https://github.com/speaches-ai/speaches): + +```bash +docker run --rm -p 8000:8000 ghcr.io/speaches-ai/speaches:latest +``` + +The endpoint is `/v1/audio/transcriptions`. + +### Network checklist + +- The server must be reachable from the client over HTTP or HTTPS. +- If you bind to `0.0.0.0`, open the port in your firewall (`ufw allow 8080/tcp`, etc.). +- For anything outside a trusted LAN, terminate TLS in front of the server (Caddy, nginx, Traefik) and require an API key — the audio you upload is private speech. +- If you plan to also use the server from a mobile dictation app, terminate TLS even on a LAN — most Android and iOS apps refuse cleartext HTTP regardless of network. + +## Client-Side Setup (Vocalinux) + +### Via the Settings dialog + +The remote engine is a power-user override and lives on the **Advanced** tab so casual users aren't confronted with it. + +1. Launch Vocalinux and open **Settings → Advanced → Remote Server**. +2. Toggle **Use remote server** on. This overrides whatever local engine is selected on the Speech Engine tab. +3. Fill in the fields: + - **Server URL**: base URL of the server, e.g. `http://192.168.1.100:8080` (no trailing slash needed; one is stripped automatically). + - **API Key** (optional): sent as `Authorization: Bearer `. Leave blank if your server doesn't require auth. + - **API Endpoint**: pick **Whisper.cpp (`/inference`)** or **OpenAI (`/v1/audio/transcriptions`)** to match your server. +4. Click **Test Connection** — a successful test means the URL is reachable and credentials (if any) are accepted. A failure here is just a warning; Vocalinux will still try again on the first transcription. + +Settings auto-save and re-initialise the engine immediately, so you can start dictating as soon as the test passes. Toggling the switch off restores the local engine selected on the Speech Engine tab. + +### Via the config file + +The same options live in `~/.config/vocalinux/config.json`: + +```json +{ + "speech_recognition": { + "engine": "remote_api", + "remote_api_url": "http://192.168.1.100:8080", + "remote_api_key": "", + "remote_api_endpoint": "/inference" + } +} +``` + +Restart Vocalinux after editing the file by hand. + +## Wire Protocol Reference + +If you're writing a custom server, here's exactly what the client sends. + +### Request — whisper.cpp format (`/inference`) + +``` +POST {server_url}/inference HTTP/1.1 +Authorization: Bearer # only if configured +Content-Type: multipart/form-data; boundary=... + +--boundary +Content-Disposition: form-data; name="file"; filename="audio.wav" +Content-Type: audio/wav + + +--boundary +Content-Disposition: form-data; name="temperature" + +0.0 +--boundary +Content-Disposition: form-data; name="temperature_inc" + +0.2 +--boundary +Content-Disposition: form-data; name="response_format" + +json +--boundary +Content-Disposition: form-data; name="language" + +en +--boundary-- +``` + +### Request — OpenAI-compatible format (`/v1/audio/transcriptions`) + +``` +POST {server_url}/v1/audio/transcriptions HTTP/1.1 +Authorization: Bearer +Content-Type: multipart/form-data; boundary=... + +--boundary +Content-Disposition: form-data; name="file"; filename="audio.wav" +Content-Type: audio/wav + + +--boundary +Content-Disposition: form-data; name="model" + +whisper-1 +--boundary +Content-Disposition: form-data; name="language" + +en +--boundary-- +``` + +### Response (both formats) + +```json +{ "text": "the transcribed utterance" } +``` + +Empty utterances should return `{"text": ""}`. HTTP 4xx/5xx responses are surfaced as transcription errors in the Vocalinux log. + +### Language handling + +- Vocalinux's UI language `auto` is sent as **no `language` field** (server auto-detects). +- `en-us` is normalised to `en`. +- Anything else is forwarded verbatim. + +### Timeouts + +- Connection test: **5 s** +- Transcription request: **30 s** + +A request that exceeds the transcription timeout is logged and dropped; the next utterance is sent fresh. + +## Verifying with `curl` + +Quick sanity check without launching the GUI: + +```bash +# Whisper.cpp server +curl -F file=@sample.wav \ + -F temperature=0.0 \ + -F response_format=json \ + http://192.168.1.100:8080/inference + +# OpenAI-compatible +curl -H "Authorization: Bearer $API_KEY" \ + -F file=@sample.wav \ + -F model=whisper-1 \ + http://192.168.1.100:8000/v1/audio/transcriptions +``` + +Both should return JSON with a `text` field. + +## Troubleshooting + +| Symptom | Likely cause | Fix | +|---|---|---| +| Connection test reports "failed" | Wrong host/port, firewall, or server not running | Check `curl ` from the client; check server logs | +| HTTP 401/403 | API key missing or wrong | Set the **API Key** field; ensure the server is configured for that key | +| HTTP 404 on first send | Endpoint format mismatch | Switch the **API Endpoint** dropdown to the format your server actually exposes | +| Transcription always empty | Server rejected the WAV (sample rate, channels) | Confirm server accepts 16 kHz mono 16-bit WAV; check server logs | +| First transcription is slow | Cold model load on the server | Warm the server before dictating, or increase the client timeout server-side | +| Garbled text in foreign language | Wrong `language` setting | Set the language explicitly in Settings instead of `auto` | + +Logs live at `~/.local/share/vocalinux/vocalinux.log` (or wherever your `XDG_DATA_HOME` points). Look for lines tagged `Remote API` for client-side detail. + +## Security Notes + +- The API key is stored in plain JSON in `~/.config/vocalinux/config.json`. Treat that file the same way you treat any other secret on disk. +- Audio is sent as raw WAV — anyone on the wire can hear it. Use HTTPS for any deployment outside a fully trusted network. +- Vocalinux does **not** validate TLS certificate pins; standard system trust is used. Self-signed certs require importing the CA into your system trust store. diff --git a/install.sh b/install.sh index a094921..6f3c25b 100755 --- a/install.sh +++ b/install.sh @@ -195,7 +195,7 @@ while [[ $# -gt 0 ]]; do echo "Options:" echo " --interactive, -i Force interactive mode (default)" echo " --auto Non-interactive automatic installation" - echo " --engine=NAME Speech engine: whisper_cpp (default), whisper, vosk" + echo " --engine=NAME Speech engine: whisper_cpp (default), whisper, vosk, remote_api" echo " --dev Install in development mode with all dev dependencies" echo " --test Run tests after installation" echo " --venv-dir=PATH Specify custom virtual environment directory" @@ -890,6 +890,15 @@ EOF echo " │ • Good for basic dictation needs │" echo " └─────────────────────────────────────────────────────────────┘" echo "" + echo " ┌─────────────────────────────────────────────────────────────┐" + echo " │ 4. REMOTE API (ADVANCED) │" + echo " │ • Offload processing to a GPU server on your network │" + echo " │ • Ideal for laptops without GPU │" + echo " │ • Supports whisper.cpp server & OpenAI-compatible APIs │" + echo " │ • Minimal local resources needed │" + echo " │ • Requires a remote server to be running │" + echo " └─────────────────────────────────────────────────────────────┘" + echo "" # Show recommendation case "$RECOMMENDED_ENGINE" in @@ -908,7 +917,7 @@ EOF esac echo "" - read -p "Choose engine [1-3] (default: $DEFAULT_CHOICE): " ENGINE_CHOICE + read -p "Choose engine [1-4] (default: $DEFAULT_CHOICE): " ENGINE_CHOICE ENGINE_CHOICE=${ENGINE_CHOICE:-$DEFAULT_CHOICE} case "$ENGINE_CHOICE" in @@ -924,6 +933,10 @@ EOF SELECTED_ENGINE="vosk" ENGINE_DISPLAY="VOSK (Lightweight)" ;; + 4) + SELECTED_ENGINE="remote_api" + ENGINE_DISPLAY="Remote API" + ;; *) SELECTED_ENGINE="whisper_cpp" ENGINE_DISPLAY="Whisper.cpp (Recommended)" @@ -1029,7 +1042,31 @@ EOF echo "" fi - # Step 4: Model download preference + # Step 3 for Remote API: Configure server URL + if [[ "$SELECTED_ENGINE" == "remote_api" ]]; then + print_header "Step 3: Configure Remote Server" + echo "" + print_info "You need a speech recognition server running on your local network." + echo "" + echo " Supported servers:" + echo " • whisper.cpp server: ./server -m model.bin --host 0.0.0.0 --port 8080" + echo " • LocalAI: docker run -p 8080:8080 localai/localai" + echo " • Faster Whisper: faster-whisper-server --host 0.0.0.0 --port 8080" + echo " • Any OpenAI-compatible speech API" + echo "" + read -p "Enter remote server URL (or leave blank to set later): " REMOTE_API_URL_INPUT + if [ -n "$REMOTE_API_URL_INPUT" ]; then + REMOTE_API_URL="$REMOTE_API_URL_INPUT" + REMOTE_DISPLAY="$REMOTE_API_URL" + else + REMOTE_API_URL="" + REMOTE_DISPLAY="(configure later in Settings)" + fi + echo "" + fi + + # Step 4: Model download preference (skip for remote_api) + if [[ "$SELECTED_ENGINE" != "remote_api" ]]; then print_header "Step 4: Model Download" echo "" echo "Speech recognition models can be downloaded now or later." @@ -1052,6 +1089,11 @@ EOF else MODELS_DISPLAY="Download now (recommended)" fi + else + # Remote API: No need to download model + SKIP_MODELS="yes" + MODELS_DISPLAY="Not needed (remote processing)" + fi # Summary print_header "Installation Summary" @@ -1063,6 +1105,9 @@ EOF echo " Note: GPU build will compile from source (2-5 minutes)" fi fi + if [[ "$SELECTED_ENGINE" == "remote_api" ]]; then + echo " Remote Server: $REMOTE_DISPLAY" + fi echo " Models: $MODELS_DISPLAY" echo " Install Location: ${INSTALL_DIR:-\$HOME/.local/share/vocalinux}" echo "" @@ -1144,7 +1189,7 @@ if [[ "$NON_INTERACTIVE" == "yes" ]] && [[ -z "$SELECTED_ENGINE" ]]; then # Default to whisper.cpp for best performance SELECTED_ENGINE="whisper_cpp" print_info "Automatic mode: Installing with whisper.cpp (default engine)" - print_info "For other engines, use: --engine=whisper or --engine=vosk" + print_info "For other engines, use: --engine=whisper or --engine=vosk or --engine=remote_api" echo "" fi @@ -2640,6 +2685,74 @@ FALLBACK_CONFIG VOSK_CONFIG fi ;; + + remote_api) + print_info "Setting up Remote API engine..." + print_info "" + print_info "╔════════════════════════════════════════════════════════╗" + print_info "║ Setting up REMOTE API Engine ║" + print_info "╠════════════════════════════════════════════════════════╣" + print_info "║ • Offloads speech recognition to a remote server ║" + print_info "║ • Ideal for laptops without GPU ║" + print_info "║ • Supports whisper.cpp server & OpenAI APIs ║" + print_info "║ • Requires: a server running on your network ║" + print_info "╚════════════════════════════════════════════════════════╝" + print_info "" + + # Ensure requests library is installed + print_info "Installing requests library..." + pip install requests --log "$PIP_LOG_FILE" || { + print_error "Failed to install requests library" + return 1 + } + print_success "requests library installed" + + # URL was collected upfront by run_interactive_install + # (or left blank in auto/non-interactive mode). + local REMOTE_API_URL="${REMOTE_API_URL:-}" + + # Create configuration file + local REMOTE_CONFIG_FILE="$CONFIG_DIR/config.json" + if [ ! -f "$REMOTE_CONFIG_FILE" ]; then + mkdir -p "$CONFIG_DIR" + cat > "$REMOTE_CONFIG_FILE" << REMOTE_CONFIG +{ + "speech_recognition": { + "engine": "remote_api", + "model_size": "small", + "vosk_model_size": "small", + "whisper_model_size": "tiny", + "whisper_cpp_model_size": "tiny", + "remote_api_url": "${REMOTE_API_URL}", + "remote_api_key": "", + "vad_sensitivity": 3, + "silence_timeout": 2.0 + }, + "audio": { + "device_index": null, + "device_name": null + }, + "shortcuts": { + "toggle_recognition": "ctrl+ctrl" + }, + "ui": { + "start_minimized": false, + "show_notifications": true + }, + "advanced": { + "debug_logging": false, + "wayland_mode": false + } +} +REMOTE_CONFIG + fi + + if [ -n "$REMOTE_API_URL" ]; then + print_success "Remote API configured with server: $REMOTE_API_URL" + else + print_warning "No server URL configured. You can set it later in Settings." + fi + ;; esac fi @@ -3376,6 +3489,10 @@ EOF ENGINE_DISPLAY_NAME="VOSK" BACKEND_INFO="Lightweight" ;; + remote_api) + ENGINE_DISPLAY_NAME="Remote API" + BACKEND_INFO="Network (offloaded to remote server)" + ;; esac echo "" diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index bac6be3..ec82f27 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -58,7 +58,7 @@ def parse_arguments(): parser.add_argument( "--engine", type=str, - choices=["vosk", "whisper", "whisper_cpp"], + choices=["vosk", "whisper", "whisper_cpp", "remote_api"], help="Speech recognition engine to use (whisper_cpp recommended for best performance)", ) parser.add_argument("--wayland", action="store_true", help="Force Wayland compatibility mode") @@ -382,6 +382,9 @@ def main(): whispercpp_logprob_thold=advanced_settings.get("whispercpp_logprob_thold", -1.0), whispercpp_no_speech_thold=advanced_settings.get("whispercpp_no_speech_thold", 0.6), whispercpp_n_threads=advanced_settings.get("whispercpp_n_threads", 0), + remote_api_url=saved_settings.get("remote_api_url", ""), + remote_api_key=saved_settings.get("remote_api_key", ""), + remote_api_endpoint=saved_settings.get("remote_api_endpoint", "/inference"), ) # Initialize text injection system diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index a183f1b..bd73b2d 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -569,6 +569,14 @@ def _filter_non_speech(text: str) -> str: [BLANK_AUDIO], music notes, or other non-speech artifacts when transcribing silent or ambiguous audio. + A trailing newline returned by the upstream transcription service is + preserved in the output. Some pipelines (post-processing proxies, + fixup layers) emit a meaningful trailing '\\n' to signal that the + typed text should be submitted (Enter key); stripping it silently + discards that signal. Pattern matching still runs against a + whitespace-stripped probe so the existing non-speech checks are + unaffected. + Args: text: The transcribed text to filter @@ -580,7 +588,8 @@ def _filter_non_speech(text: str) -> str: if not text or not text.strip(): return "" - text = text.strip() + # Probe is used for non-speech-pattern matching only. + probe = text.strip() # Non-speech patterns to filter out non_speech_patterns = [ @@ -596,20 +605,24 @@ def _filter_non_speech(text: str) -> str: ] for pattern in non_speech_patterns: - if re.match(pattern, text, re.IGNORECASE): - logger.debug(f"Filtered non-speech token: '{text}'") + if re.match(pattern, probe, re.IGNORECASE): + logger.debug(f"Filtered non-speech token: '{probe}'") return "" # Check if text has enough actual speech content # At least 30% of characters should be alphanumeric or common speech punctuation - speech_chars = sum(1 for c in text if c.isalnum() or c in ".,!?-'\"") - total_chars = len(text) + speech_chars = sum(1 for c in probe if c.isalnum() or c in ".,!?-'\"") + total_chars = len(probe) if total_chars > 0 and speech_chars / total_chars < 0.3: - logger.debug(f"Filtered low-speech-content text: '{text}'") + logger.debug(f"Filtered low-speech-content text: '{probe}'") return "" - return text + # Strip whitespace as before, but preserve a trailing newline if the + # upstream API sent one — it carries an explicit submit/Enter signal + # from post-processing proxies. + trailing_newline = text.endswith(("\n", "\r")) + return probe + "\n" if trailing_newline else probe def _show_notification(title: str, message: str, icon: str = "dialog-warning"): @@ -764,6 +777,12 @@ class SpeechRecognitionManager: self.whispercpp_no_speech_thold = kwargs.get("whispercpp_no_speech_thold", 0.6) self.whispercpp_n_threads = kwargs.get("whispercpp_n_threads", None) + # Remote API settings + self.remote_api_url = kwargs.get("remote_api_url", "") + self.remote_api_key = kwargs.get("remote_api_key", "") + self.remote_api_endpoint = kwargs.get("remote_api_endpoint", "/inference") + self._http_session = None + # Audio diagnostics tracking self._last_audio_level = 0.0 self._audio_level_callbacks: list[Callable[[float], None]] = [] @@ -801,6 +820,8 @@ class SpeechRecognitionManager: self._init_whisper() elif engine == "whisper_cpp": self._init_whispercpp() + elif engine == "remote_api": + self._init_remote_api() else: raise ValueError(f"Unsupported speech recognition engine: {engine}") @@ -1339,6 +1360,252 @@ class SpeechRecognitionManager: logger.error(f"Error in whisper.cpp transcription: {e} ({audio_info})", exc_info=True) return "" + def _init_remote_api(self): + """Initialize remote API speech recognition engine. + + Verify URL settings and try connection test. No need to load local model. + """ + if not self.remote_api_url: + logger.warning("Remote API URL not set. Please enter the server URL in settings.") + self._model_initialized = False + return + + if not self.remote_api_url.startswith(("http://", "https://")): + logger.error( + f"Remote API URL must start with http:// or https://, " + f"got: '{self.remote_api_url}'" + ) + self._model_initialized = False + return + + # Clean trailing slash from URL + self.remote_api_url = self.remote_api_url.rstrip("/") + + logger.info(f"Initialize remote API engine, server: {self.remote_api_url}") + + # Replace any existing session (prevents leak on back-to-back inits) + import requests + + if self._http_session: + self._http_session.close() + self._http_session = requests.Session() + + # Try connection test + try: + test_url = self.remote_api_url + headers = {} + if self.remote_api_key: + headers["Authorization"] = f"Bearer {self.remote_api_key}" + + response = self._http_session.get(test_url, headers=headers, timeout=5) + if response.ok: + logger.info( + f"Remote server connection test successful (status={response.status_code})" + ) + else: + logger.warning( + f"Remote server connection test returned non-success status: " + f"{response.status_code}. Will try again during recognition." + ) + except Exception as e: + logger.warning( + f"Remote server connection test failed: {e}. " + "Will try to connect again during recognition." + ) + + # Remote API does not need local models, directly mark as ready + self._model_initialized = True + logger.info("Remote API engine setup complete.") + + def _transcribe_with_remote_api(self, audio_buffer: list[bytes]) -> str: + """Transcribe audio via remote API. + + Package audio buffer into WAV format and send to remote server via HTTP POST. + Supports OpenAI compatible format (/v1/audio/transcriptions) and + whisper.cpp server format (/inference). + + Args: + audio_buffer: Audio data chunk list (16-bit PCM at 16kHz) + + Returns: + Transcribed text + """ + import io + import time + import wave + + try: + if not audio_buffer: + return "" + + if not self.remote_api_url: + logger.error("Remote API URL not set") + return "" + + # Convert audio buffer to WAV format + audio_data = b"".join(audio_buffer) + wav_buffer = io.BytesIO() + with wave.open(wav_buffer, "wb") as wav_file: + wav_file.setnchannels(1) # Mono + wav_file.setsampwidth(2) # 16-bit + wav_file.setframerate(16000) # 16kHz + wav_file.writeframes(audio_data) + + wav_buffer.seek(0) + wav_bytes = wav_buffer.read() + + duration = len(audio_data) / (2 * 16000) # 16-bit = 2 bytes/sample + logger.debug( + f"Remote API transcription: {duration:.2f} seconds audio, " + f"{len(wav_bytes)} bytes WAV" + ) + + # Prepare language parameters + lang = self.language + if lang == "en-us": + lang = "en" + elif lang == "auto": + lang = None + + # Prepare HTTP request headers + headers = {} + if self.remote_api_key: + headers["Authorization"] = f"Bearer {self.remote_api_key}" + + transcribe_start = time.time() + + text = None + if self.remote_api_endpoint == "/inference": + text = self._try_whispercpp_server_api(wav_bytes, lang, headers) + else: + text = self._try_openai_api(wav_bytes, lang, headers) + + # If both formats fail + if text is None: + logger.error( + "Remote API transcription failed: Cannot connect to server or API format not supported" + ) + return "" + + transcribe_duration = time.time() - transcribe_start + rtf = transcribe_duration / duration if duration > 0 else 0 + + # Filter non-speech content. Do NOT pre-strip — _filter_non_speech + # preserves a trailing '\n' from the upstream API (used by + # post-processing proxies to signal Enter), and a pre-strip here + # would silently discard it. + text = _filter_non_speech(text) if text else "" + + if text: + logger.info(f"Remote API transcription result: '{text}'") + logger.info( + f"Remote API transcription took {transcribe_duration:.3f}s " + f"({duration:.2f}s audio, RTF: {rtf:.2f}x)" + ) + else: + logger.debug( + f"Remote API returned blank transcription result ({transcribe_duration:.3f}s)" + ) + + return text + + except Exception as e: + audio_info = ( + f"audio buffer: {len(audio_buffer)} chunks" + if audio_buffer + else "empty audio buffer" + ) + logger.error(f"Remote API transcription error: {e} ({audio_info})", exc_info=True) + return "" + + def _try_openai_api(self, wav_bytes: bytes, lang, headers: dict): + """Try to transcribe using OpenAI compatible API format. + + Args: + wav_bytes: Audio data in WAV format + lang: Language core (e.g. "en", None for auto detect) + headers: HTTP request headers + + Returns: + Transcribed text, or None if format is not supported + """ + import requests + + url = f"{self.remote_api_url}{self.remote_api_endpoint}" + + files = {"file": ("audio.wav", wav_bytes, "audio/wav")} + data = {"model": "whisper-1"} + if lang: + data["language"] = lang + + try: + response = self._http_session.post( + url, headers=headers, files=files, data=data, timeout=30 + ) + + if response.status_code == 404: + logger.debug("OpenAI API endpoint does not exist, try other formats") + return None + + response.raise_for_status() + result = response.json() + + # OpenAI format returns {"text": "..."} + return result.get("text", "") + + except requests.exceptions.ConnectionError as e: + logger.error(f"Cannot connect to remote server {url}: {e}") + return None + except Exception as e: + logger.debug(f"OpenAI API format attempt failed: {e}") + return None + + def _try_whispercpp_server_api(self, wav_bytes: bytes, lang, headers: dict): + """Try to transcribe using whisper.cpp server API format. + + Args: + wav_bytes: Audio data in WAV format + lang: Language core (e.g. "en", None for auto detect) + headers: HTTP request headers + + Returns: + Transcribed text, or None if format is not supported + """ + import requests + + url = f"{self.remote_api_url}{self.remote_api_endpoint}" + + files = {"file": ("audio.wav", wav_bytes, "audio/wav")} + data = { + "temperature": "0.0", + "temperature_inc": "0.2", + "response_format": "json", + } + if lang: + data["language"] = lang + + try: + response = self._http_session.post( + url, headers=headers, files=files, data=data, timeout=30 + ) + + if response.status_code == 404: + logger.debug("whisper.cpp server endpoint does not exist") + return None + + response.raise_for_status() + result = response.json() + + # whisper.cpp server format returns {"text": "..."} + return result.get("text", "") + + except requests.exceptions.ConnectionError as e: + logger.error(f"Cannot connect to remote server {url}: {e}") + return None + except Exception as e: + logger.debug(f"whisper.cpp server API format attempt failed: {e}") + return None + def _download_whispercpp_model(self): """Download a whisper.cpp model with progress tracking.""" import requests @@ -1816,6 +2083,9 @@ class SpeechRecognitionManager: @property def model_ready(self) -> bool: """Check if the model is initialized and ready for recognition.""" + # Remote API does not need local models + if self.engine == "remote_api": + return self._model_initialized return self._model_initialized and self.model is not None def _get_stop_sound_guard_chunks(self) -> int: @@ -2279,6 +2549,9 @@ class SpeechRecognitionManager: elif self.engine == "whisper_cpp": text = self._transcribe_with_whispercpp(audio_buffer) + elif self.engine == "remote_api": + text = self._transcribe_with_remote_api(audio_buffer) + else: logger.error(f"Unknown engine: {self.engine}") return @@ -2418,6 +2691,7 @@ class SpeechRecognitionManager: ) restart_needed = False + old_engine = self.engine if engine is not None and engine != self.engine: self.engine = engine restart_needed = True @@ -2467,6 +2741,18 @@ class SpeechRecognitionManager: setattr(self, param_name, kwargs[param_name]) restart_needed = True + # Handle Remote API settings + if "remote_api_url" in kwargs: + new_url = kwargs.get("remote_api_url", "") + if new_url != self.remote_api_url: + self.remote_api_url = new_url + if self.engine == "remote_api": + restart_needed = True + if "remote_api_key" in kwargs: + self.remote_api_key = kwargs.get("remote_api_key", "") + if "remote_api_endpoint" in kwargs: + self.remote_api_endpoint = kwargs.get("remote_api_endpoint", "/inference") + self._voice_commands_enabled = self._resolve_voice_commands_enabled() if restart_needed: @@ -2490,6 +2776,10 @@ class SpeechRecognitionManager: # Release old resources explicitly if necessary (Python's GC might handle it) self.model = None self.recognizer = None + if old_engine == "remote_api" and self.engine != "remote_api": + if self._http_session is not None: + self._http_session.close() + self._http_session = None try: if self.engine == "vosk": self._init_vosk() @@ -2497,6 +2787,8 @@ class SpeechRecognitionManager: self._init_whisper() elif self.engine == "whisper_cpp": self._init_whispercpp() + elif self.engine == "remote_api": + self._init_remote_api() else: raise ValueError(f"Unsupported engine during reconfigure: {self.engine}") logger.info("Speech engine re-initialized successfully.") @@ -2628,6 +2920,7 @@ class SpeechRecognitionManager: with self._model_lock: self.model = None self.recognizer = None + self._http_session = None self._model_initialized = False try: @@ -2637,6 +2930,8 @@ class SpeechRecognitionManager: self._init_whisper() elif self.engine == "whisper_cpp": self._init_whispercpp() + elif self.engine == "remote_api": + self._init_remote_api() else: logger.error("Cannot reinitialize: unknown engine '%s'", self.engine) return diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 201a855..f17658b 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -29,6 +29,9 @@ DEFAULT_CONFIG = { "silence_timeout": 2.0, # Seconds of silence before stopping "stop_sound_guard_ms": 200, # Small tail trim to avoid the stop sound without clipping speech "voice_commands_enabled": None, # None = auto (enabled for VOSK, disabled for Whisper) + "remote_api_url": "", # Remote speech recognition server URL (e.g. http://192.168.1.100:8080) + "remote_api_key": "", # Remote server API key (optional) + "remote_api_endpoint": "/inference", # Remote server API endpoint format (/v1/audio/transcriptions or /inference) }, "audio": { "device_index": None, # Audio input device index (None for system default) diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 314865d..4062e04 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -73,8 +73,31 @@ ENGINE_MODELS = { "medium", "large", ], # whisper.cpp models (ggml format) + "remote_api": [], # Remote API does not need local models } +# Engine display name mapping +ENGINE_DISPLAY_NAMES = { + "vosk": "Vosk", + "whisper": "Whisper", + "whisper_cpp": "Whisper_cpp", + "remote_api": "Remote API", +} + + +def _engine_display_name(engine: str) -> str: + """Get the display name of the engine.""" + return ENGINE_DISPLAY_NAMES.get(engine, engine.capitalize()) + + +def _engine_from_display(display_name: str) -> str: + """Reverse lookup engine ID from display name.""" + for engine_id, name in ENGINE_DISPLAY_NAMES.items(): + if name == display_name: + return engine_id + return display_name.lower() + + # Whisper model metadata for display WHISPER_MODEL_INFO = { "tiny": {"size_mb": 75, "desc": "Fastest, lowest accuracy", "params": "39M"}, @@ -90,7 +113,7 @@ def get_available_engines(): Detect which speech recognition engines are available/installed. Returns a dictionary of engine_name -> availability (bool). """ - engines = {"vosk": False, "whisper": False, "whisper_cpp": False} + engines = {"vosk": False, "whisper": False, "whisper_cpp": False, "remote_api": False} # Check VOSK try: @@ -116,6 +139,14 @@ def get_available_engines(): except ImportError: pass + # Remote API is always available (only requires requests package) + try: + import requests # noqa: F401 + + engines["remote_api"] = True + except ImportError: + pass + logger.debug(f"Available engines: {engines}") return engines @@ -858,8 +889,11 @@ class SettingsDialog(Gtk.Dialog): general_label.set_tooltip_text("General settings") notebook.append_page(self.general_tab, general_label) + # Advanced tab - whisper.cpp parameters and power-user features (remote API, etc.) advanced_label = Gtk.Label(label="Advanced") - advanced_label.set_tooltip_text("Advanced whisper.cpp parameters") + advanced_label.set_tooltip_text( + "Advanced whisper.cpp parameters and settings for power users" + ) self.advanced_page_num = notebook.append_page(self.advanced_tab, advanced_label) notebook.connect("switch-page", self._on_settings_page_switched) @@ -1086,24 +1120,24 @@ class SettingsDialog(Gtk.Dialog): self.model_combo = Gtk.ComboBoxText() self.model_combo.set_size_request(180, -1) _prevent_scroll_on_hover(self.model_combo) - model_row = PreferenceRow( + self.model_row = PreferenceRow( title="Model Size", subtitle="Larger models are more accurate but slower", widget=self.model_combo, ) - group.add_row(model_row) + group.add_row(self.model_row) # Language selection self.language_combo = Gtk.ComboBoxText() self.language_combo.set_size_request(180, -1) self.language_combo.set_tooltip_text("Primary language for speech recognition") _prevent_scroll_on_hover(self.language_combo) - language_row = PreferenceRow( + self.language_row = PreferenceRow( title="Language", subtitle="Primary language for recognition", widget=self.language_combo, ) - group.add_row(language_row) + group.add_row(self.language_row) self.content_box.pack_start(group, False, False, 0) @@ -1134,10 +1168,10 @@ class SettingsDialog(Gtk.Dialog): self.content_box.pack_start(self.language_warning, False, False, 0) # Legend - legend_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=16) - legend_box.set_halign(Gtk.Align.CENTER) - legend_box.set_margin_top(4) - legend_box.set_margin_bottom(4) + self.legend_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=16) + self.legend_box.set_halign(Gtk.Align.CENTER) + self.legend_box.set_margin_top(4) + self.legend_box.set_margin_bottom(4) for symbol, text in [ ("✓", "Downloaded"), @@ -1146,15 +1180,102 @@ class SettingsDialog(Gtk.Dialog): ]: item = Gtk.Label(label=f"{symbol} {text}") item.get_style_context().add_class("status-info") - legend_box.pack_start(item, False, False, 0) + self.legend_box.pack_start(item, False, False, 0) - self.content_box.pack_start(legend_box, False, False, 0) + self.content_box.pack_start(self.legend_box, False, False, 0) # Connect signals self.engine_combo.connect("changed", self._on_engine_changed) self.model_combo.connect("changed", self._on_model_changed) self.language_combo.connect("changed", self._on_language_changed) + def _on_remote_api_settings_changed(self, widget): + """Handle remote API URL/Key/endpoint changes (Advanced tab).""" + if self._initializing or self._applying_settings: + return + + url = self.remote_api_url_entry.get_text().strip() + key = self.remote_api_key_entry.get_text().strip() + endpoint = self.remote_api_endpoint_combo.get_active_id() or "/inference" + + self.config_manager.set("speech_recognition", "remote_api_url", url) + self.config_manager.set("speech_recognition", "remote_api_key", key) + self.config_manager.set("speech_recognition", "remote_api_endpoint", endpoint) + self.config_manager.save_settings() + + # Auto apply only if remote engine is currently active + if self.use_remote_switch.get_active(): + self._auto_apply_settings() + + def _on_test_remote_connection(self, widget): + """Test remote server connection.""" + url = self.remote_api_url_entry.get_text().strip() + if not url: + self.remote_status_label.set_markup( + "✗ Please enter a server URL" + ) + return + + self.remote_test_btn.set_sensitive(False) + self.remote_test_btn.set_label("Testing...") + self.remote_status_label.set_markup("Connecting...") + + key = self.remote_api_key_entry.get_text().strip() + endpoint = self.remote_api_endpoint_combo.get_active_id() or "/inference" + + def test_connection(url=url, key=key, endpoint=endpoint): + try: + import requests + + headers = {} + if key: + headers["Authorization"] = f"Bearer {key}" + + clean_url = url.rstrip("/") + + # Try connecting to the server + response = requests.get(clean_url, headers=headers, timeout=5) + + # Determine server type + server_info = "" + + try: + if endpoint == "/v1/audio/transcriptions": + # Try OpenAI endpoint + openai_resp = requests.get( + f"{clean_url}/v1/models", headers=headers, timeout=5 + ) + if openai_resp.status_code == 200: + server_info = " (OpenAI compatible)" + elif endpoint == "/inference": + # Try whisper.cpp server endpoint + whispercpp_resp = requests.get( + f"{clean_url}/inference", headers=headers, timeout=5 + ) + if whispercpp_resp.status_code != 404: + server_info = " (whisper.cpp server)" + except Exception: + pass + + GLib.idle_add( + self.remote_status_label.set_markup, + f"✓ Connected! " + f"(status={response.status_code}){server_info}", + ) + + except Exception as e: + error_msg = str(e)[:80] + GLib.idle_add( + self.remote_status_label.set_markup, + f"✗ Connection failed: {error_msg}", + ) + + # Restore button state + GLib.idle_add(self.remote_test_btn.set_sensitive, True) + GLib.idle_add(self.remote_test_btn.set_label, "Test Connection") + + threading.Thread(target=test_connection, daemon=True).start() + def _build_recognition_section(self): """Build the Recognition Settings section.""" group = PreferencesGroup(title="Recognition Settings") @@ -1677,6 +1798,113 @@ class SettingsDialog(Gtk.Dialog): self.power_user_switch.connect("state-set", self._on_power_user_toggled) + """Build the Advanced tab (currently: Remote Server settings).""" + self.remote_api_group = PreferencesGroup( + title="Remote Server", + description=( + "Offload speech recognition to a server on your network. " + "Supports whisper.cpp and OpenAI-compatible APIs." + ), + ) + + # Toggle: enable remote engine (overrides Speech Engine tab selection) + self.use_remote_switch = Gtk.Switch() + self.use_remote_switch.set_active(False) + self.use_remote_switch.set_tooltip_text( + "When enabled, transcription is performed by the configured remote " + "server instead of the local engine selected on the Speech Engine tab." + ) + use_remote_row = PreferenceRow( + title="Use remote server", + subtitle="Override the local engine and send audio to a remote server", + widget=self.use_remote_switch, + ) + self.remote_api_group.add_row(use_remote_row) + + # Server URL + self.remote_api_url_entry = Gtk.Entry() + self.remote_api_url_entry.set_placeholder_text("http://192.168.1.100:8080") + self.remote_api_url_entry.set_tooltip_text( + "URL of the remote speech recognition server\n" + "Supports OpenAI compatible API and whisper.cpp server" + ) + self.remote_api_url_entry.set_size_request(280, -1) + remote_url_row = PreferenceRow( + title="Server URL", + subtitle="Remote speech recognition server address", + widget=self.remote_api_url_entry, + ) + self.remote_api_group.add_row(remote_url_row) + + # API Key + self.remote_api_key_entry = Gtk.Entry() + self.remote_api_key_entry.set_placeholder_text("(optional)") + self.remote_api_key_entry.set_visibility(False) + self.remote_api_key_entry.set_tooltip_text("API Key for authentication (optional)") + self.remote_api_key_entry.set_size_request(280, -1) + remote_key_row = PreferenceRow( + title="API Key", + subtitle="Authentication key (optional)", + widget=self.remote_api_key_entry, + ) + self.remote_api_group.add_row(remote_key_row) + + # API Endpoint + self.remote_api_endpoint_combo = Gtk.ComboBoxText() + self.remote_api_endpoint_combo.set_size_request(280, -1) + self.remote_api_endpoint_combo.set_tooltip_text( + "Select the API format of the remote server (API Endpoint Format)" + ) + self.remote_api_endpoint_combo.append( + "/v1/audio/transcriptions", "OpenAI (/v1/audio/transcriptions)" + ) + self.remote_api_endpoint_combo.append("/inference", "Whisper.cpp (/inference)") + _prevent_scroll_on_hover(self.remote_api_endpoint_combo) + remote_endpoint_row = PreferenceRow( + title="API Endpoint", + subtitle="API format for the remote server", + widget=self.remote_api_endpoint_combo, + ) + self.remote_api_group.add_row(remote_endpoint_row) + + # Connection test + self.remote_test_btn = Gtk.Button(label="Test Connection") + self.remote_test_btn.set_tooltip_text("Test connection to remote server") + self.remote_test_btn.connect("clicked", self._on_test_remote_connection) + remote_test_row = PreferenceRow( + title="Connection Test", + subtitle="Verify remote server is reachable", + widget=self.remote_test_btn, + ) + self.remote_api_group.add_row(remote_test_row) + + self.advanced_tab.pack_start(self.remote_api_group, False, False, 0) + + # Status label below the group + self.remote_status_label = Gtk.Label(label="", use_markup=True, xalign=0) + self.remote_status_label.set_margin_start(16) + self.remote_status_label.set_margin_top(4) + self.remote_status_label.get_style_context().add_class("status-info") + self.advanced_tab.pack_start(self.remote_status_label, False, False, 0) + + # Load saved values into the widgets + saved_url = self.config_manager.get("speech_recognition", "remote_api_url", "") + saved_key = self.config_manager.get("speech_recognition", "remote_api_key", "") + saved_endpoint = self.config_manager.get( + "speech_recognition", "remote_api_endpoint", "/inference" + ) + if saved_url: + self.remote_api_url_entry.set_text(saved_url) + if saved_key: + self.remote_api_key_entry.set_text(saved_key) + self.remote_api_endpoint_combo.set_active_id(saved_endpoint) + + # Connect signals (toggle and field changes both trigger re-apply) + self.use_remote_switch.connect("notify::active", self._on_use_remote_toggled) + self.remote_api_url_entry.connect("changed", self._on_remote_api_settings_changed) + self.remote_api_key_entry.connect("changed", self._on_remote_api_settings_changed) + self.remote_api_endpoint_combo.connect("changed", self._on_remote_api_settings_changed) + def _on_power_user_toggled(self, widget, state): """Handle the power-user opt-in toggle.""" if self._initializing or self._applying_settings: @@ -1771,6 +1999,13 @@ class SettingsDialog(Gtk.Dialog): self._auto_apply_settings() + def _on_use_remote_toggled(self, switch, _gparam): + """Toggle remote-engine override: forces engine to remote_api when on.""" + if self._initializing or self._applying_settings: + return + self._update_engine_specific_ui() + self._auto_apply_settings() + def _load_and_apply_settings(self): """Load current settings and populate the UI.""" settings = self._get_current_settings() @@ -1797,44 +2032,63 @@ class SettingsDialog(Gtk.Dialog): self.copy_to_clipboard_switch.set_active(copy_to_clipboard) self.sound_effects_switch.set_active(self.config_manager.is_sound_effects_enabled()) - # Populate engine combo with only available engines + # Populate engine combo with only available LOCAL engines. + # Remote API is not a peer engine in the combo — it lives on the + # Advanced tab as a toggle that overrides the local engine. available_engines = get_available_engines() available_count = 0 for engine in ENGINE_MODELS.keys(): + if engine == "remote_api": + continue if available_engines.get(engine, False): - capitalized_engine = engine.capitalize() - self.engine_combo.append(capitalized_engine, capitalized_engine) + display_name = _engine_display_name(engine) + self.engine_combo.append(display_name, display_name) available_count += 1 if available_count == 0: - logger.error("No speech recognition engines available!") + logger.error("No local speech recognition engines available!") # Still add them so the UI works, but log the error for engine in ENGINE_MODELS.keys(): - capitalized_engine = engine.capitalize() - self.engine_combo.append(capitalized_engine, capitalized_engine) + if engine == "remote_api": + continue + display_name = _engine_display_name(engine) + self.engine_combo.append(display_name, display_name) else: - logger.info(f"Populated {available_count} available engines: {available_engines}") + logger.info(f"Populated {available_count} available local engines: {available_engines}") - # Set engine active - check if current engine is available - if not available_engines.get(self.current_engine, False): - logger.warning( - f"Current engine '{self.current_engine}' is not available, selecting first available" - ) - # Find first available engine - for engine in ENGINE_MODELS.keys(): - if available_engines.get(engine, False): - self.current_engine = engine - break + # If config says remote_api, turn the Advanced toggle on and pick a + # local fallback for the combo (so toggling off has somewhere to land). + remote_active = self.current_engine == "remote_api" + if remote_active: + self.use_remote_switch.set_active(True) + local_fallback = "whisper_cpp" + if not available_engines.get(local_fallback, False): + for engine in ENGINE_MODELS.keys(): + if engine != "remote_api" and available_engines.get(engine, False): + local_fallback = engine + break + self.current_engine = local_fallback + else: + self.use_remote_switch.set_active(False) + if not available_engines.get(self.current_engine, False): + logger.warning( + f"Current engine '{self.current_engine}' is not available, " + "selecting first available" + ) + for engine in ENGINE_MODELS.keys(): + if engine != "remote_api" and available_engines.get(engine, False): + self.current_engine = engine + break - engine_text = self.current_engine.capitalize() + engine_text = _engine_display_name(self.current_engine) logger.info(f"Setting active engine to: {engine_text}") if not self.engine_combo.set_active_id(engine_text): logger.warning("Could not set engine by ID, trying by index") # Find index of current engine model = self.engine_combo.get_model() for i, row in enumerate(model): - if row[0].lower() == self.current_engine.lower(): + if _engine_from_display(row[0]) == self.current_engine: self.engine_combo.set_active(i) break else: @@ -1926,9 +2180,14 @@ class SettingsDialog(Gtk.Dialog): logger.warning("No engine selected during model options population") return - engine = engine_text.lower() + engine = _engine_from_display(engine_text) logger.info(f"Populating model options for engine: {engine}") + # Remote API does not need model options + if engine == "remote_api": + logger.info("Remote API engine selected, no model options needed") + return + saved_model_for_engine = self.config_manager.get_model_size_for_engine(engine) logger.info(f"Saved model for {engine}: {saved_model_for_engine}") @@ -2001,7 +2260,7 @@ class SettingsDialog(Gtk.Dialog): if not engine_text: return - engine = engine_text.lower() + engine = _engine_from_display(engine_text) current_lang = self.language_combo.get_active_id() if current_lang: @@ -2009,7 +2268,7 @@ class SettingsDialog(Gtk.Dialog): current_lang == "auto" or not SUPPORTED_LANGUAGES.get(current_lang, {}).get("vosk") ): self.language = "en-us" - elif engine in ["whisper", "whisper_cpp"] and not current_lang: + elif engine in ["whisper", "whisper_cpp", "remote_api"] and not current_lang: self.language = "auto" self._populate_model_options() @@ -2026,7 +2285,7 @@ class SettingsDialog(Gtk.Dialog): if voice_commands_enabled is None: engine_text = self.engine_combo.get_active_text() - engine = engine_text.lower() if engine_text else sr_config.get("engine", "whisper_cpp") + engine = _engine_from_display(engine_text) if engine_text else "whisper_cpp" auto_enabled = engine == "vosk" self.voice_commands_switch.set_active(auto_enabled) @@ -2070,7 +2329,7 @@ class SettingsDialog(Gtk.Dialog): if not engine: return - engine = engine.lower() + engine = _engine_from_display(engine) for lang_code, lang_info in SUPPORTED_LANGUAGES.items(): display_text = lang_info["name"] @@ -2081,7 +2340,7 @@ class SettingsDialog(Gtk.Dialog): continue is_downloaded = _is_vosk_model_downloaded("small", lang_code) display_text += " ✓" if is_downloaded else " ↓" - elif engine in ["whisper", "whisper_cpp"]: + elif engine in ["whisper", "whisper_cpp", "remote_api"]: # Both Whisper and whisper.cpp support auto-detect if lang_code == "auto": display_text += " ⚠" @@ -2105,7 +2364,7 @@ class SettingsDialog(Gtk.Dialog): self._processing_language_change = True try: - engine = engine.lower() + engine = _engine_from_display(engine) lang_info = SUPPORTED_LANGUAGES.get(lang_code, {}) if lang_info.get("warning"): @@ -2124,7 +2383,26 @@ class SettingsDialog(Gtk.Dialog): self._processing_language_change = False def _update_engine_specific_ui(self): - """Show/hide UI elements specific to the selected engine.""" + """Show/hide UI elements driven by the active engine. + + The engine combo is local-only (vosk/whisper/whisper_cpp). The remote + engine is enabled via the Advanced tab's "Use remote server" switch, + which overrides the local selection. + """ + is_remote = ( + getattr(self, "use_remote_switch", None) is not None + and self.use_remote_switch.get_active() + ) + + # When remote is active, the local model/legend UI is irrelevant + if is_remote: + self.model_row.hide() + self.model_info_card.hide() + self.legend_box.hide() + else: + self.model_row.show_all() + self.legend_box.show_all() + self._update_model_info() self._update_advanced_tab_sensitivity() @@ -2161,7 +2439,16 @@ class SettingsDialog(Gtk.Dialog): self.model_info_card.hide() return - engine = engine_text.lower() + # Remote engine override (Advanced tab) hides the local model info card. + if ( + getattr(self, "use_remote_switch", None) is not None + and self.use_remote_switch.get_active() + ): + self.model_info_card.hide() + return + + engine = _engine_from_display(engine_text) + model_id = self.model_combo.get_active_id() if not model_id: self.model_info_card.hide() @@ -2341,14 +2628,18 @@ class SettingsDialog(Gtk.Dialog): model_id = self.model_combo.get_active_id() language_id = self.language_combo.get_active_id() - engine = engine_text.lower() if engine_text else "vosk" + # Advanced "Use remote server" toggle overrides the local engine combo. + if self.use_remote_switch.get_active(): + engine = "remote_api" + else: + engine = _engine_from_display(engine_text) if engine_text else "vosk" model_size = model_id.lower() if model_id else "small" language = language_id if language_id else "auto" vad = int(self.vad_spin.get_value()) silence = self.silence_spin.get_value() - return { + settings = { "engine": engine, "model_size": model_size, "language": language, @@ -2368,6 +2659,16 @@ class SettingsDialog(Gtk.Dialog): "whispercpp_no_speech_thold": self.advanced_no_speech_thold_spin.get_value(), } + # Remote API additional settings + if engine == "remote_api": + settings["remote_api_url"] = self.remote_api_url_entry.get_text().strip() + settings["remote_api_key"] = self.remote_api_key_entry.get_text().strip() + settings["remote_api_endpoint"] = ( + self.remote_api_endpoint_combo.get_active_id() or "/inference" + ) + + return settings + def _on_test_clicked(self, widget): """Handle click on the test button.""" if self._test_active: diff --git a/tests/test_main.py b/tests/test_main.py index 2986ce8..3c57376 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -231,6 +231,9 @@ class TestMainModule(unittest.TestCase): whispercpp_logprob_thold=-1.0, whispercpp_no_speech_thold=0.6, whispercpp_n_threads=0, + remote_api_url="", + remote_api_key="", + remote_api_endpoint="/inference", ) mock_text.assert_called_once_with(wayland_mode=True) mock_action_handler.assert_called_once_with(mock_text_instance) diff --git a/tests/test_recognition_manager_helpers.py b/tests/test_recognition_manager_helpers.py index 8af5047..92abf51 100644 --- a/tests/test_recognition_manager_helpers.py +++ b/tests/test_recognition_manager_helpers.py @@ -89,6 +89,23 @@ class TestFilterNonSpeech(unittest.TestCase): self.assertIn("are", result) self.assertIn("you", result) + def test_trailing_newline_preserved(self): + # A trailing '\n' from the upstream API is meaningful (e.g. a + # post-processing proxy emitting Enter) and must survive filtering. + self.assertEqual(_filter_non_speech("cd ..\n"), "cd ..\n") + + def test_trailing_newline_with_leading_whitespace(self): + # Leading whitespace is stripped; trailing '\n' is preserved. + self.assertEqual(_filter_non_speech(" hello world\n"), "hello world\n") + + def test_trailing_spaces_stripped_newline_kept(self): + # Spaces/tabs/CR before a trailing '\n' should be cleaned, '\n' kept. + self.assertEqual(_filter_non_speech("cmd \t\n"), "cmd\n") + + def test_lone_newline_filtered_as_whitespace(self): + # A bare '\n' (no speech) is whitespace-only -> filtered to "". + self.assertEqual(_filter_non_speech("\n"), "") + class TestShowNotification(unittest.TestCase): """Test _show_notification function.""" diff --git a/tests/test_recognition_manager_remote_api.py b/tests/test_recognition_manager_remote_api.py new file mode 100644 index 0000000..49dc3a8 --- /dev/null +++ b/tests/test_recognition_manager_remote_api.py @@ -0,0 +1,697 @@ +""" +Test coverage for remote API speech recognition engine. + +These tests cover the remote API engine added in PR #335. +""" + +import sys +import unittest +from unittest.mock import ANY, MagicMock, Mock + +import pytest + + +# autouse fixture: inject mock modules before each test, restore sys.modules after test, +# prevent polluting other test files in the same pytest session. +@pytest.fixture(autouse=True) +def _mock_heavy_deps(monkeypatch): + """Mock heavy dependencies via sys.modules, restoring after each test.""" + for mod in [ + "vosk", + "whisper", + "pyaudio", + "wave", + "tqdm", + "numpy", + "torch", + "psutil", + ]: + monkeypatch.setitem(sys.modules, mod, MagicMock()) + + # pywhispercpp requires submodule structure + mock_pywhispercpp = MagicMock() + mock_pywhispercpp.model = MagicMock() + mock_pywhispercpp.model.Model = MagicMock() + monkeypatch.setitem(sys.modules, "pywhispercpp", mock_pywhispercpp) + monkeypatch.setitem(sys.modules, "pywhispercpp.model", mock_pywhispercpp.model) + + # requests needs to keep real exception classes for except statements to work properly + mock_requests = MagicMock() + mock_requests.exceptions.ConnectionError = ConnectionError + mock_requests.exceptions.RequestException = Exception + monkeypatch.setitem(sys.modules, "requests", mock_requests) + + yield + + +def _import_manager(): + """Delayed import of SpeechRecognitionManager to ensure fixture mocks take effect before execution.""" + from vocalinux.speech_recognition.recognition_manager import ( + SpeechRecognitionManager, + ) + + return SpeechRecognitionManager + + +def _get_mock_requests(): + """Get the mock requests object currently injected into sys.modules.""" + return sys.modules["requests"] + + +def _get_mock_session(): + """Get the mock session instance returned by requests.Session().""" + return _get_mock_requests().Session.return_value + + +def _setup_requests_get_ok(status_code=200): + """Configure mock session.get to return a successful response.""" + mock_session = _get_mock_session() + mock_response = Mock() + mock_response.ok = True + mock_response.status_code = status_code + mock_session.get.return_value = mock_response + return mock_session.get + + +def _setup_requests_get_error(exception): + """Configure mock session.get to throw an exception.""" + mock_session = _get_mock_session() + mock_session.get.side_effect = exception + return mock_session.get + + +def _setup_requests_get_non_ok(status_code=500): + """Configure mock session.get to return a non-successful response.""" + mock_session = _get_mock_session() + mock_response = Mock() + mock_response.ok = False + mock_response.status_code = status_code + mock_session.get.return_value = mock_response + return mock_session.get + + +def _setup_requests_post_ok(json_data, status_code=200): + """Configure mock session.post to return a successful response.""" + mock_session = _get_mock_session() + mock_response = Mock() + mock_response.status_code = status_code + mock_response.json.return_value = json_data + mock_session.post.return_value = mock_response + return mock_session.post + + +def _setup_requests_post_error(exception): + """Configure mock session.post to throw an exception.""" + mock_session = _get_mock_session() + mock_session.post.side_effect = exception + return mock_session.post + + +def _setup_requests_post_status(status_code, raise_for_status_error=None): + """Configure mock session.post to return a response with a specific status code.""" + mock_session = _get_mock_session() + mock_response = Mock() + mock_response.status_code = status_code + if raise_for_status_error: + mock_response.raise_for_status.side_effect = raise_for_status_error + mock_session.post.return_value = mock_response + return mock_session.post + + +class TestRemoteAPIEngine(unittest.TestCase): + """Test cases for remote API speech recognition engine.""" + + def test_remote_api_init_without_url(self): + """Test remote API initialization without URL.""" + SpeechRecognitionManager = _import_manager() + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="", + ) + # Should not raise, just set _model_initialized to False + self.assertFalse(manager._model_initialized) + + def test_remote_api_init_with_invalid_url(self): + """Test remote API initialization with invalid URL format.""" + SpeechRecognitionManager = _import_manager() + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="invalid-url", + ) + # Should not raise, just set _model_initialized to False + self.assertFalse(manager._model_initialized) + + def test_remote_api_init_with_valid_url(self): + """Test remote API initialization with valid HTTP URL.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + # Should set _model_initialized to True + self.assertTrue(manager._model_initialized) + + def test_remote_api_init_with_valid_https_url(self): + """Test remote API initialization with HTTPS URL.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="https://192.168.1.100:9090", + ) + self.assertTrue(manager._model_initialized) + + def test_remote_api_init_strips_trailing_slash(self): + """Test remote API initialization strips trailing slash from URL.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090/", + ) + # URL should have trailing slash stripped + self.assertFalse(manager.remote_api_url.endswith("/")) + + def test_remote_api_init_with_connection_failure(self): + """Test remote API initialization handles connection failure gracefully.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_error(Exception("Connection refused")) + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + # Should still set _model_initialized to True (will retry later) + self.assertTrue(manager._model_initialized) + + def test_remote_api_init_with_server_error(self): + """Test remote API initialization handles non-OK server response.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_non_ok(status_code=500) + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + # Should still set _model_initialized to True (warns but continues) + self.assertTrue(manager._model_initialized) + + def test_remote_api_init_with_api_key(self): + """Test remote API initialization includes API key in headers.""" + SpeechRecognitionManager = _import_manager() + mock_get = _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + remote_api_key="test-api-key", + ) + self.assertTrue(manager._model_initialized) + self.assertEqual(manager.remote_api_key, "test-api-key") + # Verify the API key was sent in the request headers + mock_get.assert_called_once_with( + "http://localhost:9090", + headers={"Authorization": "Bearer test-api-key"}, + timeout=5, + ) + + def test_remote_api_init_with_inference_endpoint(self): + """Test remote API initialization with whisper.cpp server endpoint.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + remote_api_endpoint="/inference", + ) + self.assertEqual(manager.remote_api_endpoint, "/inference") + + def test_remote_api_init_with_openai_endpoint(self): + """Test remote API initialization with OpenAI compatible endpoint.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:8080", + remote_api_endpoint="/v1/audio/transcriptions", + ) + self.assertEqual(manager.remote_api_endpoint, "/v1/audio/transcriptions") + + def test_model_ready_remote_api_initialized(self): + """Test model_ready returns True for remote API when initialized.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + # Should be available after initialization + self.assertTrue(manager.model_ready) + + def test_model_ready_remote_api_not_initialized(self): + """Test model_ready returns False when remote API not initialized.""" + SpeechRecognitionManager = _import_manager() + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="", + ) + # Should not be available without URL + self.assertFalse(manager.model_ready) + + +class TestRemoteAPITranscription(unittest.TestCase): + """Test cases for remote API transcription functionality.""" + + def setUp(self): + """Set up test fixtures.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + self.manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + remote_api_key="test-key", + remote_api_endpoint="/inference", + ) + + def test_transcribe_with_empty_buffer(self): + """Test transcription with empty audio buffer.""" + result = self.manager._transcribe_with_remote_api([]) + self.assertEqual(result, "") + + def test_transcribe_with_no_url(self): + """Test transcription when URL is not set.""" + self.manager.remote_api_url = "" + result = self.manager._transcribe_with_remote_api([b"test"]) + self.assertEqual(result, "") + + def test_transcribe_with_openai_endpoint_success(self): + """Test successful transcription via OpenAI endpoint.""" + from unittest.mock import patch + + with patch.object(self.manager, "_try_openai_api") as mock_openai: + mock_openai.return_value = "test transcription" + + # Change endpoint to OpenAI + self.manager.remote_api_endpoint = "/v1/audio/transcriptions" + result = self.manager._transcribe_with_remote_api([b"test-audio-data"]) + + self.assertEqual(result, "test transcription") + + def test_transcribe_with_whisper_cpp_server_endpoint_success(self): + """Test successful transcription via whisper.cpp server endpoint.""" + from unittest.mock import patch + + with patch.object(self.manager, "_try_whispercpp_server_api") as mock_server: + mock_server.return_value = "hello world" + + result = self.manager._transcribe_with_remote_api([b"audio-bytes"]) + + self.assertEqual(result, "hello world") + + def test_transcribe_with_whispercpp_api_failing(self): + """Test transcription returns empty string when whisper.cpp server API fails.""" + from unittest.mock import patch + + with patch.object(self.manager, "_try_whispercpp_server_api") as mock_server: + mock_server.return_value = None + + result = self.manager._transcribe_with_remote_api([b"some-audio"]) + + # Should return empty string when API returns None + self.assertEqual(result, "") + mock_server.assert_called_once() + + def test_transcribe_with_openai_api_failing(self): + """Test transcription returns empty string when OpenAI API fails.""" + from unittest.mock import patch + + self.manager.remote_api_endpoint = "/v1/audio/transcriptions" + with patch.object(self.manager, "_try_openai_api") as mock_openai: + mock_openai.return_value = None + + result = self.manager._transcribe_with_remote_api([b"some-audio"]) + + # Should return empty string when API returns None + self.assertEqual(result, "") + mock_openai.assert_called_once() + + def test_transcribe_with_language_en_us_conversion(self): + """Test language code conversion from en-us to en.""" + from unittest.mock import patch + + with patch.object(self.manager, "_try_whispercpp_server_api") as mock_server: + mock_server.return_value = "test" + + self.manager.language = "en-us" + self.manager._transcribe_with_remote_api([b"audio"]) + + # Should have been called with "en" (converted from "en-us") + mock_server.assert_called_once_with(ANY, "en", ANY) + + def test_transcribe_with_language_auto(self): + """Test with auto language detection.""" + from unittest.mock import patch + + with patch.object(self.manager, "_try_whispercpp_server_api") as mock_server: + mock_server.return_value = "test" + + self.manager.language = "auto" + self.manager._transcribe_with_remote_api([b"audio"]) + + # Should pass None for auto detection + mock_server.assert_called_once_with(ANY, None, ANY) + + def test_transcribe_unexpected_exception_returns_empty(self): + """An unexpected exception inside transcription is caught and returns ''.""" + from unittest.mock import patch + + with patch.object(self.manager, "_try_whispercpp_server_api") as mock_server: + mock_server.side_effect = RuntimeError("boom") + + result = self.manager._transcribe_with_remote_api([b"audio"]) + + self.assertEqual(result, "") + + +class TestOpenAIAPIFormat(unittest.TestCase): + """Test cases for OpenAI compatible API format.""" + + def setUp(self): + """Set up test fixtures.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + self.manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:8080", + remote_api_key="my-api-key", + remote_api_endpoint="/v1/audio/transcriptions", + ) + + def test_try_openai_api_success(self): + """Test successful transcription via OpenAI API format.""" + _setup_requests_post_ok({"text": "transcribed text"}) + + result = self.manager._try_openai_api(b"wav-bytes", "en", {"Authorization": "Bearer key"}) + + self.assertEqual(result, "transcribed text") + + def test_try_openai_api_with_404(self): + """Test OpenAI API format returns None for 404.""" + _setup_requests_post_status(404) + + result = self.manager._try_openai_api(b"wav", "en", {}) + + self.assertIsNone(result) + + def test_try_openai_api_connection_error(self): + """Test OpenAI API handles connection errors gracefully.""" + _setup_requests_post_error(Exception("Connection refused")) + + result = self.manager._try_openai_api(b"wav", "en", {}) + + self.assertIsNone(result) + + def test_try_openai_api_with_language(self): + """Test OpenAI API includes language in request.""" + mock_post = _setup_requests_post_ok({"text": "result"}) + + self.manager._try_openai_api(b"wav-bytes", "fr", {}) + + # Check that language was included in the data + mock_post.assert_called_once() + call_kwargs = mock_post.call_args[1] + self.assertEqual(call_kwargs["data"]["language"], "fr") + + def test_try_openai_api_server_error(self): + """Test OpenAI API handles 500 server error via raise_for_status.""" + _setup_requests_post_status(500, Exception("500 Server Error")) + + result = self.manager._try_openai_api(b"wav", "en", {}) + + self.assertIsNone(result) + + def test_try_openai_api_auth_error(self): + """Test OpenAI API handles 401 unauthorized via raise_for_status.""" + _setup_requests_post_status(401, Exception("401 Unauthorized")) + + result = self.manager._try_openai_api(b"wav", "en", {}) + + self.assertIsNone(result) + + +class TestWhisperCppServerAPIFormat(unittest.TestCase): + """Test cases for whisper.cpp server API format.""" + + def setUp(self): + """Set up test fixtures.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + self.manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + remote_api_key="server-key", + remote_api_endpoint="/inference", + ) + + def test_try_whispercpp_server_api_success(self): + """Test successful transcription via whisper.cpp server API.""" + _setup_requests_post_ok({"text": "spoken words"}) + + result = self.manager._try_whispercpp_server_api( + b"audio-wav", "de", {"Authorization": "Bearer key"} + ) + + self.assertEqual(result, "spoken words") + + def test_try_whispercpp_server_api_404(self): + """Test whisper.cpp server returns None for 404.""" + _setup_requests_post_status(404) + + result = self.manager._try_whispercpp_server_api(b"wav", "es", {}) + + self.assertIsNone(result) + + def test_try_whispercpp_server_api_connection_error(self): + """Test whisper.cpp server handles connection errors.""" + _setup_requests_post_error(Exception("Connection refused")) + + result = self.manager._try_whispercpp_server_api(b"wav", "es", {}) + + self.assertIsNone(result) + + def test_try_whispercpp_server_api_server_error(self): + """Test whisper.cpp server handles 500 error via raise_for_status.""" + _setup_requests_post_status(500, Exception("500 Server Error")) + + result = self.manager._try_whispercpp_server_api(b"wav", "en", {}) + + self.assertIsNone(result) + + def test_try_whispercpp_server_api_auth_error(self): + """Test whisper.cpp server handles 401 unauthorized via raise_for_status.""" + _setup_requests_post_status(401, Exception("401 Unauthorized")) + + result = self.manager._try_whispercpp_server_api(b"wav", "en", {}) + + self.assertIsNone(result) + + +class TestRemoteAPIReconfiguration(unittest.TestCase): + """Test cases for remote API settings updates.""" + + def setUp(self): + """Set up test fixtures.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + self.manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + + def test_update_remote_api_url(self): + """Test updating remote API URL via reconfigure triggers reinit.""" + _setup_requests_get_ok() + + self.manager.reconfigure(remote_api_url="https://new-server:9090") + + self.assertEqual(self.manager.remote_api_url, "https://new-server:9090") + # Verify successful reinitialization after reconfigure + self.assertTrue(self.manager._model_initialized) + self.assertTrue(self.manager.model_ready) + + def test_update_remote_api_key(self): + """Test updating remote API key via reconfigure.""" + self.manager.reconfigure(remote_api_key="new-secret-key") + self.assertEqual(self.manager.remote_api_key, "new-secret-key") + + def test_update_remote_api_endpoint(self): + """Test updating remote API endpoint via reconfigure.""" + self.manager.reconfigure(remote_api_endpoint="/v1/audio/transcriptions") + self.assertEqual(self.manager.remote_api_endpoint, "/v1/audio/transcriptions") + + +class TestRemoteAPIReinitializeAfterResume(unittest.TestCase): + """Test cases for reinitialize_after_resume with remote_api engine.""" + + def test_reinitialize_after_resume_calls_init_remote_api(self): + """Test that reinitialize_after_resume dispatches to _init_remote_api.""" + from unittest.mock import patch + + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + + with patch.object(manager, "_init_remote_api") as mock_init: + manager.reinitialize_after_resume() + + mock_init.assert_called_once() + + def test_reinitialize_after_resume_without_url_does_not_raise(self): + """Test that reinitialize_after_resume with empty URL sets _model_initialized False.""" + SpeechRecognitionManager = _import_manager() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="", + ) + + # Should not raise any exception + manager.reinitialize_after_resume() + + self.assertFalse(manager._model_initialized) + + def test_reinitialize_after_resume_unknown_engine_returns_silently(self): + """Unknown engine value short-circuits reinit without raising or dispatching.""" + from unittest.mock import patch + + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + manager.engine = "bogus_engine" + + with ( + patch.object(manager, "_init_remote_api") as mock_remote, + patch.object(manager, "_init_vosk") as mock_vosk, + patch.object(manager, "_init_whisper") as mock_whisper, + patch.object(manager, "_init_whispercpp") as mock_whispercpp, + ): + manager.reinitialize_after_resume() + + mock_remote.assert_not_called() + mock_vosk.assert_not_called() + mock_whisper.assert_not_called() + mock_whispercpp.assert_not_called() + self.assertFalse(manager._model_initialized) + + +class TestHTTPSession(unittest.TestCase): + """Test cases for manager-owned requests.Session (connection pooling).""" + + def test_init_remote_api_creates_session(self): + """_init_remote_api creates a requests.Session and stores it on _http_session.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + + mock_requests = _get_mock_requests() + mock_requests.Session.assert_called() + self.assertIs(manager._http_session, mock_requests.Session.return_value) + + def test_transcribe_uses_session_post(self): + """_try_whispercpp_server_api calls self._http_session.post, not module-level requests.post.""" + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + remote_api_endpoint="/inference", + ) + + mock_session = _get_mock_session() + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"text": "hello"} + mock_session.post.return_value = mock_response + + result = manager._try_whispercpp_server_api(b"wav-bytes", "en", {}) + + self.assertEqual(result, "hello") + mock_session.post.assert_called_once() + # Module-level requests.post must NOT have been called + self.assertFalse(_get_mock_requests().post.called) + + def test_reconfigure_away_from_remote_api_closes_session(self): + """reconfigure(engine='vosk') from remote_api calls session.close() and nulls _http_session.""" + from unittest.mock import patch + + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + + session = manager._http_session + + with patch.object(manager, "_init_vosk"): + manager.reconfigure(engine="vosk") + + session.close.assert_called_once() + self.assertIsNone(manager._http_session) + + def test_reinit_closes_previous_session_before_creating_new(self): + """A second _init_remote_api call closes the previous session before creating a new one.""" + from unittest.mock import MagicMock + + SpeechRecognitionManager = _import_manager() + _setup_requests_get_ok() + + # Give Session() distinct return values for each call so we can track them + mock_requests = _get_mock_requests() + first_session = MagicMock(name="session1") + second_session = MagicMock(name="session2") + first_session.get.return_value = Mock(ok=True, status_code=200) + second_session.get.return_value = Mock(ok=True, status_code=200) + mock_requests.Session.side_effect = [first_session, second_session] + + manager = SpeechRecognitionManager( + engine="remote_api", + remote_api_url="http://localhost:9090", + ) + + self.assertIs(manager._http_session, first_session) + + # Call _init_remote_api a second time (simulates back-to-back inits) + manager._init_remote_api() + + first_session.close.assert_called_once() + self.assertIs(manager._http_session, second_session) + + +if __name__ == "__main__": + unittest.main()