feat(visualizer): add waterfall, radial and braille spectrum modes
This commit is contained in:
1 parent
b9d3d7e809
commit
6d188c00bc
2 files changed
+217
-11
No files matched your search
@@ -36,7 +36,7 @@ CHANNELS_CACHE = CACHE_DIR / "channels.json"
|
||||
PLAYLIST_CACHE = CACHE_DIR / "playlists.json"
|
||||
SONGS_CACHE_DIR = CACHE_DIR / "songs"
|
||||
|
||||
VISUALIZERS = ("both", "bars", "wave")
|
||||
VISUALIZERS = ("both", "bars", "wave", "waterfall", "radial", "braille")
|
||||
|
||||
|
||||
def ensure_dirs() -> None:
|
||||
|
||||
+216
-10
@@ -2,6 +2,8 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from collections import deque
|
||||
from collections.abc import Sequence
|
||||
from typing import TYPE_CHECKING, cast
|
||||
|
||||
@@ -15,6 +17,28 @@ from .common import LEVELS, SPECTRUM_STOPS, gradient
|
||||
if TYPE_CHECKING:
|
||||
from ..app import SomaFMApp
|
||||
|
||||
#: 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
|
||||
@@ -24,12 +48,14 @@ class SpectrumPanel(Widget):
|
||||
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 = {"both": "MIRROR", "bars": "BARS", "wave": "WAVE"}.get(
|
||||
mode, mode.upper()
|
||||
)
|
||||
label = VISUALIZER_LABELS.get(mode, mode.upper())
|
||||
title = f"SPECTRUM · {label} · {status}"
|
||||
if self.border_title != title:
|
||||
self.border_title = title
|
||||
@@ -43,10 +69,17 @@ class SpectrumPanel(Widget):
|
||||
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:
|
||||
@@ -63,10 +96,33 @@ class SpectrumPanel(Widget):
|
||||
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)
|
||||
@@ -154,18 +210,24 @@ class SpectrumPanel(Widget):
|
||||
lines.append(line)
|
||||
return lines
|
||||
|
||||
def _wave(
|
||||
self, frame: Frame, width: int, height: int, phase: int, dim: bool
|
||||
) -> list[Text]:
|
||||
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)
|
||||
# The engine normalises each chunk, but smoothing and averaging make
|
||||
# the trace sit around the middle. A gentle display gain lets the
|
||||
# oscilloscope use the full height without amplifying silence into
|
||||
# noise.
|
||||
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)
|
||||
@@ -200,3 +262,147 @@ class SpectrumPanel(Widget):
|
||||
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
|
||||
Reference in new issue
Block a user