Add real-time CLI spectrum visualizer

src/focus/analysis/realtime.py

@@ -0,0 +1,115 @@
+"""Real-time spectrum analysis for the live terminal visualizer.
+
+Reuses the same FFT primitives as :mod:`focus.analysis.fft` (``scipy.fft``), but
+maintains a rolling window and per-band smoothing state so it can be driven at
+video frame rates from the playback loop.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+from scipy.fft import rfft, rfftfreq
+
+
+class SpectrumAnalyzer:
+    """Turns a stream of audio chunks into normalized log-spaced band levels.
+
+    All work happens on the caller's thread (the asyncio event loop); the
+    PortAudio callback is never involved. ``push`` is cheap (a downmix + buffer
+    roll); ``compute`` performs one ``rfft`` per call.
+    """
+
+    def __init__(
+        self,
+        sample_rate: int = 48000,
+        fft_size: int = 2048,
+        f_min: float = 30.0,
+        f_max: float = 16000.0,
+        floor_db: float = -70.0,
+        ceil_db: float = -12.0,
+        attack: float = 0.6,
+        decay: float = 0.18,
+    ) -> None:
+        self.sample_rate = sample_rate
+        self.fft_size = fft_size
+        self.f_min = f_min
+        self.f_max = min(f_max, sample_rate / 2.0)
+        self.floor_db = floor_db
+        self.ceil_db = ceil_db
+        self.attack = attack
+        self.decay = decay
+
+        self._buf = np.zeros(fft_size, dtype=np.float32)
+        self._window = np.hanning(fft_size).astype(np.float32)
+        self._freqs = rfftfreq(fft_size, 1.0 / sample_rate)
+        # Coherent gain of the window: normalizing by it makes a full-scale
+        # sine read ~0 dBFS, so the floor/ceil dB thresholds are meaningful and
+        # independent of fft_size.
+        self._norm = self._window.sum() / 2.0
+
+        # Smoothing state and cached band edges, both keyed by num_bands.
+        self._levels: np.ndarray | None = None
+        self._band_bins: list[tuple[int, int]] | None = None
+        self._num_bands: int | None = None
+
+    def push(self, chunk: np.ndarray) -> None:
+        """Feed a new audio chunk (mono or stereo float array)."""
+        if chunk is None or len(chunk) == 0:
+            return
+        mono = chunk.mean(axis=1) if chunk.ndim == 2 else chunk
+        mono = np.asarray(mono, dtype=np.float32)
+        n = len(mono)
+        if n >= self.fft_size:
+            self._buf[:] = mono[-self.fft_size :]
+        else:
+            # Shift left in place. np.roll would allocate a new array on every
+            # push; numpy handles the overlapping slice copy correctly.
+            self._buf[:-n] = self._buf[n:]
+            self._buf[-n:] = mono
+
+    def push_silence(self) -> None:
+        """Zero the rolling window so bars fall to the floor (e.g. while paused)."""
+        self._buf[:] = 0.0
+
+    def _ensure_bands(self, num_bands: int) -> None:
+        if self._num_bands == num_bands and self._band_bins is not None:
+            return
+        edges = np.logspace(np.log10(self.f_min), np.log10(self.f_max), num_bands + 1)
+        idx = np.searchsorted(self._freqs, edges)
+        bins: list[tuple[int, int]] = []
+        max_bin = len(self._freqs)
+        for i in range(num_bands):
+            lo = int(idx[i])
+            hi = int(idx[i + 1])
+            # Guarantee at least one bin per band so high bands aren't empty.
+            if hi <= lo:
+                hi = min(lo + 1, max_bin)
+            bins.append((lo, hi))
+        self._band_bins = bins
+        self._num_bands = num_bands
+        self._levels = np.zeros(num_bands, dtype=np.float32)
+
+    def compute(self, num_bands: int) -> np.ndarray:
+        """Return smoothed band levels in ``[0, 1]`` (length ``num_bands``)."""
+        self._ensure_bands(num_bands)
+        assert self._band_bins is not None and self._levels is not None
+
+        mag = np.abs(rfft(self._buf * self._window)) / self._norm
+        db = 20.0 * np.log10(mag + 1e-9)
+
+        target = np.empty(num_bands, dtype=np.float32)
+        span = self.ceil_db - self.floor_db
+        for i, (lo, hi) in enumerate(self._band_bins):
+            band_db = db[lo:hi].max() if hi > lo else self.floor_db
+            target[i] = np.clip((band_db - self.floor_db) / span, 0.0, 1.0)
+
+        # Fast attack, slow decay; never dip below the instantaneous target.
+        prev = self._levels
+        rising = target >= prev
+        smoothed = np.where(
+            rising,
+            self.attack * target + (1.0 - self.attack) * prev,
+            np.maximum(prev - self.decay, target),
+        )
+        self._levels = np.clip(smoothed, 0.0, 1.0).astype(np.float32)
+        return self._levels

src/focus/audio/output.py

@@ -59,6 +59,14 @@ class AudioOutput:
     _blocks_written: int = field(default=0, init=False)
     _blocks_played: int = field(default=0, init=False)
 
+    # Visualizer tap: a mono ring buffer of the audio actually sent to the
+    # speakers, written from the PortAudio callback and read by the spectrum
+    # renderer on the event-loop thread. Lock-free (visual-only; a torn read is
+    # at most a one-frame glitch) so the callback never blocks.
+    _viz_size: int = field(default=8192, init=False)
+    _viz_ring: np.ndarray | None = field(default=None, init=False, repr=False)
+    _viz_write: int = field(default=0, init=False)
+
     def __post_init__(self):
         if not SOUNDDEVICE_AVAILABLE:
             raise ImportError(
@@ -67,6 +75,7 @@ def __post_init__(self):
         self._queue = queue.Queue(maxsize=self.buffersize)
         # Recovery fade-in over ~50ms for smooth transitions
         self._recovery_samples = int(0.05 * self.sample_rate)
+        self._viz_ring = np.zeros(self._viz_size, dtype=np.float32)
 
     def start(self) -> None:
         """Prepare for audio output (stream starts when buffer is filled)."""
@@ -153,6 +162,42 @@ def _audio_callback(self, outdata: np.ndarray, frames: int, time_info, status) -
         if self.volume != 1.0:
             outdata *= self.volume
 
+        # Capture what's actually being played for the spectrum visualizer.
+        self._capture_visualizer(outdata)
+
+    def _capture_visualizer(self, outdata: np.ndarray) -> None:
+        """Write a mono downmix of the played block into the ring buffer."""
+        ring = self._viz_ring
+        if ring is None:
+            return
+        mono = outdata.mean(axis=1) if outdata.ndim == 2 else outdata
+        n = len(mono)
+        size = self._viz_size
+        w = self._viz_write
+        end = w + n
+        if end <= size:
+            ring[w:end] = mono
+        else:
+            first = size - w
+            ring[w:] = mono[:first]
+            ring[: end - size] = mono[first:]
+        self._viz_write = end % size
+
+    def latest_samples(self, n: int) -> np.ndarray:
+        """Return the most recent ``n`` played mono samples, in order (a copy).
+
+        Safe to call from another thread; reads may momentarily straddle a
+        callback write, which only ever causes a harmless one-frame glitch.
+        """
+        ring = self._viz_ring
+        if ring is None:
+            return np.zeros(n, dtype=np.float32)
+        n = min(n, self._viz_size)
+        w = self._viz_write
+        if w >= n:
+            return ring[w - n : w].copy()
+        return np.concatenate((ring[self._viz_size - (n - w) :], ring[:w]))
+
     def write(self, audio: np.ndarray) -> None:
         """Write audio data to the output buffer.
 
