"""The flashy bit: an ANSI-colour spectrum analyser / oscilloscope.""" from __future__ import annotations import math from collections import deque from collections.abc import Sequence from typing import TYPE_CHECKING, cast from rich.console import Group, RenderableType from rich.text import Text from textual.widget import Widget from ..spectrum import Frame, SpectrumEngine from .common import LEVELS, SPECTRUM_STOPS, gradient if TYPE_CHECKING: from ..app import BeefyPlayerApp #: One waterfall column per analyser tick; a deque this long covers ~25 s of #: history at 20 Hz, plenty for even a very wide terminal. WATERFALL_HISTORY = 512 WATERFALL_LEVELS = " ░▒▓█" #: Braille dot bit value, indexed by sub-row and sub-column inside a cell. BRAILLE_BITS = ( (0x01, 0x08), (0x02, 0x10), (0x04, 0x20), (0x40, 0x80), ) VISUALIZER_LABELS = { "both": "MIRROR", "bars": "BARS", "wave": "WAVE", "waterfall": "WATERFALL", "radial": "RADIAL", "braille": "BRAILLE", } class SpectrumPanel(Widget): can_focus = False def __init__(self, engine: SpectrumEngine) -> None: super().__init__() self.engine = engine self.border_title = "SPECTRUM" self._palettes: dict[tuple[int, int, bool], list[str]] = {} #: Recent band frames, newest last, for the waterfall display. Filled #: even while another mode is showing so switching to it has history. self._history: deque[tuple[float, ...]] = deque(maxlen=WATERFALL_HISTORY) self._history_tick: int | None = None # Called from the app ticker so the border title tracks the analyser mode. def sync(self, mode: str, status: str) -> None: label = VISUALIZER_LABELS.get(mode, mode.upper()) title = f"SPECTRUM · {label} · {status}" if self.border_title != title: self.border_title = title self.refresh() def render(self) -> RenderableType: app = cast("BeefyPlayerApp", self.app) frame = self.engine.snapshot() width = max(8, self.size.width) height = max(2, self.size.height) mode = getattr(app, "visualizer", "both") phase = int(getattr(app, "tick_count", 0) // 2) dim = frame.status == "idle" self._record(frame, int(getattr(app, "tick_count", 0))) lines: list[Text] if mode == "wave": lines = self._wave(frame, width, height, phase, dim) elif mode == "braille": lines = self._braille(frame, width, height, phase, dim) elif mode == "waterfall": lines = self._waterfall(width, height, dim) elif mode == "radial": lines = self._radial(frame, width, height, phase, dim) elif mode == "bars" or height < 8: lines = self._bars(frame, width, height, phase, dim) else: # Default mode: a full-height mirrored spectrum. The top half # grows up from the centre line, the bottom half grows down from # it (same values, flipped rows, dimmed like a reflection). top_height = height // 2 bottom_height = height - top_height lines = self._bars(frame, width, top_height, phase, dim) mirrored = self._bars( frame, width, bottom_height, phase + 5, True, peaks=False ) mirrored.reverse() lines.extend(mirrored) return Group(*lines) # ------------------------------------------------------------------ # # history # ------------------------------------------------------------------ # def _record(self, frame: Frame, tick: int) -> None: """Remember one column per analyser tick (not one per render).""" if tick == self._history_tick: return self._history_tick = tick self._history.append(frame.bands) # ------------------------------------------------------------------ # # drawing # ------------------------------------------------------------------ # @staticmethod def _sample(values: Sequence[float], position: float) -> float: """Linearly interpolate ``values`` at a fractional index.""" if not values: return 0.0 last = len(values) - 1 position = max(0.0, min(float(last), position)) low = int(position) high = min(low + 1, last) fraction = position - low return float(values[low]) * (1.0 - fraction) + float(values[high]) * fraction def _row_palette(self, height: int, phase: int, dim: bool) -> list[str]: key = (height, phase % 12, dim) cached = self._palettes.get(key) if cached is not None: return cached palette: list[str] = [] for row in range(height): position = (height - 1 - row) / max(1, height - 1) colour = gradient(SPECTRUM_STOPS, position) style = f"bold {colour}" if (row + phase) % 7 == 0 else colour if dim: style = f"dim {style}" palette.append(style) if len(self._palettes) > 96: self._palettes.clear() self._palettes[key] = palette return palette @staticmethod def _resample(values: Sequence[float], width: int) -> list[float]: count = len(values) if count == 0: return [0.0] * width if width <= 1 or count == 1: return [float(values[0])] * width last = count - 1 output: list[float] = [] for column in range(width): position = column * last / (width - 1) low = int(position) high = min(low + 1, last) fraction = position - low output.append(values[low] * (1.0 - fraction) + values[high] * fraction) return output @staticmethod def _resample_mean(values: Sequence[float], width: int) -> list[float]: count = len(values) if count == 0: return [0.0] * width output: list[float] = [] for column in range(width): low = int(column * count / width) high = max(low + 1, int((column + 1) * count / width)) high = min(high, count) chunk = values[low:high] output.append(sum(chunk) / len(chunk) if chunk else 0.0) return output def _bars( self, frame: Frame, width: int, height: int, phase: int, dim: bool, peaks: bool = True, ) -> list[Text]: values = self._resample(frame.bands, width) peaks = self._resample(frame.peaks, width) palette = self._row_palette(height, phase, dim) scale = height * 8.0 lines: list[Text] = [] for row in range(height): line = Text(no_wrap=True) row_top = (height - row - 1) * 8.0 style = palette[row] for column in range(width): units = min(values[column], 1.0) * scale - row_top peak_units = min(peaks[column], 1.0) * scale - row_top if units >= 8.0: line.append("█", style=style) elif units > 0.0: level = max(1, min(7, int(units))) line.append(LEVELS[level], style=style) elif ( peaks and 0.0 <= peak_units < 8.0 and peak_units > units + 1.0 ): line.append("▁", style="dim" if dim else "bold") else: line.append(" ", style="") lines.append(line) return lines def _wave_values(self, frame: Frame, width: int) -> list[float]: """The waveform resampled to ``width`` points, with display gain. The engine normalises each chunk, but smoothing and averaging make the trace sit around the middle. A gentle display gain lets an oscilloscope use the full height without amplifying silence into noise. """ wave = self._resample_mean(frame.wave, width) peak = max((abs(value) for value in wave), default=0.0) if peak > 0.02: gain = min(2.5, 0.9 / peak) wave = [value * gain for value in wave] return wave def _wave( self, frame: Frame, width: int, height: int, phase: int, dim: bool ) -> list[Text]: wave = self._wave_values(frame, width) palette = self._row_palette(height, phase, dim) centre = (height - 1) / 2.0 amplitude = max(0.75, (height - 1) / 2.0) grid = [[" " for _ in range(width)] for _ in range(height)] previous_row: int | None = None for column, value in enumerate(wave): clamped = max(-1.0, min(1.0, float(value))) position = centre - clamped * amplitude row = max(0, min(height - 1, round(position))) grid[row][column] = "█" delta = position - row if delta > 0.35 and row + 1 < height: grid[row + 1][column] = "▄" elif delta < -0.35 and row - 1 >= 0: grid[row - 1][column] = "▀" if previous_row is not None and previous_row != row: low, high = sorted((previous_row, row)) for between in range(low + 1, high): if grid[between][column] == " ": grid[between][column] = "│" previous_row = row lines: list[Text] = [] for row in range(height): line = Text(no_wrap=True) style = palette[row] for char in grid[row]: if char == " ": line.append(" ", style="") else: line.append(char, style=style) lines.append(line) return lines def _braille( self, frame: Frame, width: int, height: int, phase: int, dim: bool ) -> list[Text]: """Oscilloscope drawn with braille dots: 2x4 sub-cells per character.""" dots_width = width * 2 dots_height = height * 4 values = self._wave_values(frame, dots_width) centre = (dots_height - 1) / 2.0 amplitude = max(1.0, (dots_height - 1) / 2.0) cells = [[0] * width for _ in range(height)] def set_dot(x: int, y: int) -> None: cells[y // 4][x // 2] |= BRAILLE_BITS[y % 4][x % 2] previous: int | None = None for column, value in enumerate(values): clamped = max(-1.0, min(1.0, float(value))) row = max(0, min(dots_height - 1, round(centre - clamped * amplitude))) set_dot(column, row) # Join steep segments so the trace stays continuous. if previous is not None and previous != row: low, high = sorted((previous, row)) for between in range(low + 1, high): set_dot(column, between) previous = row palette = self._row_palette(height, phase, dim) lines: list[Text] = [] for row in range(height): line = Text(no_wrap=True) style = palette[row] for bits in cells[row]: if bits: line.append(chr(0x2800 | bits), style=style) else: line.append(" ", style="") lines.append(line) return lines def _waterfall(self, width: int, height: int, dim: bool) -> list[Text]: """Scrolling spectrogram: history left-to-right, frequency bottom-up.""" columns = list(self._history)[-width:] if not columns: columns = [()] last = max(1, len(columns[-1]) - 1) # Row 0 is the treble end; the bass sits on the bottom row. positions = [ (height - 1 - row) / max(1, height - 1) * last for row in range(height) ] offset = width - len(columns) fade_after = max(3, int(len(columns) * 0.75)) grid = [[" "] * width for _ in range(height)] styles = [[""] * width for _ in range(height)] for index, column in enumerate(columns): if not column: continue x = offset + index age = len(columns) - index faded = dim or age > fade_after for row, position in enumerate(positions): value = self._sample(column, position) level = int(value * (len(WATERFALL_LEVELS) - 1)) if level <= 0: continue style = gradient(SPECTRUM_STOPS, value) if faded: style = f"dim {style}" elif age == 1: # The newest column is the leading edge of the scan. style = f"bold {style}" grid[row][x] = WATERFALL_LEVELS[level] styles[row][x] = style lines: list[Text] = [] for row in range(height): line = Text(no_wrap=True) for x in range(width): char = grid[row][x] line.append(char, style=styles[row][x] if char != " " else "") lines.append(line) return lines def _radial( self, frame: Frame, width: int, height: int, phase: int, dim: bool ) -> list[Text]: """The spectrum wrapped into a glowing circle around the centre.""" centre_x = (width - 1) / 2.0 centre_y = (height - 1) / 2.0 # Terminal cells are about twice as tall as they are wide, so the # horizontal radius is twice the vertical one to draw a round circle. radius_y = min(centre_y, centre_x / 2.0) if radius_y < 2.5: return self._bars(frame, width, height, phase, dim) radius_x = radius_y * 2.0 bands = list(frame.bands) count = max(1, len(bands)) hole = 0.18 grid = [[" "] * width for _ in range(height)] styles = [[""] * width for _ in range(height)] for row in range(height): dy = (row - centre_y) / radius_y if abs(dy) > 1.0: continue for column in range(width): dx = (column - centre_x) / radius_x distance = math.hypot(dx, dy) if distance > 1.0: continue angle = (math.atan2(dy, dx) + math.pi / 2.0) % (2.0 * math.pi) band = bands[int(angle / (2.0 * math.pi) * count) % count] edge = hole + max(0.0, min(1.0, float(band))) * (1.0 - hole) if distance <= hole + 0.02: # A faint inner ring gives silent stations a focus. if distance > hole * 0.35: grid[row][column] = "·" styles[row][column] = "dim" continue if distance > edge: continue depth = edge - distance if depth < 0.06: grid[row][column] = "█" styles[row][column] = ("dim " if dim else "") + gradient( SPECTRUM_STOPS, edge ) elif depth < 0.15: grid[row][column] = "▓" styles[row][column] = "dim " + gradient(SPECTRUM_STOPS, edge) elif depth < 0.25: grid[row][column] = "▒" styles[row][column] = "dim" lines: list[Text] = [] for row in range(height): line = Text(no_wrap=True) for column in range(width): char = grid[row][column] line.append(char, style=styles[row][column] if char != " " else "") lines.append(line) return lines