diff --git a/somafm_tui/app.py b/somafm_tui/app.py index 8b1bbc3..6c83014 100644 --- a/somafm_tui/app.py +++ b/somafm_tui/app.py @@ -324,6 +324,9 @@ class SomaFMApp(App[None]): self._set_notice("audio disabled (--no-audio)", 5) return + # A tune gets a clean analyser: forget any previous capture failure and + # clear the old station's bars before the new stream starts. + self.spectrum.reset() self._resolve_and_play(station.id) self._fetch_history(station.id) diff --git a/somafm_tui/spectrum.py b/somafm_tui/spectrum.py index 00df02f..a7fdc95 100644 --- a/somafm_tui/spectrum.py +++ b/somafm_tui/spectrum.py @@ -47,9 +47,19 @@ MAX_FREQ = 16000.0 SOURCE_MODES = ("auto", "monitor", "stream", "sim") #: ~46 ms of all-zero audio per block; give an output capture ~6 s to prove -#: itself before falling back to the stream tap. +#: itself before falling back to the stream tap. This only applies before the +#: capture has ever delivered real audio. Once a capture has worked, silence +#: means the output really is silent (or mpv is rebuffering), so SYNC is kept +#: and the bars simply fall instead of switching to the drifting stream tap. SILENT_MONITOR_BLOCKS = 130 +#: How long a monitor may deliver no data at all before the stream tap takes +#: over. Kept longer than a normal station change. +MONITOR_IDLE_SECONDS = 10.0 + +#: How often a quiet monitor pipe is polled so stop/URL changes stay responsive. +MONITOR_POLL_SECONDS = 0.5 + #: Frames per analysis block for the macOS tap (float32, ~43 ms at 48 kHz). TAP_BLOCK_FRAMES = 2048 @@ -58,7 +68,7 @@ TAP_BLOCK_FRAMES = 2048 #: the window playback gets to start before the stream tap takes over. TAP_IDLE_SECONDS = 10.0 -#: How long the stream tap covers for us before the tap is tried again. +#: How long the stream tap covers for us before an output capture is retried. TAP_RETRY_SECONDS = 60.0 @@ -118,8 +128,18 @@ class SpectrumEngine: self._url: str | None = None self._proc: subprocess.Popen[bytes] | None = None self._capture_kind: str | None = None + #: Incremented whenever the source changes; captures from an older + #: generation must not affect the new one (or latch a failure). + self._generation = 0 + self._reset_pending = False + #: True once any output capture has ever delivered real audio. A later + #: silent capture is then a gap, not a misconfigured device. + self._monitor_worked = False self._monitor_failed = False + #: Next time an output capture may be tried (shared by parec and tap). self._tap_retry_at = 0.0 + #: Why ``_capture`` last returned ("eof", "idle", "silent", ...). + self._capture_reason = "eof" self._monitor_device_cache: str | None = None self._notice: str | None = None self._lock = threading.Lock() @@ -151,20 +171,37 @@ class SpectrumEngine: def set_source(self, url: str | None) -> None: """Start analysing playback, or idle when ``url`` is ``None``. - The URL is only a fallback source for the stream tap. A monitor - capture is tied to the actual audio output, so it deliberately keeps - running when the station changes underneath it. + A station change is a clean slate: any capture belonging to the old + station is retired, a previous monitor failure is forgotten, and the + analyser's display state is reset. That way every tune gets a fresh + chance to lock onto the real output (SYNC) instead of inheriting the + drifting stream tap (FFT) or the synthesized fallback (SIM). """ url = url or None with self._lock: changed = url != self._url self._url = url - kind = self._capture_kind + if changed: + self._request_reset() if url is None: self._status = "idle" - if changed and (url is None or kind != "monitor"): + if changed: + # Unblock a stream read and retire a subprocess monitor. A tap has + # no process; its generation check makes it stop and restart. self._kill_proc() + def reset(self) -> None: + """Give the output capture a clean slate (called when tuning). + + Forgets a previous monitor failure - including a hard one that set + ``_monitor_failed`` - and clears the displayed state, so the next + capture may use the real output (SYNC) again. Unlike :meth:`set_source` + this does not change the source, so re-tuning the same station works. + """ + with self._lock: + self._request_reset() + self._kill_proc() + def snapshot(self) -> Frame: with self._lock: return Frame( @@ -213,6 +250,41 @@ class SpectrumEngine: with self._lock: return self._url + def _request_reset(self) -> None: + """Arm a clean restart; the caller must hold ``self._lock``.""" + self._generation += 1 + self._reset_pending = True + self._monitor_failed = False + self._tap_retry_at = 0.0 + self._capture_reason = "eof" + # The default sink may have changed along with the stream; resolve it + # again for the next monitor capture. + self._monitor_device_cache = None + + def _read_generation(self) -> int: + with self._lock: + return self._generation + + def _take_reset(self) -> bool: + with self._lock: + pending, self._reset_pending = self._reset_pending, False + return pending + + def _reset_analysis(self) -> None: + """Drop the previous station's samples and display state. + + Only called from the analyser thread (or before it starts), so the + history buffer needs no lock; the values read by ``snapshot`` do. + """ + self._history[:] = 0.0 + self._sim_phase = 0.0 + with self._lock: + self._levels[:] = 0.0 + self._peaks[:] = 0.0 + self._wave[:] = 0.0 + self._level = 0.0 + self._gain = -24.0 + def _kill_proc(self) -> None: with self._lock: proc = self._proc @@ -227,13 +299,11 @@ class SpectrumEngine: def _use_monitor(self) -> bool: if self._monitor_failed or self.source_mode in ("sim", "stream"): return False - kind = self._monitor_kind() - if kind is None: + if self._monitor_kind() is None: return False - if kind != "tap": - return True - # The tap needs the default output to actually render something; when - # it does not, the stream tap covers us until the retry deadline. + # A capture that had nothing to give us (a station change leaves the + # sink quiet while mpv buffers, a tap is idle before playback starts) + # is retried later instead of being written off for the session. return time.monotonic() >= self._tap_retry_at def _monitor_kind(self) -> str | None: @@ -316,6 +386,8 @@ class SpectrumEngine: def _run(self) -> None: while not self._stop.is_set(): + if self._take_reset(): + self._reset_analysis() url = self._read_url() if not url: self._idle_frame() @@ -327,15 +399,21 @@ class SpectrumEngine: continue if self._use_monitor(): + generation = self._read_generation() outcome = self._capture_monitor(url) + if generation != self._read_generation(): + # The station changed while this capture was running; its + # result belongs to the old source. Start over cleanly. + continue if outcome == "done": continue if outcome == "retry": - # Nothing was rendered (yet): try the tap again later. + # Nothing was rendered (yet), or the sink went quiet while + # the station changed: try the output capture again later. self._tap_retry_at = time.monotonic() + TAP_RETRY_SECONDS else: - # Device gone, or the macOS audio-capture permission was - # never granted: do not retry in this session. + # Hard failure (the capture never produced audio): do not + # retry until the source changes (which resets this). self._monitor_failed = True if self._stop.is_set(): break @@ -343,20 +421,36 @@ class SpectrumEngine: continue if not self._ffmpeg: - self._simulate_for(url, 0.05) + # Keep re-checking: a pending output-capture retry may become + # due while we are showing the synthesized fallback. + self._simulate_for(url, 0.05, duration=5.0) continue + generation = self._read_generation() proc = self._spawn(self._command(url), "stream") if proc is None: with self._lock: self._status = "sim" self._simulate_for(url, 0.06, duration=5.0) continue - self._capture(proc, url, stop_on_url_change=True, status="live") + self._capture( + proc, + url, + stop_on_url_change=True, + status="live", + generation=generation, + ) self._stop_proc(proc) if self._stop.is_set(): break + if generation != self._read_generation(): + continue + # The stream tap only covers periods where no output capture is + # available. Once the retry deadline has passed, go back around + # and try the real thing instead of simulating on top of the tap. + if self._use_monitor(): + continue if self._read_url() == url: with self._lock: self._status = "sim" @@ -407,25 +501,44 @@ class SpectrumEngine: """Capture the real audio output. Returns ``"done"`` when playback stopped, ``"failed"`` when the - monitor is unusable, and ``"retry"`` when it merely had nothing to - capture yet. + capture never produced audio (the device is gone or the sink is not + routed into it), and ``"retry"`` when it is temporarily unavailable. """ + generation = self._read_generation() if self._monitor_kind() == "tap": - return self._capture_tap() + outcome = self._capture_tap() + if generation != self._read_generation(): + return "done" + return outcome command = self._monitor_command() proc = self._spawn(command, "monitor") if command else None if proc is None: return "failed" try: - self._capture(proc, url, stop_on_url_change=False, status="monitor") + self._capture( + proc, + url, + stop_on_url_change=False, + status="monitor", + generation=generation, + ) finally: self._stop_proc(proc) + if generation != self._read_generation(): + # Deliberately stopped/restarted between stations; not a failure. + return "done" if self._stop.is_set() or self._read_url() is None: return "done" + # A sink that is quiet for a moment (mpv rebuffers a new station, a + # device is briefly suspended) is not broken: let the stream tap cover + # us until the next retry instead of permanently writing SYNC off. + if self._capture_reason == "idle": + return "retry" return "failed" def _capture_tap(self) -> str: """Capture the native macOS output tap.""" + generation = self._read_generation() tap = macos_tap.SystemAudioTap() try: tap.start() @@ -438,21 +551,28 @@ class SpectrumEngine: self._status = "monitor" try: self._retune(tap.sample_rate) - return self._pump_tap(tap) + return self._pump_tap(tap, generation) finally: tap.stop() with self._lock: if self._capture_kind == "monitor": self._capture_kind = None - def _pump_tap(self, tap: macos_tap.SystemAudioTap) -> str: + def _pump_tap( + self, tap: macos_tap.SystemAudioTap, generation: int | None = None + ) -> str: """Feed tap audio to the analyser until playback stops or it fails.""" + if generation is None: + generation = self._read_generation() channels = max(1, tap.channels) block_bytes = TAP_BLOCK_FRAMES * max(4, tap.bytes_per_frame) buffer = bytearray() silent_blocks = 0 + ever_worked = self._monitor_worked last_audio = time.monotonic() while not self._stop.is_set(): + if self._read_generation() != generation: + return "done" if self._read_url() is None: return "done" chunk = tap.read(0.5) @@ -473,19 +593,24 @@ class SpectrumEngine: while len(buffer) >= block_bytes: block = bytes(buffer[:block_bytes]) del buffer[:block_bytes] - # A tap that was never granted permission is created happily - # but only ever hands out digital silence; detect that here. if not block.strip(b"\x00"): silent_blocks += 1 - if silent_blocks >= SILENT_MONITOR_BLOCKS: + # A tap that was never granted permission is created + # happily but only ever hands out digital silence; detect + # that here. Silence after real audio, however, is just a + # station change or an actually silent output, so keep + # monitoring (the bars fall by themselves). + if not ever_worked and silent_blocks >= SILENT_MONITOR_BLOCKS: self._notify( "macOS is not letting the app record audio — allow " "'System Audio Recording' for your terminal in " "System Settings, then restart (using FFT)" ) return "failed" - continue - silent_blocks = 0 + else: + silent_blocks = 0 + ever_worked = True + self._monitor_worked = True self.feed_audio(block, channels, "monitor") return "done" @@ -496,56 +621,103 @@ class SpectrumEngine: *, stop_on_url_change: bool, status: str, + generation: int | None = None, ) -> bool: - """Read PCM until playback stops, the source changes, or the process dies.""" + """Read PCM until playback stops, the source changes, or the process dies. + + ``self._capture_reason`` records why the loop ended so the caller can + tell a capture that has gone for good (``"eof"``/``"dead"``) from one + that is merely quiet while a station is being tuned (``"idle"`` / + ``"restart"``). + """ stdout = proc.stdout if stdout is None: + self._capture_reason = "eof" return False + if generation is None: + generation = self._read_generation() fd = stdout.fileno() got_data = False buffer = bytearray() silent_blocks = 0 + ever_worked = self._monitor_worked + last_data = time.monotonic() + reason = "stop" while not self._stop.is_set(): + if self._read_generation() != generation: + reason = "restart" + break current = self._read_url() if current is None: + reason = "idle" break if stop_on_url_change and current != url: + reason = "url" + break + # The stream tap yields once an output capture is due for a retry, + # so a monitor with a momentary lapse does not keep us on the + # (drifting) stream tap for the rest of the session. + if status == "live" and self._use_monitor(): + reason = "monitor" break - # A monitor can go quiet if the device disappears; don't block the - # analyser thread forever. os.read returns whatever is available - # (instead of waiting for a full 4 KiB), which keeps latency low. + # A monitor can go quiet while mpv rebuffers a newly tuned station, + # or when the device disappears. Don't block the analyser thread + # forever, but give silence a fair chance to turn back into audio. + # os.read returns whatever is available (instead of waiting for a + # full 4 KiB), which keeps latency low. if status == "monitor" and os.name == "posix": try: - ready, _, _ = select.select([fd], [], [], 2.0) + ready, _, _ = select.select( + [fd], [], [], MONITOR_POLL_SECONDS + ) except (OSError, ValueError): ready = [fd] if not ready: - break + if proc.poll() is not None: + reason = "eof" + break + if time.monotonic() - last_data >= MONITOR_IDLE_SECONDS: + reason = "idle" + break + continue try: data = os.read(fd, READ_BYTES) except OSError: + reason = "eof" break if not data: + reason = "eof" break + now = time.monotonic() + last_data = now buffer.extend(data) # Process fixed-size blocks so attack/decay smoothing is tied to # time (~46 ms per block), regardless of how the pipe fragments. while len(buffer) >= READ_BYTES: block = bytes(buffer[:READ_BYTES]) del buffer[:READ_BYTES] - # A monitor that was never granted permission (macOS) or that - # nobody routes into (PulseAudio) delivers endless digital - # silence. SYNC bars that never move are worse than the - # (slightly drifting) stream tap, so give up and let the - # caller fall back. if status == "monitor": - silent_blocks = silent_blocks + 1 if block.count(0) == len(block) else 0 - if silent_blocks >= SILENT_MONITOR_BLOCKS: + if block.count(0) == len(block): + silent_blocks += 1 + else: + silent_blocks = 0 + ever_worked = True + self._monitor_worked = True + # A capture that was never granted permission (macOS) or + # that nobody routes into (PulseAudio) delivers endless + # digital silence. SYNC bars that never move are worse + # than the (slightly drifting) stream tap, so give up on a + # backend that has never worked. Once it has worked, + # silence is real (or mpv rebuffering): keep SYNC and let + # the bars fall. + if not ever_worked and silent_blocks >= SILENT_MONITOR_BLOCKS: + self._capture_reason = "dead" return False got_data = True self.feed_pcm(block, status) + self._capture_reason = reason return got_data # ------------------------------------------------------------------ # @@ -668,6 +840,10 @@ class SpectrumEngine: while not self._stop.is_set(): if self._read_url() != url: return + # Stop pretending as soon as a real output capture is available + # again (e.g. the retry deadline expired during the fallback). + if self._use_monitor(): + return self._simulate_step() if self._stop.wait(delay): return