@@ -277,6 +322,10 @@ def pause(self) -> None:
         self._in_underrun = False
         self._underrun_fade_pos = 0
         self._recovery_fade_pos = 0
+        # Silence the visualizer tap so the bars fall to the floor while paused
+        # (the callback is torn down here, so it can no longer clear the ring).
+        if self._viz_ring is not None:
+            self._viz_ring[:] = 0.0
 
     def resume(self) -> None:
         """Resume playback. The stream is recreated lazily once the buffer re-fills."""

src/focus/cli.py

@@ -12,6 +12,9 @@
 from focus.profiles import FocusProfile, get_profile, list_profiles
 from focus.ui.transport import KeyboardController, PlaybackState, StatusLine
 
+# Spectrum visualizer is optional; imported lazily in _run_session so the CLI
+# still works if numpy/scipy are missing (guarded there).
+
 # Check for optional dependencies
 try:
     import numpy as np
@@ -150,6 +153,12 @@ def show_profiles():
     default=9,
     help="Duration of each track before rotation (1-9 minutes, default: 9)",
 )
+@click.option(
+    "--spectrum/--no-spectrum",
+    is_flag=True,
+    default=True,
+    help="Live audio spectrum visualizer (interactive terminal only, default: on)",
+)
 def start_session(
     profile: str,
     frequency: float | None,
@@ -163,6 +172,7 @@ def start_session(
     limiter: bool,
     verbose: bool,
     track_duration: int,
+    spectrum: bool,
 ):
     """Start a focus music session.
 
@@ -187,6 +197,7 @@ def start_session(
         limiter=limiter,
         verbose=verbose,
         track_duration=track_duration,
+        spectrum=spectrum,
     )
 
 
@@ -203,6 +214,7 @@ def launch_session(
     limiter: bool = True,
     verbose: bool = False,
     track_duration: int = 9,
+    spectrum: bool = True,
 ):
     """Resolve a profile, apply overrides, and run a session.
 
@@ -271,6 +283,7 @@ def launch_session(
                 limiter=limiter,
                 verbose=verbose,
                 track_duration=track_duration,
+                spectrum=spectrum,
             )
         )
     except KeyboardInterrupt:
