chore: add operation timing logs

Author: DavidHruby1

src/audio.py

@@ -19,7 +19,7 @@
     sd = None  # type: ignore[assignment]
 
 from src.config import DEFAULT_RMS_THRESHOLD
-from src.utils import AppError, ScreamerError
+from src.utils import AppError, ScreamerError, log_duration
 
 log = logging.getLogger(__name__)
 
@@ -122,22 +122,22 @@ def calibrate(self, duration: float = 2.0) -> float:
         """Record ambient noise and return a usable silence-gate threshold."""
         _require_sd()
         try:
-            log.info("Calibrating RMS threshold for %.1fs...", duration)
-            recording = sd.rec(
-                int(duration * self._sample_rate),
-                samplerate=self._sample_rate,
-                channels=CHANNELS,
-                dtype=DTYPE,
-                device=self._device_id,
-            )
-            sd.wait()
-            noise_floor = float(np.sqrt(np.mean(recording.astype(np.float64) ** 2)))
-            threshold = noise_floor * 2.0
-            if threshold < DEFAULT_RMS_THRESHOLD:
-                threshold = DEFAULT_RMS_THRESHOLD
-            self._rms_threshold = threshold
-            log.info("Calibration done: noise_floor=%.1f, threshold=%.1f", noise_floor, threshold)
-            return threshold
+            with log_duration(log, f"Calibration for {duration:.1f}s"):
+                recording = sd.rec(
+                    int(duration * self._sample_rate),
+                    samplerate=self._sample_rate,
+                    channels=CHANNELS,
+                    dtype=DTYPE,
+                    device=self._device_id,
+                )
+                sd.wait()
+                noise_floor = float(np.sqrt(np.mean(recording.astype(np.float64) ** 2)))
+                threshold = noise_floor * 2.0
+                if threshold < DEFAULT_RMS_THRESHOLD:
+                    threshold = DEFAULT_RMS_THRESHOLD
+                self._rms_threshold = threshold
+                log.info("Calibration done: noise_floor=%.1f, threshold=%.1f", noise_floor, threshold)
+                return threshold
         except Exception as e:
             log.warning("Calibration failed: %s; using fallback %.1f", e, DEFAULT_RMS_THRESHOLD)
             self._rms_threshold = DEFAULT_RMS_THRESHOLD

src/injector.py

@@ -6,7 +6,7 @@
 import platform
 import time
 
-from src.utils import AppError, ScreamerError
+from src.utils import AppError, ScreamerError, log_duration
 
 log = logging.getLogger(__name__)
 
@@ -111,21 +111,22 @@ def _utf16_units(value: str) -> list[str]:
         return [chr(int.from_bytes(encoded[i : i + 2], "little")) for i in range(0, len(encoded), 2)]
 
     try:
-        log.info("Typing %d characters", len(text))
-        for ch in _utf16_units(text):
-            _send_unicode(ch)
-            _send_unicode(ch, key_up=True)
-
-        # Post-type key with 0.05s delay.
-        if post_key and post_key != "none":
-            vk = _POST_KEY_VK.get(post_key.lower())
-            if vk is not None:
-                time.sleep(0.05)
-                _send_vk(vk)
-                _send_vk(vk, key_up=True)
-                log.info("Post-type key pressed: %s", post_key)
-            else:
-                log.warning("Unknown post-type key: %s", post_key)
+        with log_duration(log, f"Text injection ({len(text)} chars)"):
+            log.info("Typing %d characters", len(text))
+            for ch in _utf16_units(text):
+                _send_unicode(ch)
+                _send_unicode(ch, key_up=True)
+
+            # Post-type key with 0.05s delay.
+            if post_key and post_key != "none":
+                vk = _POST_KEY_VK.get(post_key.lower())
+                if vk is not None:
+                    time.sleep(0.05)
+                    _send_vk(vk)
+                    _send_vk(vk, key_up=True)
+                    log.info("Post-type key pressed: %s", post_key)
+                else:
+                    log.warning("Unknown post-type key: %s", post_key)
 
     except ScreamerError:
         raise

src/rewrite.py

@@ -7,7 +7,7 @@
 import httpx
 
 from src.config import AppConfig, ProviderConfig, parse_custom_headers
-from src.utils import AppError, PipelineResult, ScreamerError
+from src.utils import AppError, PipelineResult, ScreamerError, log_duration
 
 log = logging.getLogger(__name__)
 
