412 lines
16 KiB
Python
412 lines
16 KiB
Python
"""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 RenderableType
|
|
from rich.text import Text
|
|
from textual.widget import Widget
|
|
|
|
from ..spectrum import Frame, SpectrumEngine
|
|
from . import frame
|
|
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)
|
|
snapshot = self.engine.snapshot()
|
|
outer_width, outer_height = max(12, self.size.width), max(4, self.size.height)
|
|
width, height = frame.inner_size(outer_width, outer_height)
|
|
mode = getattr(app, "visualizer", "both")
|
|
phase = int(getattr(app, "tick_count", 0) // 2)
|
|
dim = snapshot.status == "idle"
|
|
self._record(snapshot, int(getattr(app, "tick_count", 0)))
|
|
|
|
lines: list[Text]
|
|
if mode == "wave":
|
|
lines = self._wave(snapshot, width, height, phase, dim)
|
|
elif mode == "braille":
|
|
lines = self._braille(snapshot, width, height, phase, dim)
|
|
elif mode == "waterfall":
|
|
lines = self._waterfall(width, height, dim)
|
|
elif mode == "radial":
|
|
lines = self._radial(snapshot, width, height, phase, dim)
|
|
elif mode == "bars" or height < 8:
|
|
lines = self._bars(snapshot, 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(snapshot, width, top_height, phase, dim)
|
|
mirrored = self._bars(
|
|
snapshot, width, bottom_height, phase + 5, True, peaks=False
|
|
)
|
|
mirrored.reverse()
|
|
lines.extend(mirrored)
|
|
return frame.wrap(
|
|
lines, outer_width, outer_height, self.border_title
|
|
)
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# 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
|