diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 8d00575e..a602db23 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -707,7 +707,7 @@ In overlay mode, `mic_osd_style` picks one of three visualizations (notification - `waveform` — the full themed waveform with transcript preview (default) - `vu_meter` — a VU meter -- `pill` — a compact monochrome status pill (no transcript text): idle dots, live bars while recording, a travelling wave while processing, a pulse on error, a checkmark on success +- `pill` — a compact monochrome status pill: idle dots, live bars while recording, a travelling wave while processing, a pulse on error, and a checkmark on success. Can optionally show an animated live transcript on realtime backends that stream partial results (OpenAI, ElevenLabs). ![Pill OSD states](assets/pill-states.png) @@ -723,6 +723,27 @@ Restart the service after changing the style: systemctl --user restart hyprwhspr ``` +#### Pill live transcript + +Shows the last few words while recording. Opt-in; needs `mic_osd_style: pill` plus a `realtime-ws` backend that streams partial transcripts (OpenAI, ElevenLabs — not Gemini yet). Final transcription and paste are unaffected. + +```jsonc +{ + "transcription_backend": "realtime-ws", + "websocket_provider": "elevenlabs", + "websocket_model": "scribe_v2_realtime", + "realtime_mode": "transcribe", + "mic_osd_style": "pill", + "mic_osd_pill_transcript_enabled": true +} +``` + +| Setting | Default | Description | +|---|---:|---| +| `mic_osd_pill_transcript_enabled` | `false` | Enable the pill transcript. | +| `mic_osd_pill_transcript_word_limit` | `4` | Recent words shown (1–12). | +| `mic_osd_pill_transcript_idle_timeout_ms` | `1400` | Hide after this much idle time; `0` disables. | + #### GNOME/Mutter waveform overlay GNOME users who want an animated waveform instead of notifications can install the opt-in GNOME Shell extension in `contrib/gnome-shell-extension/` — it draws inside gnome-shell, always-on-top, without stealing focus: diff --git a/docs/assets/pill-live-transcript.gif b/docs/assets/pill-live-transcript.gif new file mode 100644 index 00000000..ecf2bb6c Binary files /dev/null and b/docs/assets/pill-live-transcript.gif differ diff --git a/lib/mic_osd/transcript_preview.py b/lib/mic_osd/transcript_preview.py new file mode 100644 index 00000000..6a699eec --- /dev/null +++ b/lib/mic_osd/transcript_preview.py @@ -0,0 +1,302 @@ +"""Config and animation state for the pill live-transcript preview.""" + +from __future__ import annotations + +from dataclasses import dataclass +import time +from typing import Callable, Optional, Sequence, Tuple + + +def _clamp(value, minimum, maximum): + return max(minimum, min(maximum, value)) + + +def _number(value, default): + try: + return float(value) + except (TypeError, ValueError): + return float(default) + + +def _integer(value, default): + try: + return int(value) + except (TypeError, ValueError): + return int(default) + + +@dataclass(frozen=True) +class PillTranscriptConfig: + """Validated settings for the transcript shown above the compact pill.""" + + enabled: bool = False + word_limit: int = 4 + idle_timeout_seconds: float = 1.40 + + # Visual tuning is intentionally part of the pill style, not public config. + enter_seconds: float = 0.20 + exit_seconds: float = 0.20 + stagger_seconds: float = 0.028 + font_family: str = "sans-serif" + font_size: float = 17.0 + offset_y: float = 7.0 + rise_px: float = 9.0 + max_width: float = 320.0 + + @classmethod + def from_getter(cls, get_setting: Callable[[str, object], object]): + return cls( + enabled=bool( + get_setting("mic_osd_pill_transcript_enabled", cls.enabled) + ), + word_limit=_clamp( + _integer( + get_setting( + "mic_osd_pill_transcript_word_limit", cls.word_limit + ), + cls.word_limit, + ), + 1, + 12, + ), + idle_timeout_seconds=_clamp( + _number( + get_setting( + "mic_osd_pill_transcript_idle_timeout_ms", + cls.idle_timeout_seconds * 1000, + ), + cls.idle_timeout_seconds * 1000, + ) + / 1000.0, + 0.0, + 30.0, + ), + ) + + @classmethod + def load(cls): + """Load from the normal hyprwhspr config, falling back safely.""" + try: + try: + from ..src.config_manager import ConfigManager + except (ImportError, ValueError): + from src.config_manager import ConfigManager + return cls.from_getter(ConfigManager(verbose=False).get_setting) + except Exception: + return cls() + + +@dataclass(frozen=True) +class RenderWord: + """One word to paint for the current animation frame.""" + + text: str + source: str + index: int + alpha: float + y_offset: float = 0.0 + matched_from: Optional[int] = None + layout_progress: float = 1.0 + + +@dataclass(frozen=True) +class TranscriptFrame: + previous_words: Tuple[str, ...] + current_words: Tuple[str, ...] + words: Tuple[RenderWord, ...] + + +def _ease_out_cubic(value: float) -> float: + value = _clamp(value, 0.0, 1.0) + return 1.0 - (1.0 - value) ** 3 + + +def _ease_in_out(value: float) -> float: + value = _clamp(value, 0.0, 1.0) + return value * value * (3.0 - 2.0 * value) + + +def _progress(elapsed: float, duration: float, delay: float = 0.0) -> float: + if duration <= 0.0: + return 1.0 + return _clamp((elapsed - delay) / duration, 0.0, 1.0) + + +def _lcs_matches( + previous: Sequence[str], current: Sequence[str] +) -> Tuple[Tuple[int, int], ...]: + """Duplicate-safe longest-common-subsequence matches for tiny word windows.""" + + rows = len(previous) + 1 + cols = len(current) + 1 + table = [[0] * cols for _ in range(rows)] + + for old_index in range(len(previous) - 1, -1, -1): + for new_index in range(len(current) - 1, -1, -1): + if previous[old_index] == current[new_index]: + table[old_index][new_index] = ( + table[old_index + 1][new_index + 1] + 1 + ) + else: + table[old_index][new_index] = max( + table[old_index + 1][new_index], + table[old_index][new_index + 1], + ) + + matches = [] + old_index = 0 + new_index = 0 + while old_index < len(previous) and new_index < len(current): + if previous[old_index] == current[new_index]: + matches.append((old_index, new_index)) + old_index += 1 + new_index += 1 + elif table[old_index + 1][new_index] >= table[old_index][new_index + 1]: + old_index += 1 + else: + new_index += 1 + return tuple(matches) + + +class PillTranscriptAnimator: + """Latest-state transcript animation with no event queue. + + Incoming partials replace the target immediately. Character-by-character + corrections to the last token update in place instead of restarting the + animation. Word-boundary changes animate, so 200–250 WPM speech remains + readable while the renderer always converges on the newest four words. + """ + + def __init__( + self, + config: PillTranscriptConfig, + clock: Callable[[], float] = time.monotonic, + ): + self.config = config + self._clock = clock + self.previous_words: Tuple[str, ...] = () + self.current_words: Tuple[str, ...] = () + self.matches: Tuple[Tuple[int, int], ...] = () + self.transition_started_at = self._clock() + self.last_activity_at = self.transition_started_at + + def _extract_words(self, text: str) -> Tuple[str, ...]: + words = tuple((text or "").split()) + return words[-self.config.word_limit :] + + def set_text(self, text: str, now: Optional[float] = None) -> bool: + now = self._clock() if now is None else now + words = self._extract_words(text) + + if words: + self.last_activity_at = now + + if words == self.current_words: + return False + + # Realtime providers often revise only the unfinished final token + # ("trans" -> "transcript"). Do not restart a 200ms animation for every + # character; preserve the current transition and replace that token. + if ( + words + and self.current_words + and len(words) == len(self.current_words) + and words[:-1] == self.current_words[:-1] + ): + self.current_words = words + return True + + self._begin_transition(words, now) + return True + + def clear(self, now: Optional[float] = None) -> bool: + now = self._clock() if now is None else now + if not self.current_words: + return False + self._begin_transition((), now) + return True + + def _begin_transition(self, words: Tuple[str, ...], now: float): + self.previous_words = self.current_words + self.current_words = words + self.matches = _lcs_matches(self.previous_words, self.current_words) + self.transition_started_at = now + + def _expire_if_idle(self, now: float): + timeout = self.config.idle_timeout_seconds + if ( + timeout > 0 + and self.current_words + and (now - self.last_activity_at) >= timeout + ): + self._begin_transition((), now) + + def frame(self, now: Optional[float] = None) -> TranscriptFrame: + now = self._clock() if now is None else now + self._expire_if_idle(now) + + if not self.config.enabled: + return TranscriptFrame((), (), ()) + + elapsed = max(0.0, now - self.transition_started_at) + stable_by_new = {new: old for old, new in self.matches} + stable_old = {old for old, _ in self.matches} + rendered = [] + + # Removed words leave first, rising away from the pill. They are never + # queued; a newer transcript immediately replaces this transition. + for old_index, word in enumerate(self.previous_words): + if old_index in stable_old: + continue + reverse_index = len(self.previous_words) - 1 - old_index + delay = reverse_index * self.config.stagger_seconds * 0.55 + progress = _ease_in_out( + _progress(elapsed, self.config.exit_seconds, delay) + ) + rendered.append( + RenderWord( + word, + "previous", + old_index, + 1.0 - progress, + y_offset=-self.config.rise_px * progress, + ) + ) + + for new_index, word in enumerate(self.current_words): + matched_from = stable_by_new.get(new_index) + if matched_from is not None: + layout_progress = _ease_out_cubic( + _progress(elapsed, self.config.enter_seconds) + ) + rendered.append( + RenderWord( + word, + "current", + new_index, + 1.0, + matched_from=matched_from, + layout_progress=layout_progress, + ) + ) + continue + + delay = new_index * self.config.stagger_seconds + progress = _ease_out_cubic( + _progress(elapsed, self.config.enter_seconds, delay) + ) + rendered.append( + RenderWord( + word, + "current", + new_index, + progress, + y_offset=self.config.rise_px * (1.0 - progress), + ) + ) + + return TranscriptFrame( + self.previous_words, + self.current_words, + tuple(rendered), + ) diff --git a/lib/mic_osd/visualizations/pill.py b/lib/mic_osd/visualizations/pill.py index 85824083..6c2477f0 100644 --- a/lib/mic_osd/visualizations/pill.py +++ b/lib/mic_osd/visualizations/pill.py @@ -12,10 +12,13 @@ class PillVisualization(BaseVisualization): """A compact black pill with white audio bars and state animations.""" - show_preview = False + show_preview = True + preview_mode = "pill" PILL_WIDTH = 126 PILL_HEIGHT = 42 + # Surfaces this tall are reserving space for the preview text above. + PREVIEW_HEIGHT_THRESHOLD = PILL_HEIGHT + 34 NUM_BARS = 13 BAR_WIDTH = 3.0 BAR_GAP = 3.0 @@ -120,12 +123,16 @@ def update(self, level: float, samples: np.ndarray = None): def _pill_geometry(self, width: int, height: int): pill_w = min(self.PILL_WIDTH, width - 4) pill_h = min(self.PILL_HEIGHT, height - 4) - return ( - (width - pill_w) / 2.0, - (height - pill_h) / 2.0, - pill_w, - pill_h, - ) + x = (width - pill_w) / 2.0 + + # A taller surface reserves space for live text above the pill. Keep + # legacy standalone/preview-disabled geometry unchanged at the old size. + if height >= self.PREVIEW_HEIGHT_THRESHOLD: + y = height - pill_h - 4.0 + else: + y = (height - pill_h) / 2.0 + + return x, y, pill_w, pill_h def _success_fade(self) -> float: if self.state_manager.current_state != VisualizerState.SUCCESS: diff --git a/lib/mic_osd/window.py b/lib/mic_osd/window.py index 091a8bc2..74795a8b 100644 --- a/lib/mic_osd/window.py +++ b/lib/mic_osd/window.py @@ -7,6 +7,8 @@ from __future__ import annotations +import time + import gi gi.require_version('Gtk', '4.0') gi.require_version('Gdk', '4.0') @@ -20,6 +22,7 @@ from gi.repository import Gtk, Gdk, GLib from .theme import theme +from .transcript_preview import PillTranscriptAnimator, PillTranscriptConfig if LAYER_SHELL_AVAILABLE: from gi.repository import Gtk4LayerShell @@ -36,7 +39,13 @@ class OSDWindow(Gtk.Window): PREVIEW_WORD_LIMIT = 14 PREVIEW_TIMER_RESERVE = 58 - def __init__(self, visualization, width=300, height=60): + def __init__( + self, + visualization, + width=300, + height=60, + transcript_config=None, + ): """ Initialize the OSD window. @@ -44,14 +53,43 @@ def __init__(self, visualization, width=300, height=60): visualization: A BaseVisualization instance width: Window width in pixels height: Window height in pixels + transcript_config: Validated PillTranscriptConfig for pill previews. """ super().__init__() self.visualization = visualization + self._is_pill_preview = getattr(visualization, 'preview_mode', None) == 'pill' + self._pill_transcript_config = ( + transcript_config + if transcript_config is not None + else PillTranscriptConfig.load() + ) + if self._is_pill_preview and self._pill_transcript_config.enabled: + height = max( + height, + int( + getattr(self.visualization, 'PILL_HEIGHT', 42) + + self._pill_transcript_config.font_size + + self._pill_transcript_config.offset_y + + 16 + ), + # Must clear the pill's own preview-mode threshold. + getattr(self.visualization, 'PREVIEW_HEIGHT_THRESHOLD', 76) + 1, + ) + width = max( + width, + int(min(800.0, self._pill_transcript_config.max_width + 24.0)), + ) + self._width = width self._height = height self._preview_text = "" self._visualizer_state = "recording" + self._pill_transcript_animator = None + if self._is_pill_preview: + self._pill_transcript_animator = PillTranscriptAnimator( + self._pill_transcript_config + ) # Layer shell MUST be initialized immediately after window creation # and BEFORE any other window configuration @@ -60,11 +98,11 @@ def __init__(self, visualization, width=300, height=60): self._setup_drawing_area() def _setup_layer_shell(self): - """Configure layer shell for overlay behavior.""" + """Configure layer shell behavior.""" if not LAYER_SHELL_AVAILABLE: return - # Initialize layer shell - MUST be called before window is realized + # Initialize layer shell - MUST be initialized before other calls Gtk4LayerShell.init_for_window(self) # Set namespace for window rules @@ -121,7 +159,9 @@ def _on_draw(self, area, cr, width, height): # Draw the visualization self.visualization.draw(cr, width, height) - if getattr(self.visualization, 'show_preview', True): + if self._is_pill_preview: + self._draw_pill_preview_text(cr, width, height) + elif getattr(self.visualization, 'show_preview', True): self._draw_preview_text(cr, width, height) def update(self, level: float, samples=None): @@ -138,11 +178,20 @@ def update(self, level: float, samples=None): def set_preview_text(self, text: str): """Set compact transcript preview text.""" self._preview_text = (text or "").rstrip('\r\n') + if self._pill_transcript_animator is not None: + self._pill_transcript_animator.set_text(self._preview_text) self.drawing_area.queue_draw() def set_visualizer_state(self, state: str): """Track visualizer state so partial previews only render while recording.""" + previous_state = self._visualizer_state self._visualizer_state = (state or "recording").lower() + if ( + self._pill_transcript_animator is not None + and previous_state == "recording" + and self._visualizer_state != "recording" + ): + self._pill_transcript_animator.clear() self.drawing_area.queue_draw() def _draw_preview_text(self, cr: cairo.Context, width: int, height: int): @@ -180,6 +229,148 @@ def _draw_preview_text(self, cr: cairo.Context, width: int, height: int): cr.move_to(padding, y) cr.show_text(text) + def _draw_pill_preview_text( + self, + cr: cairo.Context, + width: int, + height: int, + ): + animator = self._pill_transcript_animator + config = self._pill_transcript_config + if ( + animator is None + or not config.enabled + or self._visualizer_state != "recording" + ): + return + + frame = animator.frame(time.monotonic()) + if not frame.words: + return + + cr.select_font_face( + config.font_family, + cairo.FONT_SLANT_NORMAL, + cairo.FONT_WEIGHT_NORMAL, + ) + + max_width = min(config.max_width, max(0.0, width - 20.0)) + effective_font_size = config.font_size + cr.set_font_size(effective_font_size) + previous_texts = frame.previous_words + current_texts = frame.current_words + _, previous_total = self._word_layout(cr, previous_texts, width) + _, current_total = self._word_layout(cr, current_texts, width) + + largest_total = max(previous_total, current_total) + if largest_total > max_width and largest_total > 0: + effective_font_size = max( + 8.0, + effective_font_size * max_width / largest_total, + ) + cr.set_font_size(effective_font_size) + + previous_texts = tuple( + self._ellipsize_pill_token(cr, word, max_width) + for word in previous_texts + ) + current_texts = tuple( + self._ellipsize_pill_token(cr, word, max_width) + for word in current_texts + ) + previous_positions, _ = self._word_layout(cr, previous_texts, width) + current_positions, _ = self._word_layout(cr, current_texts, width) + + pill_geometry = getattr(self.visualization, '_pill_geometry', None) + if callable(pill_geometry): + _, pill_y, _, _ = pill_geometry(width, height) + else: + pill_y = height - 46.0 + baseline = pill_y - config.offset_y + + for word in frame.words: + if word.alpha <= 0.01: + continue + + if word.source == "previous": + if word.index >= len(previous_positions): + continue + x = previous_positions[word.index] + text = previous_texts[word.index] + else: + if word.index >= len(current_positions): + continue + x = current_positions[word.index] + text = current_texts[word.index] + if ( + word.matched_from is not None + and word.matched_from < len(previous_positions) + ): + old_x = previous_positions[word.matched_from] + x = old_x + ( + x - old_x + ) * word.layout_progress + + y = baseline + word.y_offset + + # A restrained shadow keeps white text readable over bright windows + # without introducing a visible badge or background rectangle. + cr.set_source_rgba(0.0, 0.0, 0.0, 0.55 * word.alpha) + cr.move_to(x, y + 2.0) + cr.show_text(text) + + cr.set_source_rgba(1.0, 1.0, 1.0, 0.97 * word.alpha) + cr.move_to(x, y) + cr.show_text(text) + + def _word_layout(self, cr: cairo.Context, words, width: int): + if not words: + return (), 0.0 + + # Cairo's ink width for a space is zero. Layout with glyph advances so + # whitespace and side bearings are preserved between separately drawn + # words instead of making them visually run together. + space_width = self._text_advance(cr, " ") + widths = [self._text_advance(cr, word) for word in words] + total_width = sum(widths) + space_width * max(0, len(words) - 1) + x = (width - total_width) / 2.0 + positions = [] + for word_width in widths: + positions.append(x) + x += word_width + space_width + return tuple(positions), total_width + + @staticmethod + def _bisect_truncate(measure, available: float, length: int) -> int: + """Longest length in [0, length] fitting `available`.""" + low, high = 0, length + while low < high: + mid = (low + high + 1) // 2 + if measure(mid) <= available: + low = mid + else: + high = mid - 1 + return low + + def _ellipsize_pill_token( + self, + cr: cairo.Context, + text: str, + max_width: float, + ) -> str: + if self._text_advance(cr, text) <= max_width: + return text + + suffix = "…" + available = max_width - self._text_advance(cr, suffix) + if available <= 0: + return "" + + low = self._bisect_truncate( + lambda n: self._text_advance(cr, text[:n]), available, len(text) + ) + return text[:low].rstrip() + suffix if low else suffix + @staticmethod def _text_extent(extents, field: str, index: int) -> float: if hasattr(extents, field): @@ -189,6 +380,13 @@ def _text_extent(extents, field: str, index: int) -> float: def _text_width(self, cr: cairo.Context, text: str) -> float: return self._text_extent(cr.text_extents(text), 'width', 2) + def _text_advance(self, cr: cairo.Context, text: str) -> float: + extents = cr.text_extents(text) + advance = self._text_extent(extents, 'x_advance', 4) + if advance > 0: + return advance + return self._text_extent(extents, 'width', 2) + def _text_height(self, cr: cairo.Context, text: str) -> float: return self._text_extent(cr.text_extents(text), 'height', 3) @@ -210,14 +408,9 @@ def _ellipsize(self, cr: cairo.Context, text: str, max_width: float) -> str: if available <= 0: return "" - low = 0 - high = len(text) - while low < high: - mid = (low + high + 1) // 2 - if self._text_width(cr, text[-mid:]) <= available: - low = mid - else: - high = mid - 1 + low = self._bisect_truncate( + lambda n: self._text_width(cr, text[-n:]) if n else 0.0, available, len(text) + ) truncated = text[-low:].lstrip() return prefix + truncated if truncated else prefix diff --git a/lib/src/backends/realtime_ws_backend.py b/lib/src/backends/realtime_ws_backend.py index ad7a94af..8d2aab58 100644 --- a/lib/src/backends/realtime_ws_backend.py +++ b/lib/src/backends/realtime_ws_backend.py @@ -232,7 +232,7 @@ def _send_direct(audio_chunk: np.ndarray): delay = self.config.get_setting('realtime_transcription_delay', 'low') self._realtime_client.set_transcription_delay(delay) - if self._is_realtime_whisper_preview_enabled(provider_id, model_id, realtime_mode): + if self._is_partial_preview_enabled(provider_id, model_id, realtime_mode): self._realtime_client.set_partial_transcript_callback(self._realtime_partial_callback) else: self._realtime_client.set_partial_transcript_callback(None) @@ -386,30 +386,58 @@ def get_streaming_callback(self) -> Optional[Callable]: # Clear server buffer before starting new recording self._realtime_client.clear_audio_buffer() + self._clear_realtime_partial_preview() return self._realtime_streaming_callback return None def apply_partial_callback(self, callback: Optional[Callable[[str], None]]) -> None: - """Apply the partial-preview callback to a connected client (storage lives on the manager).""" - if self._realtime_client and hasattr(self._realtime_client, 'set_partial_transcript_callback'): - provider_id = self.config.get_setting('websocket_provider') - model_id = self.config.get_setting('websocket_model') - realtime_mode = self.config.get_setting('realtime_mode', 'transcribe') - if self._is_realtime_whisper_preview_enabled(provider_id, model_id, realtime_mode): - self._realtime_client.set_partial_transcript_callback(callback) - else: - self._realtime_client.set_partial_transcript_callback(None) - self._clear_realtime_partial_preview() + """Apply the partial-preview callback to the active realtime provider.""" + if not self._realtime_client: + return - def _is_realtime_whisper_preview_enabled(self, provider_id: str, model_id: str, realtime_mode: str) -> bool: - return ( - self.config.get_setting('mic_osd_enabled', True) - and provider_id == 'openai' - and model_id == 'gpt-realtime-whisper' - and realtime_mode == 'transcribe' - and self._realtime_partial_callback is not None + provider_id = self.config.get_setting('websocket_provider') + model_id = self.config.get_setting('websocket_model') + realtime_mode = self.config.get_setting('realtime_mode', 'transcribe') + enabled = self._is_partial_preview_enabled( + provider_id, + model_id, + realtime_mode, ) + if hasattr(self._realtime_client, 'set_partial_transcript_callback'): + self._realtime_client.set_partial_transcript_callback( + callback if enabled else None + ) + if not enabled: + self._clear_realtime_partial_preview() + + def _is_partial_preview_enabled( + self, + provider_id: str, + model_id: str, + realtime_mode: str, + ) -> bool: + if ( + not self.config.get_setting('mic_osd_enabled', True) + or realtime_mode != 'transcribe' + or self._realtime_partial_callback is None + ): + return False + + # Pill: any provider with partial-transcript support qualifies. + if self.config.get_setting('mic_osd_style', 'waveform') == 'pill': + return bool( + self._realtime_client is not None + and hasattr(self._realtime_client, 'set_partial_transcript_callback') + and self.config.get_setting('mic_osd_pill_transcript_enabled', False) + ) + + # Waveform: only gpt-realtime-whisper supports this today. + if provider_id == 'openai': + return model_id == 'gpt-realtime-whisper' + + return False + def _clear_realtime_partial_preview(self) -> None: if not self._realtime_partial_callback: return @@ -473,6 +501,7 @@ def discard_audio(self) -> None: if self._realtime_client: try: self._realtime_client.clear_audio_buffer() + self._clear_realtime_partial_preview() except Exception as e: print(f'[REALTIME] Failed to discard audio: {e}', flush=True) @@ -483,6 +512,7 @@ def close(self) -> None: self._realtime_client.close() self._realtime_client = None self._realtime_streaming_callback = None + self._clear_realtime_partial_preview() except Exception as e: print(f"[WARN] Failed to cleanup realtime client: {e}") diff --git a/lib/src/config_manager.py b/lib/src/config_manager.py index 1699b191..cc234632 100644 --- a/lib/src/config_manager.py +++ b/lib/src/config_manager.py @@ -160,6 +160,10 @@ def __init__(self, verbose: bool = True): # Visual feedback settings 'mic_osd_enabled': True, # Show microphone visualization overlay during recording 'mic_osd_style': 'waveform', # Overlay style: 'waveform', 'vu_meter' or 'pill' + # Live transcript above the pill OSD (ElevenLabs Scribe v2 Realtime) + 'mic_osd_pill_transcript_enabled': False, + 'mic_osd_pill_transcript_word_limit': 4, + 'mic_osd_pill_transcript_idle_timeout_ms': 1400, # Banner duration (ms) for non-critical desktop notifications. These are also # marked transient so they never accumulate in the notification center; # critical errors ignore this and persist. (GNOME Shell ignores -t entirely.) diff --git a/lib/src/elevenlabs_realtime_client.py b/lib/src/elevenlabs_realtime_client.py index 2adb1f93..2d9e4083 100644 --- a/lib/src/elevenlabs_realtime_client.py +++ b/lib/src/elevenlabs_realtime_client.py @@ -7,7 +7,7 @@ import asyncio import threading import time -from typing import Optional +from typing import Callable, Optional try: from .realtime_base import RealtimeAudioClientBase @@ -45,6 +45,7 @@ def __init__(self): self._partial_transcript = '' self._transcript_generation = 0 self._committed_segments = [] + self._partial_transcript_callback: Optional[Callable[[str], None]] = None # Auto-commit helper: # ElevenLabs punctuation often improves on committed transcripts. @@ -290,6 +291,7 @@ def on_partial_transcript(data): text = data.get('text', '') with self.lock: self._partial_transcript = text + self._emit_partial_transcript() def on_committed_transcript(data): text = data.get('text', '') @@ -304,6 +306,7 @@ def on_committed_transcript(data): self._committed_segments.append(text.strip()) self._transcript_generation += 1 self._last_transcript_audio_activity_id = self._audio_activity_id + self._emit_partial_transcript() self._transcript_event.set() def on_committed_transcript_with_timestamps(data): @@ -319,6 +322,7 @@ def on_committed_transcript_with_timestamps(data): self._committed_segments.append(text.strip()) self._transcript_generation += 1 self._last_transcript_audio_activity_id = self._audio_activity_id + self._emit_partial_transcript() self._transcript_event.set() def on_error(error): @@ -446,6 +450,38 @@ def update_language(self, language: Optional[str]): self.language = language print(f'[ELEVENLABS] Language set to: {language or "auto-detect"}', flush=True) + def set_partial_transcript_callback( + self, + callback: Optional[Callable[[str], None]], + ) -> None: + """Set the live-preview callback retained across reconnects.""" + with self.lock: + self._partial_transcript_callback = callback + if callback is not None: + self._emit_partial_transcript() + + def _committed_text_locked(self) -> str: + """Joined committed transcript. Call with self.lock held.""" + parts = [part for part in self._committed_segments if part] + return ' '.join(parts).strip() + + def _emit_partial_transcript(self) -> None: + with self.lock: + callback = self._partial_transcript_callback + committed = self._committed_text_locked() + partial = self._partial_transcript.strip() + preview = f'{committed} {partial}'.strip() if partial else committed + + if callback is None: + return + try: + callback(preview) + except Exception as exc: + print( + f'[ELEVENLABS] Failed to update partial transcript preview: {exc}', + flush=True, + ) + def clear_audio_buffer(self): """Clear state before starting a new recording""" with self.lock: @@ -462,6 +498,7 @@ def clear_audio_buffer(self): self._last_drop_log_time = 0.0 self._queue_cond.notify_all() self._transcript_event.clear() + self._emit_partial_transcript() def commit_and_get_text(self, timeout: float = 30.0) -> str: """ @@ -478,16 +515,10 @@ def commit_and_get_text(self, timeout: float = 30.0) -> str: return '' try: - def _full_committed_text_locked() -> str: - """Return concatenated committed transcript for current recording.""" - parts = [p for p in self._committed_segments if p] - # Use single spaces to stitch segments; preserve order. - return ' '.join(parts).strip() - # If we already have a committed transcript AND there is no new queued audio since then, # we can return immediately (common case: server VAD committed before user stops). with self.lock: - existing_transcript = _full_committed_text_locked() + existing_transcript = self._committed_text_locked() existing_generation = self._transcript_generation has_new_audio_since_transcript = ( self._audio_activity_id != self._last_transcript_audio_activity_id @@ -546,12 +577,12 @@ async def _commit(): # (often where punctuation gets finalized). We wait for a short quiet window and # return the latest committed text we see. best_generation = self._transcript_generation - best_text = _full_committed_text_locked() + best_text = self._committed_text_locked() # If we didn't have an existing transcript, accept the first one we get. if existing_generation == 0 and self._transcript_generation > 0: best_generation = self._transcript_generation - best_text = _full_committed_text_locked() + best_text = self._committed_text_locked() if 'best_generation' in locals(): settle_deadline = min(deadline, time.time() + 0.6) @@ -564,7 +595,7 @@ async def _commit(): # Incorporate any newer commit if self._transcript_generation > best_generation: best_generation = self._transcript_generation - best_text = _full_committed_text_locked() + best_text = self._committed_text_locked() result = best_text self._current_transcript = '' @@ -586,7 +617,7 @@ async def _commit(): # Fallback: return latest committed transcript if present, else partial, else empty. with self.lock: - full_text = _full_committed_text_locked() + full_text = self._committed_text_locked() if full_text: result = full_text self._current_transcript = '' diff --git a/lib/src/whisper_manager.py b/lib/src/whisper_manager.py index b744574b..517c5ada 100644 --- a/lib/src/whisper_manager.py +++ b/lib/src/whisper_manager.py @@ -124,7 +124,12 @@ def _initialize_backend_locked(self) -> bool: print(f"[WARN] Failed to clean up previous backend: {e}", flush=True) self._backend = backend_cls(self) - return self._backend.initialize() + initialized = self._backend.initialize() + if initialized and self._backend.name == 'realtime-ws': + self._backend.apply_partial_callback( + self._realtime_partial_callback + ) + return initialized except Exception as e: print(f"ERROR: Failed to initialize Whisper manager: {e}") diff --git a/share/config.schema.json b/share/config.schema.json index 8a51a767..4cc0a711 100644 --- a/share/config.schema.json +++ b/share/config.schema.json @@ -440,7 +440,26 @@ "type": "string", "enum": ["waveform", "vu_meter", "pill"], "default": "waveform", - "description": "Visualization style for the mic OSD overlay. 'waveform' shows the full themed waveform with transcript preview, 'vu_meter' a VU meter, 'pill' a compact monochrome status pill without preview text. Restart the service after changing." + "description": "Visualization style for the mic OSD overlay. 'waveform' shows the full themed waveform with transcript preview, 'vu_meter' a VU meter, and 'pill' a compact monochrome status pill with optional live transcript. Restart the service after changing." + }, + "mic_osd_pill_transcript_enabled": { + "type": "boolean", + "default": false, + "description": "Show a live transcript above the pill OSD. Requires a realtime-ws backend that streams partial transcripts (OpenAI, ElevenLabs)." + }, + "mic_osd_pill_transcript_word_limit": { + "type": "integer", + "minimum": 1, + "maximum": 12, + "default": 4, + "description": "Maximum number of recent words shown above the pill." + }, + "mic_osd_pill_transcript_idle_timeout_ms": { + "type": "number", + "minimum": 0, + "maximum": 30000, + "default": 1400, + "description": "Hide the preview after this many idle milliseconds. 0 disables the timeout." }, "notification_timeout_ms": { "type": "integer", diff --git a/tests/test_elevenlabs_pill_preview.py b/tests/test_elevenlabs_pill_preview.py new file mode 100644 index 00000000..c16b20e6 --- /dev/null +++ b/tests/test_elevenlabs_pill_preview.py @@ -0,0 +1,198 @@ +import sys +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "lib" / "src")) +sys.path.insert(0, str(ROOT / "lib")) + +from backends.realtime_ws_backend import RealtimeWsBackend +from elevenlabs_realtime_client import ElevenLabsRealtimeClient + + +class FakeConfig: + def __init__(self, values=None): + self.values = { + "transcription_backend": "realtime-ws", + "websocket_provider": "elevenlabs", + "websocket_model": "scribe_v2_realtime", + "realtime_mode": "transcribe", + "mic_osd_enabled": True, + "mic_osd_style": "pill", + "mic_osd_pill_transcript_enabled": True, + } + self.values.update(values or {}) + + def get_setting(self, key, default=None): + return self.values.get(key, default) + + +class FakeManager: + def __init__(self, config=None): + self.config = config or FakeConfig() + self.temp_dir = None + self.ready = True + self.current_model = None + self._last_use_time = 0.0 + self._realtime_partial_callback = None + + +class FakeRealtimeClient: + def __init__(self): + self.connected = True + self.callback = None + self.callback_updates = [] + + def set_partial_transcript_callback(self, callback): + self.callback = callback + self.callback_updates.append(callback) + + def clear_audio_buffer(self): + if self.callback: + self.callback("") + + +class ElevenLabsClientPreviewTests(unittest.TestCase): + def setUp(self): + self.client = ElevenLabsRealtimeClient() + + def test_combines_committed_and_partial_text(self): + previews = [] + self.client.set_partial_transcript_callback(previews.append) + with self.client.lock: + self.client._committed_segments = ["This is already"] + self.client._partial_transcript = "committed and live" + + self.client._emit_partial_transcript() + + self.assertEqual(previews[-1], "This is already committed and live") + + def test_callback_survives_connection_replacement(self): + previews = [] + self.client.set_partial_transcript_callback(previews.append) + self.client._connection = object() + self.client._connection = object() + with self.client.lock: + self.client._partial_transcript = "after reconnect" + + self.client._emit_partial_transcript() + + self.assertEqual(previews[-1], "after reconnect") + + def test_clear_audio_buffer_clears_preview(self): + previews = [] + self.client.set_partial_transcript_callback(previews.append) + with self.client.lock: + self.client._partial_transcript = "stale" + + self.client.clear_audio_buffer() + + self.assertEqual(previews[-1], "") + + +class ElevenLabsRealtimePreviewTests(unittest.TestCase): + def setUp(self): + self.manager = FakeManager() + self.backend = RealtimeWsBackend(self.manager) + self.client = FakeRealtimeClient() + self.backend._realtime_client = self.client + + def _apply(self): + previews = [] + self.manager._realtime_partial_callback = previews.append + self.backend.apply_partial_callback(previews.append) + return previews + + def test_enabled_pill_registers_client_callback(self): + previews = self._apply() + + self.client.callback("live words") + + self.assertEqual(previews, ["live words"]) + + def test_callback_is_updated_without_duplicate_sdk_handlers(self): + first = self._apply() + second = [] + self.manager._realtime_partial_callback = second.append + + self.backend.apply_partial_callback(second.append) + self.client.callback("latest") + + self.assertEqual(first, []) + self.assertEqual(second, ["latest"]) + self.assertEqual(len(self.client.callback_updates), 2) + + def test_disabled_pill_preview_unregisters_callback_and_clears(self): + self.manager.config.values["mic_osd_pill_transcript_enabled"] = False + previews = self._apply() + + self.assertIsNone(self.client.callback) + self.assertEqual(previews, [""]) + + def test_pill_preview_is_disabled_when_setting_is_absent(self): + self.manager.config.values.pop("mic_osd_pill_transcript_enabled") + previews = self._apply() + + self.assertIsNone(self.client.callback) + self.assertEqual(previews, [""]) + + def test_non_pill_style_does_not_enable_elevenlabs_preview(self): + self.manager.config.values["mic_osd_style"] = "waveform" + previews = self._apply() + + self.assertIsNone(self.client.callback) + self.assertEqual(previews, [""]) + + def test_openai_preview_remains_enabled_for_waveform(self): + self.manager.config.values.update({ + "websocket_provider": "openai", + "websocket_model": "gpt-realtime-whisper", + "mic_osd_style": "waveform", + }) + + self._apply() + + self.assertIsNotNone(self.client.callback) + + def test_openai_preview_is_also_shown_in_pill(self): + self.manager.config.values.update({ + "websocket_provider": "openai", + "websocket_model": "gpt-realtime-whisper", + "mic_osd_style": "pill", + }) + previews = self._apply() + + self.assertIsNotNone(self.client.callback) + + def test_gemini_preview_is_shown_in_pill(self): + self.manager.config.values.update({ + "websocket_provider": "google", + "websocket_model": "gemini-2.0-flash-live", + "mic_osd_style": "pill", + }) + previews = self._apply() + + self.assertIsNotNone(self.client.callback) + + def test_gemini_preview_is_not_shown_in_waveform(self): + self.manager.config.values.update({ + "websocket_provider": "google", + "websocket_model": "gemini-2.0-flash-live", + "mic_osd_style": "waveform", + }) + previews = self._apply() + + self.assertIsNone(self.client.callback) + self.assertEqual(previews, [""]) + + def test_pill_preview_requires_client_to_support_callback(self): + self.backend._realtime_client = object() + + previews = self._apply() + + self.assertEqual(previews, [""]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_manager_init_lifecycle.py b/tests/test_manager_init_lifecycle.py index 198efa42..5cd5b38b 100644 --- a/tests/test_manager_init_lifecycle.py +++ b/tests/test_manager_init_lifecycle.py @@ -44,6 +44,9 @@ def cleanup(self): if cleanup_error is not None: raise cleanup_error + def apply_partial_callback(self, callback): + events.append(('partial_callback', callback)) + return FakeBackend @@ -97,6 +100,19 @@ def test_failed_initialize_keeps_backend_for_missing_client_gate(self): self.assertIsNotNone(manager._backend) self.assertTrue(manager.realtime_client_missing()) + def test_realtime_callback_is_reapplied_after_client_initializes(self): + events = [] + manager = self._manager(make_fake_backend_cls( + events, + name='realtime-ws', + )) + callback = lambda text: None + manager.set_realtime_partial_callback(callback) + + self.assertTrue(manager.initialize()) + + self.assertIn(('partial_callback', callback), events) + def test_cleanup_exception_does_not_abort_initialize(self): events = [] manager = self._manager( diff --git a/tests/test_mic_osd_runner.py b/tests/test_mic_osd_runner.py index 8a8f7479..178f6822 100644 --- a/tests/test_mic_osd_runner.py +++ b/tests/test_mic_osd_runner.py @@ -248,6 +248,43 @@ def text_extents(self, text): self.assertEqual(window._text_width(ObjectContext(), "abcd"), 20) self.assertEqual(window._text_height(ObjectContext(), "abcd"), 10) + def test_pill_word_layout_uses_space_advance_instead_of_zero_ink_width(self): + window_module, _ = self._import_window_with_stubs() + window = object.__new__(window_module.OSDWindow) + + class SpaceHasNoInkContext: + def text_extents(self, text): + if text == " ": + return (0, 0, 0, 0, 6, 0) + return (0, 0, len(text) * 5, 10, len(text) * 5, 0) + + positions, total = window._word_layout( + SpaceHasNoInkContext(), + ("one", "two"), + 100, + ) + + self.assertEqual(total, 36) + self.assertEqual(positions, (32, 53)) + + def test_pill_long_token_is_ellipsized_to_available_width(self): + window_module, _ = self._import_window_with_stubs() + window = object.__new__(window_module.OSDWindow) + + class AdvanceContext: + def text_extents(self, text): + advance = len(text) * 5 + return (0, 0, advance, 10, advance, 0) + + text = window._ellipsize_pill_token( + AdvanceContext(), + "averylongunbrokentoken", + 40, + ) + + self.assertEqual(text, "averylo…") + self.assertLessEqual(window._text_advance(AdvanceContext(), text), 40) + def test_preview_text_draws_only_while_recording(self): window_module, _ = self._import_window_with_stubs() window = object.__new__(window_module.OSDWindow) diff --git a/tests/test_pill_transcript_preview.py b/tests/test_pill_transcript_preview.py new file mode 100644 index 00000000..ee9aed8e --- /dev/null +++ b/tests/test_pill_transcript_preview.py @@ -0,0 +1,125 @@ +import unittest + +from mic_osd.transcript_preview import ( + PillTranscriptAnimator, + PillTranscriptConfig, +) + + +class FakeClock: + def __init__(self): + self.now = 0.0 + + def __call__(self): + return self.now + + +class PillTranscriptConfigTests(unittest.TestCase): + def test_loads_and_clamps_public_settings(self): + values = { + "mic_osd_pill_transcript_enabled": True, + "mic_osd_pill_transcript_word_limit": 99, + "mic_osd_pill_transcript_idle_timeout_ms": 800, + } + config = PillTranscriptConfig.from_getter( + lambda key, default=None: values.get(key, default) + ) + self.assertTrue(config.enabled) + self.assertEqual(config.word_limit, 12) + self.assertEqual(config.idle_timeout_seconds, 0.8) + + def test_preview_is_opt_in_by_default(self): + config = PillTranscriptConfig.from_getter( + lambda _key, default=None: default + ) + self.assertFalse(config.enabled) + + +class PillTranscriptAnimatorTests(unittest.TestCase): + def setUp(self): + self.clock = FakeClock() + self.config = PillTranscriptConfig( + enabled=True, + word_limit=4, + idle_timeout_seconds=1.4, + ) + self.animator = PillTranscriptAnimator(self.config, clock=self.clock) + + def test_keeps_only_latest_four_words(self): + self.animator.set_text("zero one two three four", now=0.0) + self.assertEqual( + self.animator.current_words, + ("one", "two", "three", "four"), + ) + + def test_character_growth_does_not_restart_animation(self): + self.animator.set_text("trans", now=0.0) + transition_started_at = self.animator.transition_started_at + for index, text in enumerate( + ("transc", "transcr", "transcri", "transcrip", "transcript"), + start=1, + ): + self.animator.set_text(text, now=index * 0.03) + self.assertEqual( + self.animator.transition_started_at, + transition_started_at, + ) + self.assertEqual(self.animator.current_words, ("transcript",)) + + def test_250_wpm_stream_never_queues_and_converges_on_latest_words(self): + completed = [] + now = 0.0 + for word_index in range(12): + target = f"word{word_index}" + for char_index in range(1, len(target) + 1): + provisional = completed + [target[:char_index]] + self.animator.set_text(" ".join(provisional), now=now) + now += 0.04 + completed.append(target) + now = max(now, (word_index + 1) * 0.24) + self.assertEqual( + self.animator.current_words, + tuple(completed[-4:]), + ) + self.assertFalse(hasattr(self.animator, "pending_transitions")) + + def test_rolling_window_keeps_overlapping_words_stable(self): + self.animator.set_text("one two three four", now=0.0) + self.animator.set_text("two three four five", now=0.24) + self.assertEqual( + self.animator.matches, + ((1, 0), (2, 1), (3, 2)), + ) + frame = self.animator.frame(now=0.34) + stable = [word for word in frame.words if word.matched_from is not None] + incoming = [ + word + for word in frame.words + if word.source == "current" and word.matched_from is None + ] + outgoing = [word for word in frame.words if word.source == "previous"] + self.assertEqual([word.text for word in stable], ["two", "three", "four"]) + self.assertEqual([word.text for word in incoming], ["five"]) + self.assertEqual([word.text for word in outgoing], ["one"]) + + def test_idle_timeout_starts_upward_exit(self): + self.animator.set_text("last four spoken words", now=0.0) + frame = self.animator.frame(now=1.41) + self.assertEqual(frame.current_words, ()) + self.assertTrue(frame.previous_words) + later = self.animator.frame(now=1.51) + outgoing = [word for word in later.words if word.source == "previous"] + self.assertTrue(all(word.y_offset < 0 for word in outgoing)) + self.assertTrue(all(0 < word.alpha < 1 for word in outgoing)) + + def test_disabled_preview_renders_nothing(self): + animator = PillTranscriptAnimator( + PillTranscriptConfig(), + clock=self.clock, + ) + animator.set_text("should stay hidden", now=0.0) + self.assertEqual(animator.frame(now=0.1).words, ()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pill_visualization.py b/tests/test_pill_visualization.py index ab09ec75..7dd00e96 100644 --- a/tests/test_pill_visualization.py +++ b/tests/test_pill_visualization.py @@ -67,9 +67,15 @@ def test_processing_state_generates_animated_wave(self): ) self.assertGreater(float(np.ptp(visualization.bar_heights)), 0.01) - def test_compact_style_hides_transcript_preview(self): + def test_compact_style_exposes_pill_transcript_preview(self): visualization = self.PillVisualization() - self.assertFalse(visualization.show_preview) + self.assertTrue(visualization.show_preview) + self.assertEqual(visualization.preview_mode, "pill") + + def test_tall_surface_keeps_pill_at_bottom(self): + visualization = self.PillVisualization() + _, y, _, pill_height = visualization._pill_geometry(400, 84) + self.assertEqual(y, 84 - pill_height - 4) def test_registry_exposes_the_pill_style(self): module = importlib.import_module("mic_osd.visualizations")