@@ -18,35 +18,36 @@ def rewrite(text: str, config: AppConfig) -> PipelineResult:
     Returns input text unchanged in ``PipelineResult.text`` if ``config.llm_enabled`` is False.
     Provider errors return the original text with ``AppError.LLM_FAILED`` as a warning.
     """
-    if not config.llm_enabled:
-        return PipelineResult(text=text)
-
-    system_prompt = config.llm_system_prompt or ""
-    language = getattr(config, "stt_language", "")
-    if language:
-        system_prompt += f"\nThe speech language is {language}."
-
-    primary = config.llm_provider()
-    fallback = config.llm_fallback_provider()
-
-    for is_fallback, provider in ((False, primary), (True, fallback.provider)):
-        if is_fallback and not fallback.enabled:
-            continue
-        if not provider.is_complete:
-            continue
-
-        try:
-            result = _call_llm(provider=provider, system_prompt=system_prompt, user_text=text)
-            if result:
-                log.debug("%s LLM rewrite: %r → %r", "Fallback" if is_fallback else "Primary", text[:60], result[:60])
-                return PipelineResult(text=result)
-        except Exception as e:
-            log.warning("%s LLM failed: %s", "Fallback" if is_fallback else "Primary", e)
-            if not fallback.enabled:
-                return PipelineResult(text=text, warnings=[AppError.LLM_FAILED])
-
-    log.debug("LLM rewrite failed or returned empty; using original text")
-    return PipelineResult(text=text, warnings=[AppError.LLM_FAILED])
+    with log_duration(log, "LLM rewrite"):
+        if not config.llm_enabled:
+            return PipelineResult(text=text)
+
+        system_prompt = config.llm_system_prompt or ""
+        language = getattr(config, "stt_language", "")
+        if language:
+            system_prompt += f"\nThe speech language is {language}."
+
+        primary = config.llm_provider()
+        fallback = config.llm_fallback_provider()
+
+        for is_fallback, provider in ((False, primary), (True, fallback.provider)):
+            if is_fallback and not fallback.enabled:
+                continue
+            if not provider.is_complete:
+                continue
+
+            try:
+                result = _call_llm(provider=provider, system_prompt=system_prompt, user_text=text)
+                if result:
+                    log.debug("%s LLM rewrite: %r → %r", "Fallback" if is_fallback else "Primary", text[:60], result[:60])
+                    return PipelineResult(text=result)
+            except Exception as e:
+                log.warning("%s LLM failed: %s", "Fallback" if is_fallback else "Primary", e)
+                if not fallback.enabled:
+                    return PipelineResult(text=text, warnings=[AppError.LLM_FAILED])
+
+        log.debug("LLM rewrite failed or returned empty; using original text")
+        return PipelineResult(text=text, warnings=[AppError.LLM_FAILED])
 
 
 def _call_llm(
@@ -78,13 +79,14 @@ def _call_llm(
         "temperature": 0.0,
     }
 
-    log.info("LLM request: url=%s model=%s", url, provider.model)
-    resp = httpx.post(url, headers=headers, json=body, timeout=30.0)
-    resp.raise_for_status()
+    with log_duration(log, f"LLM request ({provider.model})"):
+        log.info("LLM request: url=%s model=%s", url, provider.model)
+        resp = httpx.post(url, headers=headers, json=body, timeout=30.0)
+        resp.raise_for_status()
 
-    data = resp.json()
-    content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
-    return content.strip() if content else None
+        data = resp.json()
+        content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
+        return content.strip() if content else None
 
 
 # ---------------------------------------------------------------------------

src/settings_dialog.py

@@ -43,7 +43,7 @@
     save_config,
     validate_config,
 )
-from src.utils import APP_NAME
+from src.utils import APP_NAME, log_duration
 from src.startup import is_supported
 
 log = logging.getLogger(__name__)
@@ -462,26 +462,29 @@ def _validate_and_accept(self) -> None:
 
     def _on_import_env(self) -> None:
         """Import .env into the working copy (empty fields only)."""
-        self._collect()
-        self._working = import_from_env(self._working)
-        self._populate(self._working)
-        log.info("Imported .env values into settings")
+        with log_duration(log, "Settings import from .env"):
+            self._collect()
+            self._working = import_from_env(self._working)
+            self._populate(self._working)
+            log.info("Imported .env values into settings")
 
     def _on_reset(self) -> None:
         """Reset all fields to defaults."""
-        self._working = reset_config()
-        self._populate(self._working)
-        log.info("Settings reset to defaults")
+        with log_duration(log, "Settings reset to defaults"):
+            self._working = reset_config()
+            self._populate(self._working)
+            log.info("Settings reset to defaults")
 
     def _on_apply(self) -> None:
         """Apply: collect and persist without closing."""
-        self._collect()
-        if not self._show_validation_issue():
-            return
-        if not self._sync_startup_or_warn():
-            return
-        save_config(self._working)
-        log.info("Settings applied")
+        with log_duration(log, "Settings apply"):
+            self._collect()
+            if not self._show_validation_issue():
+                return
+            if not self._sync_startup_or_warn():
+                return
+            save_config(self._working)
+            log.info("Settings applied")
 
     # ------------------------------------------------------------------
     # Overrides

src/stt.py

@@ -7,7 +7,7 @@
 import httpx
 
 from src.config import AppConfig, ProviderConfig, parse_custom_headers
-from src.utils import AppError, PipelineResult, ScreamerError
+from src.utils import AppError, PipelineResult, ScreamerError, log_duration
 
 log = logging.getLogger(__name__)
 
@@ -23,37 +23,38 @@ def transcribe(audio_wav: bytes, config: AppConfig) -> PipelineResult:
 
     *config* supplies primary and fallback providers via ``AppConfig``.
     """
