trav
e38c751e39
Reverse-engineered from its PowerPC binary and checked against the original running under emulation: textures byte for byte and every triangle, over 3,600 frames. 25 fps like the plugin; its colour maps and wave images are extracted from its own code; tests/kaleidostrobe_golden.npz was recorded from the original.
376 lines
15 KiB
Python
376 lines
15 KiB
Python
"""The transport bar's visualizer, fed by the player's decoded PCM tee.
|
|
|
|
The player's QAudioBufferOutput delivers buffers essentially in lockstep
|
|
with the audible position (verified: frames received ≈ frames played), so
|
|
no extra sync buffering is needed — we just FFT the newest samples. What a
|
|
frame turns into is the chosen style's business (``visualizer_styles``);
|
|
this widget owns intake, timing, on/dim/off and color.
|
|
"""
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
from PyQt6.QtWidgets import QWidget
|
|
from PyQt6.QtCore import Qt, QTimer, QRectF, pyqtSignal
|
|
from PyQt6.QtGui import QPainter, QColor
|
|
from PyQt6.QtMultimedia import QAudioFormat
|
|
|
|
from lintunes import theme
|
|
from lintunes.config import load_config
|
|
from lintunes.gui.icons import transport_icon
|
|
from lintunes.gui import visualizer_styles as styles
|
|
from lintunes.gui.visualizer_styles import (
|
|
BANDS, HISTORY, STYLES, STYLE_BARS, Frame, make_renderer)
|
|
|
|
|
|
FRAME_MS = 16 # ~60 fps
|
|
RADIUS = 8 # rounded corners, matching the transport button boxes
|
|
ICON_SIZE = 40 # the cast glyph shown in place of bars while casting
|
|
HISTORY_S = 1.5 # decoded audio kept, so a frame can pick what's *heard*
|
|
PENDING_S = 0.25 # the laser catches up at most this much at once
|
|
GAP_MS = 50 # a buffer this far from the last one's end is a seek
|
|
|
|
# Visualizer click cycles through these brightness modes in order.
|
|
MODE_ON = "on"
|
|
MODE_DIM = "dim"
|
|
MODE_OFF = "off"
|
|
_MODE_CYCLE = (MODE_ON, MODE_DIM, MODE_OFF)
|
|
|
|
_TOOLTIP = "Click to cycle: on / dim / off — double-click to enlarge"
|
|
|
|
_SAMPLE_DTYPES = {
|
|
QAudioFormat.SampleFormat.UInt8: (np.uint8, 127.5, 127.5),
|
|
QAudioFormat.SampleFormat.Int16: (np.int16, 0.0, 32768.0),
|
|
QAudioFormat.SampleFormat.Int32: (np.int32, 0.0, 2147483648.0),
|
|
QAudioFormat.SampleFormat.Float: (np.float32, 0.0, 1.0),
|
|
}
|
|
|
|
|
|
def _style(prefs) -> str:
|
|
saved = prefs.get("visualizer_style") if prefs is not None else None
|
|
return saved if saved in STYLES else STYLE_BARS
|
|
|
|
|
|
class VisualizerWidget(QWidget):
|
|
"""Animated visualizer; click to cycle brightness: on / dim / off, and
|
|
double-click to open it big (``VisualizerWindow``).
|
|
|
|
On = highlight-colored; dim = light gray just darker than the panel;
|
|
off = blank (the big window, if open, keeps going in color).
|
|
|
|
While casting there is no local PCM to analyze, so the panel doubles as the
|
|
cast indicator: it shows the device glyph and a click stops casting.
|
|
"""
|
|
|
|
cast_stop_requested = pyqtSignal()
|
|
|
|
def __init__(self, player, prefs=None, parent=None):
|
|
super().__init__(parent)
|
|
self._player = player
|
|
self._prefs = prefs
|
|
saved = prefs.get("visualizer_mode") if prefs is not None else None
|
|
self._mode = saved if saved in _MODE_CYCLE else MODE_ON
|
|
self._mode_before_press = self._mode
|
|
self._renderer = make_renderer(_style(prefs), styles.settings(prefs))
|
|
# Decoded audio, newest last, and the stream time its end belongs
|
|
# to. Decoding runs ahead of the speakers — normally by a buffer or
|
|
# two, but by most of a second when the GUI thread stalls and the
|
|
# buffers arrive in a burst — so each frame draws the stretch the
|
|
# player says is playing *now*, not simply the newest.
|
|
self._sample_rate = 44100
|
|
self._hist = np.zeros(int(HISTORY_S * self._sample_rate), np.float32)
|
|
self._hist_end_ms = None
|
|
self._total = 0 # samples ever appended: the history's end index
|
|
self._drawn = None # index the laser has consumed up to
|
|
self._anchor = None # (position ms, monotonic s) of the last update
|
|
self._sync_delay = styles.sync_delay_ms(load_config())
|
|
self._big = None # VisualizerWindow, made on first double-click
|
|
# True while audio is coming out of something other than this machine
|
|
# (a Chromecast), so there is no decoded PCM to draw.
|
|
self._no_pcm = False
|
|
self._device = ""
|
|
|
|
self.setFixedWidth(140)
|
|
self.setMinimumHeight(36)
|
|
self.setToolTip(_TOOLTIP)
|
|
|
|
self._timer = QTimer(self)
|
|
self._timer.setInterval(FRAME_MS)
|
|
self._timer.timeout.connect(self._advance_frame)
|
|
|
|
player.audio_buffer.connect(self._on_buffer)
|
|
player.playing_changed.connect(self._on_playing_changed)
|
|
player.track_changed.connect(self._on_track_changed)
|
|
if hasattr(player, "sink_changed"):
|
|
player.sink_changed.connect(self._on_sink_changed)
|
|
if prefs is not None:
|
|
prefs.changed.connect(self._on_prefs_changed)
|
|
|
|
# The bars' state, under the names the widget always had for it.
|
|
@property
|
|
def _bars(self):
|
|
return getattr(self._renderer, "bars", [0] * BANDS)
|
|
|
|
@_bars.setter
|
|
def _bars(self, bars):
|
|
self._renderer.bars = bars
|
|
|
|
@property
|
|
def _levels(self):
|
|
return self._renderer.levels
|
|
|
|
def _on_prefs_changed(self):
|
|
self._sync_delay = styles.sync_delay_ms(load_config())
|
|
style = _style(self._prefs)
|
|
config = styles.settings(self._prefs)
|
|
if style != self._renderer.name:
|
|
self._renderer = make_renderer(style, config)
|
|
else:
|
|
self._renderer.configure(config)
|
|
self._repaint()
|
|
|
|
def _on_sink_changed(self, sink):
|
|
"""Casting means no local decode, so there is nothing to draw. The
|
|
panel becomes the cast indicator instead: it shows where the audio
|
|
went, and clicking it stops casting."""
|
|
self._no_pcm = not getattr(sink, "provides_pcm", True)
|
|
if self._no_pcm:
|
|
self._device = getattr(sink, "name", "") or "a Chromecast"
|
|
self._clear()
|
|
self.setToolTip(f"Casting to {self._device} — click to stop")
|
|
self.setCursor(Qt.CursorShape.PointingHandCursor)
|
|
else:
|
|
self._device = ""
|
|
self.setToolTip(_TOOLTIP)
|
|
self.setCursor(Qt.CursorShape.ArrowCursor)
|
|
self._sync_timer()
|
|
self.update()
|
|
|
|
# ---- audio intake ----
|
|
|
|
def _on_buffer(self, buf):
|
|
fmt = buf.format()
|
|
spec = _SAMPLE_DTYPES.get(fmt.sampleFormat())
|
|
if spec is None:
|
|
return
|
|
dtype, offset, scale = spec
|
|
pointer = buf.data()
|
|
pointer.setsize(buf.byteCount())
|
|
raw = np.frombuffer(pointer, dtype=dtype)
|
|
channels = max(1, fmt.channelCount())
|
|
if channels > 1:
|
|
usable = (len(raw) // channels) * channels
|
|
raw = raw[:usable].reshape(-1, channels).mean(axis=1)
|
|
mono = (raw.astype(np.float32) - offset) / scale
|
|
self._feed(mono, fmt.sampleRate() or 44100, buf.startTime() / 1000)
|
|
|
|
def _reset_stream(self):
|
|
self._hist[:] = 0.0
|
|
self._hist_end_ms = None
|
|
self._drawn = None
|
|
self._anchor = None
|
|
|
|
def _feed(self, mono, rate, start_ms=None):
|
|
"""Append decoded audio. ``start_ms`` is its place in the stream;
|
|
without one (tests) it is taken to follow on from the last."""
|
|
if rate != self._sample_rate:
|
|
self._sample_rate = rate
|
|
self._hist = np.zeros(int(HISTORY_S * rate), np.float32)
|
|
self._reset_stream()
|
|
elif (start_ms is not None and self._hist_end_ms is not None
|
|
and abs(start_ms - self._hist_end_ms) > GAP_MS):
|
|
self._reset_stream() # a seek or a new track: another place
|
|
n = len(mono)
|
|
if n >= len(self._hist):
|
|
self._hist[:] = mono[-len(self._hist):]
|
|
else:
|
|
self._hist[:-n] = self._hist[n:]
|
|
self._hist[-n:] = mono
|
|
self._total += n
|
|
begin = start_ms if start_ms is not None else (self._hist_end_ms or 0.0)
|
|
self._hist_end_ms = begin + n * 1000.0 / rate
|
|
|
|
def _heard_ms(self):
|
|
"""Where the speakers are in the stream, or None if the player
|
|
can't say. Qt's position moves in buffer-sized steps; between them
|
|
it is carried forward by the clock."""
|
|
position = getattr(self._player, "position_ms", None)
|
|
if position is None or self._hist_end_ms is None:
|
|
return None
|
|
pos, now = position(), time.monotonic()
|
|
if self._anchor is None or pos != self._anchor[0]:
|
|
self._anchor = (pos, now)
|
|
heard = pos
|
|
else:
|
|
heard = pos + min((now - self._anchor[1]) * 1000.0, 100.0)
|
|
return heard - self._sync_delay
|
|
|
|
def _frame(self, dt):
|
|
"""This tick's Frame: the FFT window ending at what's heard, and
|
|
every sample between the last tick's point and this one's."""
|
|
rate = self._sample_rate
|
|
heard = self._heard_ms()
|
|
lead_ms = 0.0 if heard is None else self._hist_end_ms - heard
|
|
lead = int(round(lead_ms * rate / 1000.0))
|
|
lead = max(0, min(lead, len(self._hist) - HISTORY))
|
|
end = len(self._hist) - lead
|
|
samples = self._hist[end - HISTORY:end]
|
|
now = self._total - lead
|
|
if (self._drawn is None or now < self._drawn
|
|
or now - self._drawn > PENDING_S * rate):
|
|
pending = np.zeros(0, np.float32) # a jump: start afresh
|
|
else:
|
|
count = min(now - self._drawn, end)
|
|
pending = self._hist[end - count:end].copy()
|
|
self._drawn = now
|
|
return Frame(samples, rate, pending, dt)
|
|
|
|
# ---- state ----
|
|
|
|
def _big_visible(self):
|
|
return self._big is not None and self._big.isVisible()
|
|
|
|
def _sync_timer(self):
|
|
"""Animate while playing, unless nothing would show it."""
|
|
run = (not self._no_pcm and self._player.is_playing()
|
|
and (self._mode != MODE_OFF or self._big_visible()))
|
|
if run:
|
|
if not self._timer.isActive():
|
|
self._drawn = None # the laser skips what played meanwhile
|
|
self._timer.start()
|
|
else:
|
|
self._timer.stop()
|
|
|
|
def _on_playing_changed(self, playing):
|
|
if self._no_pcm:
|
|
return # nothing to animate; don't let play/pause restart the timer
|
|
self._sync_timer()
|
|
if not playing:
|
|
self._clear()
|
|
|
|
def _on_track_changed(self, track):
|
|
self._reset_stream()
|
|
if track is None:
|
|
self._timer.stop()
|
|
self._clear()
|
|
|
|
def _clear(self):
|
|
self._renderer.clear()
|
|
self._repaint()
|
|
|
|
def _repaint(self):
|
|
self.update()
|
|
if self._big_visible():
|
|
self._big.update()
|
|
|
|
def _set_mode(self, mode):
|
|
self._mode = mode
|
|
if self._prefs is not None:
|
|
self._prefs.set("visualizer_mode", self._mode)
|
|
if self._mode == MODE_OFF and not self._big_visible():
|
|
self._clear() # off = blank, not the last frame held in place
|
|
self._sync_timer()
|
|
self.update()
|
|
|
|
def mousePressEvent(self, event):
|
|
if self._no_pcm:
|
|
# While casting the panel is the cast indicator, not a spectrum:
|
|
# cycling its brightness would be meaningless, so a click stops
|
|
# casting instead.
|
|
self.cast_stop_requested.emit()
|
|
return
|
|
# Cycle on -> dim -> off -> on.
|
|
self._mode_before_press = self._mode
|
|
idx = _MODE_CYCLE.index(self._mode)
|
|
self._set_mode(_MODE_CYCLE[(idx + 1) % len(_MODE_CYCLE)])
|
|
|
|
def mouseDoubleClickEvent(self, event):
|
|
"""Qt sends press, release, *double-click*: the first press already
|
|
cycled the brightness, so put it back before opening big. A brief
|
|
flicker, in exchange for a single click that never waits to see
|
|
whether a second one is coming."""
|
|
if self._no_pcm:
|
|
return # the first click already stopped casting
|
|
self._set_mode(self._mode_before_press)
|
|
self.open_big()
|
|
|
|
def open_big(self):
|
|
from lintunes.gui.visualizer_window import VisualizerWindow
|
|
if self._big is None:
|
|
self._big = VisualizerWindow(self)
|
|
self._big.open(self.screen())
|
|
self._sync_timer()
|
|
|
|
def big_window_closed(self):
|
|
if self._mode == MODE_OFF:
|
|
self._clear()
|
|
self._sync_timer()
|
|
|
|
# ---- frame computation ----
|
|
|
|
def _advance_frame(self):
|
|
if self._renderer.advance(self._frame(FRAME_MS / 1000)):
|
|
self._repaint()
|
|
|
|
# ---- painting ----
|
|
|
|
def _bar_color(self):
|
|
if self._mode == MODE_DIM:
|
|
return self._gray()
|
|
return QColor(self.palette().highlight().color())
|
|
|
|
def _grid_gray(self):
|
|
"""The laser's graticule: the gray, faded so the beam reads over it
|
|
even when the beam is gray too."""
|
|
gray = self._gray().darker(115)
|
|
gray.setAlphaF(0.75)
|
|
return gray
|
|
|
|
def _gray(self):
|
|
"""Dim mode's color."""
|
|
if self._prefs is not None:
|
|
saved = self._prefs.get("color_visualizer_gray")
|
|
if saved is not None:
|
|
return theme.gray(saved)
|
|
# Light gray just darker than the panel: take alternateBase and
|
|
# drop its lightness ~12% so it tracks theme/scale changes.
|
|
base = QColor(self.palette().alternateBase().color())
|
|
h, s, l, a = base.getHslF()
|
|
base.setHslF(h, s, l * 0.88, a)
|
|
return base
|
|
|
|
def panel_color(self):
|
|
return self.palette().alternateBase().color()
|
|
|
|
def paint_renderer(self, painter, rect, big=False):
|
|
"""Paint the current frame into ``rect`` — the panel's, or the big
|
|
window's. The panel draws nothing when off; the big window draws
|
|
regardless (in color), since opening it is asking to see it."""
|
|
if self._no_pcm or (self._mode == MODE_OFF and not big):
|
|
return
|
|
# A style with colours of its own (Kaleidostrobe) can't take the
|
|
# gray as its colour; it is told instead.
|
|
self._renderer.dimmed = self._mode == MODE_DIM
|
|
self._renderer.paint(painter, rect, self._bar_color(), self._grid_gray())
|
|
|
|
def paintEvent(self, event):
|
|
painter = QPainter(self)
|
|
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
|
# A rounded panel debossed as far as the transport buttons stand out:
|
|
# it shows things, it isn't pressed.
|
|
rect = QRectF(self.rect()).adjusted(0.5, 0.5, -0.5, -0.5)
|
|
panel = theme.paint_deboss(painter, rect, self.panel_color(), RADIUS)
|
|
painter.setClipPath(panel) # keep the drawing inside the rounded shape
|
|
if self._no_pcm:
|
|
# Casting: the audio never passes through this machine, so show
|
|
# where it went instead of pretending to analyze silence.
|
|
glyph = transport_icon(
|
|
"cast", self.palette().highlight().color(),
|
|
size=ICON_SIZE).pixmap(ICON_SIZE, ICON_SIZE)
|
|
painter.drawPixmap(
|
|
(self.width() - ICON_SIZE) // 2,
|
|
(self.height() - ICON_SIZE) // 2, glyph)
|
|
painter.end()
|
|
return
|
|
self.paint_renderer(painter, QRectF(self.rect()))
|