diff --git a/src/focus/analysis/realtime.py b/src/focus/analysis/realtime.py new file mode 100644 index 0000000..f689aec --- /dev/null +++ b/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 diff --git a/src/focus/audio/output.py b/src/focus/audio/output.py index af059c4..ea7cd6c 100644 --- a/src/focus/audio/output.py +++ b/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.""" diff --git a/src/focus/cli.py b/src/focus/cli.py index 42b792d..91ba030 100644 --- a/src/focus/cli.py +++ b/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,16 +379,32 @@ 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) @@ -380,6 +412,8 @@ def _on_key(): 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) diff --git a/src/focus/ui/spectrum.py b/src/focus/ui/spectrum.py new file mode 100644 index 0000000..e735a38 --- /dev/null +++ b/src/focus/ui/spectrum.py @@ -0,0 +1,125 @@ +"""Terminal spectrum visualizer: a classic bar analyzer rising from the bottom. + +Bars are drawn with Unicode block glyphs (black & white). Rendering uses the +same in-place multi-row ANSI pattern as :mod:`focus.ui.launcher` (cursor-up + +erase-to-end each frame), and reuses :func:`format_status_line` for the bottom +status row so the status text stays single-sourced. +""" + +from __future__ import annotations + +import shutil +import sys + +import numpy as np + +from focus.analysis.realtime import SpectrumAnalyzer +from focus.ui.transport import PlaybackState, format_status_line + +# Glyph ramp indexed by eighths of a cell: 0 = empty, 8 = full block. +_GLYPHS = np.array(list(" ▁▂▃▄▅▆▇█")) + + +class SpectrumRenderer: + """Pure, testable conversion of band levels into terminal rows. + + Bars rise from the bottom: the last row is densest. Each returned string is a + full row of ``len(values)`` glyphs; rows are ordered top -> bottom. + """ + + @staticmethod + def render_rows(values: np.ndarray, height: int) -> list[str]: + if height < 1: + return [] + # Filled eighths per column, then per (row, column) how much of that cell + # is filled. Vectorized so a wide terminal stays cheap at 25+ fps. + eighths = np.clip(np.asarray(values, dtype=float), 0.0, 1.0) * (height * 8) + # Row 0 is the top; a row's distance from the bottom scales its threshold. + from_bottom = np.arange(height - 1, -1, -1)[:, None] + cell = eighths[None, :] - from_bottom * 8 # (height, num_bands) + idx = np.clip(np.floor(cell), 0, 8).astype(int) + grid = _GLYPHS[idx] + return ["".join(row) for row in grid] + + +class SpectrumDisplay: + """Owns the in-place terminal block: spectrum bars plus a status row.""" + + def __init__( + self, + state: PlaybackState, + analyzer: SpectrumAnalyzer, + stream=None, + height: int = 12, + fps: int = 25, + source=None, + ) -> None: + self.state = state + self.analyzer = analyzer + self.stream = stream or sys.stdout + self.height = height + self.fps = fps + # Callable returning the latest N played mono samples. Pulled every frame + # so the bars track what's actually heard (not the producer, which runs + # seconds ahead behind the playback buffer). May be set after construction + # once the audio output exists. + self.source = source + self._active = False + self._prev_lines = 0 + + def start(self) -> None: + self._active = True + self._prev_lines = 0 + # Hide the cursor for a flicker-free repaint. + self.stream.write("\x1b[?25l") + self.stream.flush() + + def push(self, chunk: np.ndarray) -> None: + self.analyzer.push(chunk) + + def render(self) -> None: + if not self._active: + return + # Refresh the analysis window from the live playback tap each frame. + if self.source is not None: + self.analyzer.push(self.source(self.analyzer.fft_size)) + size = shutil.get_terminal_size(fallback=(80, 24)) + cols = max(1, size.columns) + lines = max(1, size.lines) + + status = format_status_line(self.state) + if len(status) > cols - 1: + status = status[: cols - 1] + + # Very short terminals: fall back to a status-only line. + if lines < 4: + rows: list[str] = [] + else: + h = max(1, min(self.height, lines - 2)) + num_bands = cols + values = self.analyzer.compute(num_bands) + rows = SpectrumRenderer.render_rows(values, h) + + block = "\n".join(rows + [status]) + out = ["\x1b[?7l"] # disable autowrap for the repaint + if self._prev_lines: + out.append(f"\x1b[{self._prev_lines}A") + out.append("\r\x1b[J") # move to column 0, erase from cursor to end + out.append(block) + out.append("\x1b[?7h") # restore autowrap + self.stream.write("".join(out)) + self.stream.flush() + self._prev_lines = len(rows) # cursor ends on the status row + + def finish(self) -> None: + if not self._active: + return + out = ["\x1b[?7l"] + if self._prev_lines: + out.append(f"\x1b[{self._prev_lines}A") + out.append("\r\x1b[J") # erase the whole block + out.append("\x1b[?7h\x1b[?25h") # restore autowrap + show cursor + self.stream.write("".join(out)) + self.stream.flush() + self._active = False + self._prev_lines = 0 diff --git a/src/focus/ui/transport.py b/src/focus/ui/transport.py index ff150a0..f885f82 100644 --- a/src/focus/ui/transport.py +++ b/src/focus/ui/transport.py @@ -166,7 +166,7 @@ def start(self) -> None: def render(self, state: PlaybackState) -> None: if not self._active: return - line = self._format(state) + line = format_status_line(state) # Truncate to one less than the terminal width (cosmetic right-edge # cleanup). This counts code points, not display columns, so glyphs that # render 2 wide (♫, ⏸, ↑↓ on some terminals) can still overflow — hence @@ -191,20 +191,25 @@ def finish(self) -> None: self.stream.flush() self._active = False - @staticmethod - def _format(s: PlaybackState) -> str: - token = _STATUS_TOKENS.get(s.status, s.status) - vol = f"{int(round(s.volume * 100))}%" - info = ( - f"♫ {s.profile_name} · {s.modulation_freq:.0f}Hz @ {s.modulation_depth:.0%}" - f" {token} {_format_time(s.elapsed_seconds)}" - f" vol {vol} buf {s.buffer_seconds:.1f}s" + +def format_status_line(s: PlaybackState) -> str: + """Render the one-line status text shared by the status line and spectrum. + + Public so other displays (e.g. :class:`focus.ui.spectrum.SpectrumDisplay`) + can compose it as a row without reaching into private helpers. + """ + token = _STATUS_TOKENS.get(s.status, s.status) + vol = f"{int(round(s.volume * 100))}%" + info = ( + f"♫ {s.profile_name} · {s.modulation_freq:.0f}Hz @ {s.modulation_depth:.0%}" + f" {token} {_format_time(s.elapsed_seconds)}" + f" vol {vol} buf {s.buffer_seconds:.1f}s" + ) + if s.show_help: + hints = ( + "[space/p] pause [n] next take [↑↓ or +/-] volume " + "[?] hide help [q] quit [Ctrl+C] stop" ) - if s.show_help: - hints = ( - "[space/p] pause [n] next take [↑↓ or +/-] volume " - "[?] hide help [q] quit [Ctrl+C] stop" - ) - else: - hints = "[space] pause [n] next [↑↓] volume [?] help [q] quit" - return f"{info} · {hints}" + else: + hints = "[space] pause [n] next [↑↓] volume [?] help [q] quit" + return f"{info} · {hints}" diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py new file mode 100644 index 0000000..dd31db7 --- /dev/null +++ b/tests/test_spectrum.py @@ -0,0 +1,247 @@ +"""Tests for the real-time spectrum analyzer and terminal visualizer.""" + +import io + +import numpy as np +import pytest +from click.testing import CliRunner + +from focus.analysis.realtime import SpectrumAnalyzer +from focus.audio.output import SOUNDDEVICE_AVAILABLE, AudioOutput +from focus.cli import main +from focus.ui.spectrum import SpectrumDisplay, SpectrumRenderer +from focus.ui.transport import PlaybackState + +requires_sounddevice = pytest.mark.skipif( + not SOUNDDEVICE_AVAILABLE, reason="sounddevice/PortAudio unavailable" +) + + +def _sine(freq: float, sr: int = 48000, n: int = 24000, amp: float = 0.8) -> np.ndarray: + t = np.arange(n) / sr + mono = (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32) + return np.column_stack([mono, mono]) + + +class TestSpectrumAnalyzer: + def test_output_shape_and_range(self): + a = SpectrumAnalyzer() + a.push(_sine(1000)) + v = a.compute(60) + assert len(v) == 60 + assert float(v.min()) >= 0.0 + assert float(v.max()) <= 1.0 + + def test_tone_peaks_in_matching_band(self): + a = SpectrumAnalyzer() + a.push(_sine(1000)) + for _ in range(6): # let the fast attack settle + v = a.compute(60) + peak = int(v.argmax()) + assert v[peak] > 0.7 + # The peak band's FFT-bin range must actually cover 1 kHz. + lo, hi = a._band_bins[peak] + band_freqs = a._freqs[lo:hi] + assert band_freqs.min() <= 1000.0 <= band_freqs.max() + a._freqs[1] + # A far-away low band should be well below the peak. + assert v[0] < v[peak] - 0.4 + + def test_push_shifts_window_in_place_for_small_chunks(self): + # Chunks smaller than fft_size shift the rolling window left in place + # (an overlapping slice copy); the newest samples must land at the end. + a = SpectrumAnalyzer(fft_size=8) + a.push(np.arange(1, 6, dtype=np.float32)) # 5 samples into an 8-window + assert np.array_equal(a._buf, [0, 0, 0, 1, 2, 3, 4, 5]) + a.push(np.arange(6, 9, dtype=np.float32)) # 3 more, oldest fall off + assert np.array_equal(a._buf, [1, 2, 3, 4, 5, 6, 7, 8]) + + def test_push_keeps_tail_of_oversized_chunk(self): + a = SpectrumAnalyzer(fft_size=4) + a.push(np.arange(10, dtype=np.float32)) + assert np.array_equal(a._buf, [6, 7, 8, 9]) + + def test_silence_is_near_zero(self): + a = SpectrumAnalyzer() + a.push_silence() + for _ in range(20): + v = a.compute(40) + assert float(v.max()) < 1e-3 + + def test_slow_decay_after_drop(self): + a = SpectrumAnalyzer(decay=0.1) + a.push(_sine(1000)) + for _ in range(6): + high = a.compute(60).copy() + peak = int(high.argmax()) + a.push_silence() + after_one = a.compute(60) + # One frame of silence must not collapse the bar; it decays gradually. + assert after_one[peak] >= high[peak] - 0.1 - 1e-6 + assert after_one[peak] > 0.0 + + def test_rise_is_fast(self): + a = SpectrumAnalyzer(attack=0.6) + a.push_silence() + a.compute(60) + a.push(_sine(1000)) + one = a.compute(60) + peak = int(one.argmax()) + # A single frame of attack should already be a large fraction of target. + assert one[peak] > 0.4 + + +class TestSpectrumRenderer: + def test_full_column_is_all_blocks(self): + rows = SpectrumRenderer.render_rows(np.array([1.0]), 5) + assert rows == ["█"] * 5 + + def test_empty_column_is_all_spaces(self): + rows = SpectrumRenderer.render_rows(np.array([0.0]), 5) + assert rows == [" "] * 5 + + def test_bars_rise_from_the_bottom(self): + rows = SpectrumRenderer.render_rows(np.array([0.5]), 4) + # Top rows empty, bottom rows full (rises from the bottom). + assert rows[0] == " " + assert rows[-1] == "█" + + def test_row_width_matches_band_count(self): + rows = SpectrumRenderer.render_rows(np.array([0.2, 0.5, 0.9]), 3) + assert all(len(r) == 3 for r in rows) + + +class TestSpectrumDisplay: + def test_render_brackets_and_status_and_cursor_up(self, monkeypatch): + monkeypatch.setenv("COLUMNS", "40") + monkeypatch.setenv("LINES", "24") + state = PlaybackState(profile_name="deep-work", status="playing") + analyzer = SpectrumAnalyzer() + analyzer.push(_sine(1000)) + buf = io.StringIO() + display = SpectrumDisplay(state, analyzer, stream=buf, height=8) + display.start() + display.render() + out = buf.getvalue() + assert "\x1b[?25l" in out # cursor hidden on start + assert "\x1b[?7l" in out and "\x1b[?7h" in out # autowrap toggled + assert "deep-work" in out # status row reuses format_status_line + # Second render moves the cursor up over the previously drawn block. + buf.truncate(0) + buf.seek(0) + display.render() + # 8 spectrum rows joined to the status row => 8 newlines => move up 8. + assert "\x1b[8A" in buf.getvalue() + + def test_finish_restores_cursor_and_autowrap(self, monkeypatch): + monkeypatch.setenv("COLUMNS", "40") + monkeypatch.setenv("LINES", "24") + state = PlaybackState(profile_name="deep-work") + display = SpectrumDisplay(state, SpectrumAnalyzer(), stream=io.StringIO()) + display.start() + display.render() + display.stream.truncate(0) + display.stream.seek(0) + display.finish() + out = display.stream.getvalue() + assert "\x1b[?25h" in out # cursor restored + assert "\x1b[?7h" in out # autowrap restored + + def test_short_terminal_falls_back_to_status_only(self, monkeypatch): + monkeypatch.setenv("COLUMNS", "40") + monkeypatch.setenv("LINES", "3") + state = PlaybackState(profile_name="deep-work", status="playing") + buf = io.StringIO() + display = SpectrumDisplay(state, SpectrumAnalyzer(), stream=buf) + display.start() + display.render() + # Only the status row is drawn (no preceding spectrum rows). + assert display._prev_lines == 0 + assert "deep-work" in buf.getvalue() + + +class TestSourceDrivenRender: + def test_render_pulls_a_fresh_window_from_the_source_each_frame(self, monkeypatch): + monkeypatch.setenv("COLUMNS", "40") + monkeypatch.setenv("LINES", "24") + calls = {"n": 0} + + def source(n): + calls["n"] += 1 + return _sine(1000)[:, 0] # mono window + + display = SpectrumDisplay( + PlaybackState(profile_name="x"), + SpectrumAnalyzer(), + stream=io.StringIO(), + source=source, + ) + display.start() + display.render() + display.render() + assert calls["n"] == 2 # one pull per frame, so bars track live audio + + +@requires_sounddevice +class TestAudioOutputVisualizerTap: + def test_latest_samples_returns_recent_played_audio(self): + o = AudioOutput() + for i in range(6): + o._capture_visualizer(np.full((2048, 2), float(i), dtype=np.float32)) + tail = o.latest_samples(2048) + assert len(tail) == 2048 + assert np.all(tail == 5.0) # most recent block + + def test_ring_wraps_around(self): + o = AudioOutput() + ramp = np.arange(3000, dtype=np.float32) + for _ in range(3): # 9000 samples into an 8192 ring forces a wrap + o._capture_visualizer(np.column_stack([ramp, ramp])) + last = o.latest_samples(500) + assert np.all(last == np.arange(2500, 3000)) + + def test_pause_clears_the_tap(self): + o = AudioOutput() + o._capture_visualizer(np.ones((2048, 2), dtype=np.float32)) + o.pause() + assert np.all(o.latest_samples(2048) == 0.0) # bars fall while paused + + +class TestCliGate: + @staticmethod + def _invoke(monkeypatch, args): + """Run the CLI with the session stubbed out (no audio device, no stream).""" + captured = {} + + async def fake_run_session(profile, use_mock, duration, output_path=None, **kwargs): + captured.update(kwargs) + + monkeypatch.setattr("focus.cli._run_session", fake_run_session) + result = CliRunner().invoke(main, args) + return result, captured + + def test_spectrum_flag_threads_through(self, monkeypatch): + result, captured = self._invoke(monkeypatch, ["start", "--mock", "--spectrum"]) + assert result.exit_code == 0 + assert captured["spectrum"] is True + + def test_no_spectrum_flag_threads_through(self, monkeypatch): + result, captured = self._invoke(monkeypatch, ["start", "--mock", "--no-spectrum"]) + assert result.exit_code == 0 + assert captured["spectrum"] is False + + def test_spectrum_defaults_on(self, monkeypatch): + _, captured = self._invoke(monkeypatch, ["start", "--mock"]) + assert captured["spectrum"] is True + + def test_no_ansi_block_in_non_tty(self, monkeypatch): + # CliRunner is not a tty, so the visualizer must never hide the cursor + # or emit its in-place block into piped output. + result, _ = self._invoke(monkeypatch, ["start", "--mock", "--spectrum"]) + assert "\x1b[?25l" not in result.output + assert "\x1b[J" not in result.output + + def test_duration_below_minimum_fast_fails(self): + # Guards the real fast-fail path; no session (and no audio) is started. + result = CliRunner().invoke(main, ["start", "--mock", "--duration", "59"]) + assert result.exit_code != 0 + assert "at least 60 seconds" in result.output diff --git a/tests/test_ui.py b/tests/test_ui.py index 9632a97..ea2171e 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -18,7 +18,13 @@ from focus.cli import main from focus.profiles import FocusProfile from focus.ui import launcher -from focus.ui.transport import KeyboardController, PlaybackState, StatusLine, _format_time +from focus.ui.transport import ( + KeyboardController, + PlaybackState, + StatusLine, + _format_time, + format_status_line, +) # The pty-backed tests exercise the raw terminal readers; pty is Unix-only. requires_pty = pytest.mark.skipif( @@ -87,14 +93,14 @@ def test_format_time(self): def test_format_includes_profile_and_hints(self): s = PlaybackState(profile_name="deep-work", modulation_freq=18.0, status="playing") - line = StatusLine._format(s) + line = format_status_line(s) assert "deep-work" in line assert "18Hz" in line assert "[q] quit" in line def test_help_toggle_changes_hints(self): s = PlaybackState(show_help=True) - assert "next take" in StatusLine._format(s) + assert "next take" in format_status_line(s) def test_render_truncates_to_terminal_width(self, monkeypatch): # The rendered payload must stay under the terminal width so writing it