@@ -291,6 +304,7 @@ async def _run_session(
     limiter: bool = True,
     verbose: bool = False,
     track_duration: int = 9,
+    spectrum: bool = True,
 ):
     """Run the audio generation session."""
     try:
@@ -355,6 +369,8 @@ def make_client(phase: str):
     state = None
     keyboard = None
     status_line = None
+    display = None
+    spectrum_task = None
     if interactive:
         state = PlaybackState(
             profile_name=profile.name,
@@ -363,23 +379,41 @@ def make_client(phase: str):
             status="connecting",
         )
         if not verbose:
-            # The live status line and -v logging both want the bottom line;
-            # when verbose, the logs already convey state, so skip the line.
-            status_line = StatusLine()
-            status_line.start()
-            status_line.render(state)
+            # The live status line and -v logging both want the bottom region;
+            # when verbose, the logs already convey state, so skip the display.
+            if spectrum:
+                # The spectrum display owns the bottom rows and draws the same
+                # status text as its last row (via format_status_line).
+                try:
+                    from focus.analysis.realtime import SpectrumAnalyzer
+                    from focus.ui.spectrum import SpectrumDisplay
+
+                    analyzer = SpectrumAnalyzer(sample_rate=sample_rate)
+                    display = SpectrumDisplay(state, analyzer)
+                    display.start()
+                    display.render()
+                except ImportError:
+                    display = None
+            if display is None:
+                status_line = StatusLine()
+                status_line.start()
+                status_line.render(state)
 
         # Redraw immediately on each keypress so pause / volume / help toggles
         # are reflected instantly instead of on the next audio chunk.
         def _on_key():
-            if status_line is not None:
+            if display is not None:
+                display.render()
+            elif status_line is not None:
                 status_line.render(state)
 
         keyboard = KeyboardController(state, on_change=_on_key)
         try:
             keyboard.start()
         except Exception:
             keyboard.stop()
+            if display is not None:
+                display.finish()
             if status_line is not None:
                 status_line.finish()
             raise
@@ -432,6 +466,23 @@ def _on_key():
     session_complete = False
 
     try:
+        # Drive the spectrum redraw at a fixed frame rate, decoupled from the
+        # irregular arrival of audio chunks. Created inside the try so the
+        # finally block always cancels it and restores the terminal.
+        if display is not None:
+            # Feed the visualizer from the playback tap (what's actually heard),
+            # now that the output stream exists.
+            display.source = output.latest_samples
+
+            async def _spectrum_loop():
+                try:
+                    while state is None or not state.quit_requested:
+                        display.render()
+                        await asyncio.sleep(1.0 / display.fps)
+                except asyncio.CancelledError:
+                    pass
+
+            spectrum_task = asyncio.create_task(_spectrum_loop())
         # Outer loop: each iteration consumes one generator until it ends or an
         # interactive control (pause / next take) asks us to reconnect.
         while not session_complete:
@@ -644,8 +695,16 @@ def _on_key():
             traceback.print_exc()
     finally:
         # Restore the terminal before any further output
+        if spectrum_task is not None:
+            spectrum_task.cancel()
+            try:
+                await spectrum_task
+            except asyncio.CancelledError:
+                pass
         if keyboard is not None:
             keyboard.stop()
+        if display is not None:
+            display.finish()
         if status_line is not None:
             status_line.finish()
         # Flush any remaining buffered audio to file (for Ctrl+C case with duration set)