-    warnings: list[AppError] = []
-
-    primary = config.stt_provider()
-    fallback = config.stt_fallback_provider()
-
-    if not primary.is_complete and not fallback.is_complete:
-        raise ScreamerError(AppError.STT_FAILED, "No STT API key configured")
-
-    for is_fallback, provider, language in (
-        (False, primary, config.stt_language),
-        (True, fallback.provider, ""),
-    ):
-        if is_fallback and not fallback.enabled:
-            continue
-        if not provider.is_complete:
-            continue
-
-        try:
-            text = _call_stt(provider=provider, language=language, audio_wav=audio_wav)
-            if text is not None:
-                if is_fallback:
-                    warnings.append(AppError.STT_FALLBACK_USED)
-                return PipelineResult(text=text, warnings=warnings)
-        except ScreamerError:
-            raise
-        except Exception as e:
-            log.warning("%s STT failed: %s", "Fallback" if is_fallback else "Primary", e)
-            if not fallback.enabled:
-                raise ScreamerError(AppError.STT_FAILED, str(e)) from e
-
-    raise ScreamerError(AppError.STT_FAILED, "Both primary and fallback STT failed or returned no speech")
+    with log_duration(log, "STT transcription"):
+        warnings: list[AppError] = []
+
+        primary = config.stt_provider()
+        fallback = config.stt_fallback_provider()
+
+        if not primary.is_complete and not fallback.is_complete:
+            raise ScreamerError(AppError.STT_FAILED, "No STT API key configured")
+
+        for is_fallback, provider, language in (
+            (False, primary, config.stt_language),
+            (True, fallback.provider, ""),
+        ):
+            if is_fallback and not fallback.enabled:
+                continue
+            if not provider.is_complete:
+                continue
+
+            try:
+                text = _call_stt(provider=provider, language=language, audio_wav=audio_wav)
+                if text is not None:
+                    if is_fallback:
+                        warnings.append(AppError.STT_FALLBACK_USED)
+                    return PipelineResult(text=text, warnings=warnings)
+            except ScreamerError:
+                raise
+            except Exception as e:
+                log.warning("%s STT failed: %s", "Fallback" if is_fallback else "Primary", e)
+                if not fallback.enabled:
+                    raise ScreamerError(AppError.STT_FAILED, str(e)) from e
+
+        raise ScreamerError(AppError.STT_FAILED, "Both primary and fallback STT failed or returned no speech")
 
 
 def _call_stt(
@@ -79,26 +80,27 @@ def _call_stt(
 
     files = {"file": ("recording.wav", audio_wav, "audio/wav")}
 
-    log.info("STT request: url=%s model=%s", url, provider.model)
-    resp = httpx.post(url, headers=headers, data=data, files=files, timeout=60.0)
-    resp.raise_for_status()
+    with log_duration(log, f"STT request ({provider.model})"):
+        log.info("STT request: url=%s model=%s", url, provider.model)
+        resp = httpx.post(url, headers=headers, data=data, files=files, timeout=60.0)
+        resp.raise_for_status()
 
-    result = resp.json()
-    segments = result.get("segments", [])
+        result = resp.json()
+        segments = result.get("segments", [])
 
-    # Filter: keep if ANY segment has no_speech_prob < threshold.
-    if segments:
-        has_speech = any(seg.get("no_speech_prob", 0.0) < _NO_SPEECH_THRESHOLD for seg in segments)
-        if not has_speech:
-            log.debug("All segments above no_speech_prob threshold; filtering out")
-            raise ScreamerError(AppError.NO_SPEECH)
+        # Filter: keep if ANY segment has no_speech_prob < threshold.
+        if segments:
+            has_speech = any(seg.get("no_speech_prob", 0.0) < _NO_SPEECH_THRESHOLD for seg in segments)
+            if not has_speech:
+                log.debug("All segments above no_speech_prob threshold; filtering out")
+                raise ScreamerError(AppError.NO_SPEECH)
 
-    text = (result.get("text") or "").strip()
-    if not text:
-        raise ScreamerError(AppError.NO_SPEECH)
+        text = (result.get("text") or "").strip()
+        if not text:
+            raise ScreamerError(AppError.NO_SPEECH)
 
-    log.debug("STT result: %s", text[:80])
-    return text
+        log.debug("STT result: %s", text[:80])
+        return text
 
 
 # ---------------------------------------------------------------------------

src/utils.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+import logging
 import os
+from contextlib import contextmanager
 from dataclasses import dataclass, field
 from enum import Enum
+from time import perf_counter
 
 from PySide6.QtCore import QObject, Signal
 
@@ -51,3 +54,16 @@ class PipelineResult:
 
     text: str
     warnings: list[AppError] = field(default_factory=list)
+
+
+@contextmanager
+def log_duration(logger: logging.Logger, label: str, level: int = logging.INFO):
+    """Log how long a block takes, even if it raises."""
+    start = perf_counter()
+    try:
+        yield
+    except Exception:
+        logger.log(level, "%s failed after %.3fs", label, perf_counter() - start)
+        raise
+    else:
+        logger.log(level, "%s finished in %.3fs", label, perf_counter() - start)