Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .jules/palette.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
## 2024-05-24 - Persistent CLI Status with Rich
**Learning:** `rich.console.Status` context managers can wrap blocking calls like `keyboard.wait()` to provide a persistent "Ready" state, and updates from worker threads are thread-safe if the same `Status` object is used.
**Action:** Use `with console.status("Ready"): blocking_call()` for CLI apps that need a persistent status line, and update it via `status.update()` from event handlers.
46 changes: 28 additions & 18 deletions src/chirp/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,11 @@ def __init__(self, *, verbose: bool = False) -> None:
if not console:
console = Console(stderr=True)

self.console = console
self.status_indicator = self.console.status("Ready", spinner="dots")

try:
with console.status("[bold green]Initializing Parakeet model...[/bold green]", spinner="dots"):
with self.console.status("[bold green]Initializing Parakeet model...[/bold green]", spinner="dots"):
self.parakeet = ParakeetManager(
model_name=self.config.parakeet_model,
quantization=self.config.parakeet_quantization,
Expand Down Expand Up @@ -94,7 +97,8 @@ def run(self) -> None:
try:
self._register_hotkey()
self.logger.info("Chirp ready. Toggle recording with %s", self.config.primary_shortcut)
self.keyboard.wait()
with self.status_indicator:
self.keyboard.wait()
except KeyboardInterrupt:
self.logger.info("Interrupted, exiting.")

Expand Down Expand Up @@ -122,6 +126,7 @@ def _start_recording(self) -> None:
self.audio_feedback.play_error(self.config.error_sound_path)
return
self._recording = True
self.status_indicator.update("Recording...", spinner="point")
self.audio_feedback.play_start(self.config.start_sound_path)
self.logger.info("Recording started")

Expand All @@ -143,28 +148,33 @@ def _stop_recording(self) -> None:
self.logger.debug("Stopping audio capture")
waveform = self.audio_capture.stop()
self._recording = False
self.status_indicator.update("Transcribing...", spinner="dots")
self.audio_feedback.play_stop(self.config.stop_sound_path)
self.logger.info("Recording stopped (%s samples)", waveform.size)
self._executor.submit(self._transcribe_and_inject, waveform)

def _transcribe_and_inject(self, waveform) -> None:
start_time = time.perf_counter()
if waveform.size == 0:
self.logger.warning("No audio samples captured")
return
try:
text = self.parakeet.transcribe(waveform, sample_rate=16_000, language=self.config.language)
except Exception as exc:
self.logger.exception("Transcription failed: %s", exc)
self.audio_feedback.play_error(self.config.error_sound_path)
return
duration = time.perf_counter() - start_time
self.logger.debug("Transcription finished in %.2fs (chars=%s)", duration, len(text))
if not text.strip():
self.logger.info("Transcription empty; skipping paste")
return
self.logger.debug("Transcription: %s", text)
self.text_injector.inject(text)
start_time = time.perf_counter()
if waveform.size == 0:
self.logger.warning("No audio samples captured")
return
try:
text = self.parakeet.transcribe(waveform, sample_rate=16_000, language=self.config.language)
except Exception as exc:
self.logger.exception("Transcription failed: %s", exc)
self.audio_feedback.play_error(self.config.error_sound_path)
return
duration = time.perf_counter() - start_time
self.logger.debug("Transcription finished in %.2fs (chars=%s)", duration, len(text))
if not text.strip():
self.logger.info("Transcription empty; skipping paste")
return
self.logger.debug("Transcription: %s", text)
self.text_injector.inject(text)
finally:
if not self._recording:
self.status_indicator.update("Ready", spinner="dots")

def _log_capture_status(self, message: str) -> None:
self.logger.debug("Audio status: %s", message)
Expand Down
97 changes: 97 additions & 0 deletions tests/test_ui_status.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import sys
from unittest.mock import MagicMock, call

# Mock low-level dependencies before imports to avoid environment issues
sys.modules["sounddevice"] = MagicMock()
sys.modules["winsound"] = MagicMock()
sys.modules["keyboard"] = MagicMock()

import unittest
from unittest.mock import patch, ANY

# Import after mocking
from chirp.main import ChirpApp

class TestUIStatus(unittest.TestCase):
@patch("chirp.main.get_logger")
@patch("chirp.main.ConfigManager")
@patch("chirp.main.ParakeetManager")
@patch("chirp.main.AudioCapture")
@patch("chirp.main.AudioFeedback")
@patch("chirp.main.KeyboardShortcutManager")
@patch("chirp.main.TextInjector")
@patch("chirp.main.Console") # Mock the Console class used in main.py
def test_status_lifecycle(self, MockConsole, MockInjector, MockKeyboard, MockFeedback, MockCapture, MockParakeet, MockConfig, MockGetLogger):
"""Verify that the UI status indicator transitions correctly through states."""

# Setup mocks
mock_console_instance = MockConsole.return_value
mock_status = MagicMock()
mock_console_instance.status.return_value = mock_status

# Mock Config
mock_config = MockConfig.return_value.load.return_value
mock_config.max_recording_duration = 0
mock_config.audio_feedback = False

# Ensure logger has no RichHandler
mock_logger = MockGetLogger.return_value
mock_logger.handlers = []

# Initialize App
app = ChirpApp()

# Prevent ThreadPoolExecutor from actually running the task in background during test to avoid race
# We can just verify _executor.submit was called, and manually call _transcribe_and_inject
app._executor = MagicMock()

# 1. Verify initialization
self.assertTrue(hasattr(app, "status_indicator"))
mock_console_instance.status.assert_any_call("Ready", spinner="dots")

# 2. Test Start Recording
app._start_recording()
mock_status.update.assert_called_with("Recording...", spinner="point")

# 3. Test Stop Recording
app.audio_capture.stop.return_value = MagicMock(size=100)
app._stop_recording()
# Should set to Transcribing
mock_status.update.assert_called_with("Transcribing...", spinner="dots")

# Verify task submitted
app._executor.submit.assert_called_with(app._transcribe_and_inject, ANY)

# 4. Test Transcribe Finished
# Manually call the worker method
app._transcribe_and_inject(MagicMock(size=100))
mock_status.update.assert_called_with("Ready", spinner="dots")

@patch("chirp.main.get_logger")
@patch("chirp.main.ConfigManager")
@patch("chirp.main.ParakeetManager")
@patch("chirp.main.AudioCapture")
@patch("chirp.main.AudioFeedback")
@patch("chirp.main.KeyboardShortcutManager")
@patch("chirp.main.TextInjector")
@patch("chirp.main.Console")
def test_run_wraps_keyboard_wait(self, MockConsole, MockInjector, MockKeyboard, MockFeedback, MockCapture, MockParakeet, MockConfig, MockGetLogger):
"""Verify that app.run() wraps keyboard.wait() in the status context."""
mock_console_instance = MockConsole.return_value
mock_status = MagicMock()
mock_console_instance.status.return_value = mock_status

# Mock Config
mock_config = MockConfig.return_value.load.return_value
mock_config.max_recording_duration = 0

app = ChirpApp()

app.run()

mock_status.__enter__.assert_called()
app.keyboard.wait.assert_called()
mock_status.__exit__.assert_called()

if __name__ == "__main__":
unittest.main()