v0.52.0: volume normalization — every song plays at one loudness
Measured once per file with ffmpeg's ebur128 (cached on this machine), applied as a gain on the output volume; the next songs are measured ahead and an unmeasured one glides to its level. Toggle in Controls and Preferences → General.
This commit is contained in:
@@ -147,6 +147,17 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal
|
||||
each load, and on `refresh_volume()` (MainWindow calls that when the
|
||||
playing track is edited). `Player.volume()` stays the bare slider value.
|
||||
|
||||
- **Volume normalization** (Round 83, `loudness.py`, Controls → Normalize
|
||||
Volume / Preferences → General): each file's EBU R128 integrated loudness
|
||||
from the `ffmpeg` CLI's `ebur128` (~0.4 s a song), cached by path + size +
|
||||
mtime in `$XDG_CACHE_HOME/lintunes/loudness.json` (machine-local, never
|
||||
the synced dir). `Player` turns `TARGET_LUFS − measured` into a dB gain in
|
||||
`_output_volume` (`volume_with_gain`, linear amplitude through Qt's own
|
||||
volume curve, capped at full). The playing song is measured urgently and
|
||||
the next `LOOKAHEAD` ahead; an unmeasured one starts at the cache's median
|
||||
and **glides** (never jumps) to its own gain. No DSP, no tags: Qt's
|
||||
pipeline has no insert point, and none is needed.
|
||||
|
||||
- **`lintunes/player.py`** — `Player(QObject)` walks a queue through a swappable
|
||||
**`PlaybackSink`**. Player owns the queue, shuffle walk, per-track start/stop
|
||||
times and the play-count/scrobble bookkeeping; a sink owns only "make this file
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""LinTunes — iTunes-style music library manager and player for Linux."""
|
||||
|
||||
__version__ = "0.51.2"
|
||||
__version__ = "0.52.0"
|
||||
|
||||
@@ -10,7 +10,7 @@ from PyQt6.QtWidgets import (
|
||||
from PyQt6.QtCore import Qt, QEvent, QObject, QTimer, pyqtSignal
|
||||
from PyQt6.QtGui import QAction, QKeySequence
|
||||
|
||||
from lintunes import bpm_detect, device_sync, music_folder, theme, url_import
|
||||
from lintunes import bpm_detect, device_sync, loudness, music_folder, theme, url_import
|
||||
from lintunes.andtunes import layout as andtunes_layout
|
||||
from lintunes.andtunes import install as andtunes_install
|
||||
from lintunes.andtunes import sync as andtunes_sync
|
||||
@@ -114,6 +114,9 @@ class MainWindow(QMainWindow):
|
||||
self._lastfm = lastfm
|
||||
self._prefs_dialog = None
|
||||
self.player = Player(manager, self)
|
||||
# Volume normalization measures each song with ffmpeg as it comes up;
|
||||
# nothing is measured while it's off (the default, and in tests).
|
||||
self.player.loudness = loudness.LoudnessAnalyzer(parent=self)
|
||||
# Owns the cast session, if any. Built before the transport bar, which
|
||||
# takes it to drive the little cast indicator under the volume slider.
|
||||
self._cast = CastController(self.player, self)
|
||||
@@ -367,7 +370,21 @@ class MainWindow(QMainWindow):
|
||||
|
||||
# ---- preferences / theming ----
|
||||
|
||||
def _apply_normalize_pref(self):
|
||||
on = bool(self._prefs.get("normalize_volume", False))
|
||||
if on and not loudness.ffmpeg_available():
|
||||
on = False
|
||||
self.statusBar().showMessage(
|
||||
"Volume normalization needs ffmpeg, which isn't installed.",
|
||||
8000)
|
||||
self.player.set_normalize(on)
|
||||
if self._normalize_action.isChecked() != on:
|
||||
self._normalize_action.blockSignals(True)
|
||||
self._normalize_action.setChecked(on)
|
||||
self._normalize_action.blockSignals(False)
|
||||
|
||||
def _on_prefs_changed(self):
|
||||
self._apply_normalize_pref()
|
||||
theme.apply_theme(QApplication.instance(), self._prefs)
|
||||
self.apply_ui_metrics()
|
||||
self._transport.refresh_theme()
|
||||
@@ -441,6 +458,13 @@ class MainWindow(QMainWindow):
|
||||
self._add_action(controls, "Play/Pause", "", self.play_pause)
|
||||
self._add_action(controls, "Next", "", self.player.next)
|
||||
self._add_action(controls, "Previous", "", self.player.previous)
|
||||
controls.addSeparator()
|
||||
self._normalize_action = QAction("Normalize Volume", self)
|
||||
self._normalize_action.setCheckable(True)
|
||||
self._normalize_action.toggled.connect(
|
||||
lambda on: self._prefs.set("normalize_volume", on))
|
||||
controls.addAction(self._normalize_action)
|
||||
self._apply_normalize_pref()
|
||||
|
||||
view_menu = bar.addMenu("&View")
|
||||
self._add_action(view_menu, "Toggle Column Browser", "Ctrl+B",
|
||||
@@ -2361,5 +2385,6 @@ class MainWindow(QMainWindow):
|
||||
self._cassette.shutdown()
|
||||
self._cast.shutdown()
|
||||
self.player.shutdown()
|
||||
self.player.loudness.save()
|
||||
self._manager.flush()
|
||||
super().closeEvent(event)
|
||||
|
||||
@@ -11,7 +11,7 @@ from PyQt6.QtWidgets import (
|
||||
from PyQt6.QtCore import Qt, QTimer
|
||||
from PyQt6.QtGui import QFont, QColor, QFontInfo
|
||||
|
||||
from lintunes import theme
|
||||
from lintunes import loudness, theme
|
||||
from lintunes.config import load_config, save_config
|
||||
from lintunes.export import upload
|
||||
from lintunes.gui import visualizer_styles
|
||||
@@ -112,7 +112,8 @@ class PreferencesDialog(QDialog):
|
||||
self._color_save_timer.timeout.connect(self._prefs.save)
|
||||
|
||||
self._headings = []
|
||||
general = [self._build_sound_group(), self._build_wrapped_group()]
|
||||
general = [self._build_playback_group(), self._build_sound_group(),
|
||||
self._build_wrapped_group()]
|
||||
if self._music_folder is not None:
|
||||
general.insert(0, self._build_music_group())
|
||||
self.pages_list = QListWidget()
|
||||
@@ -201,6 +202,38 @@ class PreferencesDialog(QDialog):
|
||||
self.pages.setCurrentIndex(row)
|
||||
PreferencesDialog._last_page = row
|
||||
|
||||
# ---- Playback ----
|
||||
|
||||
def _build_playback_group(self) -> QGroupBox:
|
||||
group = QGroupBox("Playback")
|
||||
column = QVBoxLayout(group)
|
||||
self._normalize_check = QCheckBox("Normalize volume")
|
||||
self._normalize_check.setChecked(
|
||||
bool(self._prefs.get("normalize_volume", False)))
|
||||
self._normalize_check.toggled.connect(
|
||||
lambda on: self._prefs.set("normalize_volume", on))
|
||||
column.addWidget(self._normalize_check)
|
||||
if loudness.ffmpeg_available():
|
||||
note = ("Plays every song at the same loudness, so a quiet old "
|
||||
"recording and a loud new one sit at one level. Each song "
|
||||
"is measured once as it comes up; the files aren't "
|
||||
"changed. Quiet songs can only be raised while the volume "
|
||||
"slider is below the top.")
|
||||
else:
|
||||
note = "Needs ffmpeg, which isn't installed."
|
||||
self._normalize_check.setEnabled(False)
|
||||
column.addWidget(self._note(note))
|
||||
# Controls → Normalize Volume flips the same pref while this is open.
|
||||
self._prefs.changed.connect(self._sync_normalize_check)
|
||||
return group
|
||||
|
||||
def _sync_normalize_check(self):
|
||||
on = bool(self._prefs.get("normalize_volume", False))
|
||||
if self._normalize_check.isChecked() != on:
|
||||
self._normalize_check.blockSignals(True)
|
||||
self._normalize_check.setChecked(on)
|
||||
self._normalize_check.blockSignals(False)
|
||||
|
||||
# ---- Sound ----
|
||||
|
||||
def _build_sound_group(self) -> QGroupBox:
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
"""Volume normalization: how loud each song is, so they all play at one level.
|
||||
|
||||
Each file is measured once with ffmpeg's ``ebur128`` filter (EBU R128
|
||||
integrated loudness, in LUFS — what Spotify and YouTube normalize by), and
|
||||
``Player`` turns the difference from ``TARGET_LUFS`` into a gain on its
|
||||
output volume. Nothing touches the audio itself: no DSP in the playback
|
||||
path, no tag written, the file is only read. A whole 4-minute song measures
|
||||
in about 0.4 s, so songs are measured as they're about to play rather than
|
||||
by a library scan.
|
||||
|
||||
The measurements are this machine's business (they're keyed by path), so they
|
||||
live in ``$XDG_CACHE_HOME/lintunes/loudness.json``, never the synced data
|
||||
dir, and a file whose size or mtime changed is measured again.
|
||||
|
||||
Shaped like ``bpm_detect.py``: the parse is pure, then the ``ffmpeg``
|
||||
subprocess (the CLI binary, detected at runtime), then ``LoudnessAnalyzer``
|
||||
with one daemon thread working through a queue.
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
from PyQt6.QtCore import QObject, QTimer, pyqtSignal
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
TARGET_LUFS = -14.0 # Spotify's and YouTube's level: loud enough for a party
|
||||
MAX_BOOST_DB = 12.0 # a near-silent intro track mustn't be blown up to hiss
|
||||
MAX_CUT_DB = 24.0
|
||||
SILENT_LUFS = -60.0 # ebur128 reports -70 for silence: nothing to match
|
||||
TYPICAL_LUFS = -10.0 # assumed for a song not measured yet, until a few are
|
||||
SAVE_DELAY_MS = 5000
|
||||
|
||||
_INTEGRATED = re.compile(r"Integrated loudness:\s*\n\s*I:\s*(-?[\d.]+|-inf)\s*LUFS")
|
||||
|
||||
|
||||
def parse_ebur128(output: str) -> float | None:
|
||||
"""The integrated loudness from ffmpeg's ebur128 summary, or None when
|
||||
there isn't one (no audio stream, a decode failure) or it's silence."""
|
||||
match = _INTEGRATED.search(output)
|
||||
if not match or match.group(1) == "-inf":
|
||||
return None
|
||||
lufs = float(match.group(1))
|
||||
return lufs if lufs > SILENT_LUFS else None
|
||||
|
||||
|
||||
def gain_db(lufs: float | None, target: float = TARGET_LUFS) -> float:
|
||||
"""How many dB to move a song measured at `lufs` to reach `target`.
|
||||
0 for an unmeasurable one, which is left as it is."""
|
||||
if lufs is None:
|
||||
return 0.0
|
||||
return max(-MAX_CUT_DB, min(MAX_BOOST_DB, target - lufs))
|
||||
|
||||
|
||||
def ffmpeg_available() -> bool:
|
||||
return shutil.which("ffmpeg") is not None
|
||||
|
||||
|
||||
def measure_lufs(path: str) -> float | None:
|
||||
"""Decode the whole file once through ebur128. None on any failure."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["ffmpeg", "-nostdin", "-nostats", "-hide_banner", "-i", path,
|
||||
"-map", "0:a:0", "-af", "ebur128=framelog=quiet", "-f", "null", "-"],
|
||||
capture_output=True, text=True, errors="replace", timeout=300)
|
||||
except (OSError, subprocess.SubprocessError) as exc:
|
||||
log.info("loudness: couldn't measure %s: %s", path, exc)
|
||||
return None
|
||||
return parse_ebur128(result.stderr)
|
||||
|
||||
|
||||
def default_cache_path() -> Path:
|
||||
base = os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache"
|
||||
return Path(base) / "lintunes" / "loudness.json"
|
||||
|
||||
|
||||
def _stamp(path: str):
|
||||
try:
|
||||
st = os.stat(path)
|
||||
except OSError:
|
||||
return None
|
||||
return [st.st_size, st.st_mtime_ns]
|
||||
|
||||
|
||||
class LoudnessAnalyzer(QObject):
|
||||
"""Measured loudness by file path, measuring what's asked for on a
|
||||
daemon thread. ``measured(path, lufs)`` arrives on the GUI thread (lufs
|
||||
may be None: tried, and there was nothing to measure)."""
|
||||
|
||||
measured = pyqtSignal(str, object)
|
||||
|
||||
def __init__(self, cache_path: Path | None = None, measure=measure_lufs,
|
||||
parent=None):
|
||||
super().__init__(parent)
|
||||
self._cache_path = cache_path or default_cache_path()
|
||||
self._measure = measure
|
||||
self._cache: dict | None = None # path -> [size, mtime_ns, lufs]
|
||||
self._queue: list[str] = []
|
||||
self._queued: set[str] = set()
|
||||
self._lock = threading.Lock()
|
||||
self._wake = threading.Condition(self._lock)
|
||||
self._thread: threading.Thread | None = None
|
||||
self._save_timer = QTimer(self)
|
||||
self._save_timer.setSingleShot(True)
|
||||
self._save_timer.setInterval(SAVE_DELAY_MS)
|
||||
self._save_timer.timeout.connect(self.save)
|
||||
self.measured.connect(self._remember)
|
||||
|
||||
# ---- the cache ----
|
||||
|
||||
def _entries(self) -> dict:
|
||||
if self._cache is None:
|
||||
try:
|
||||
data = json.loads(self._cache_path.read_text())
|
||||
self._cache = data if isinstance(data, dict) else {}
|
||||
except (OSError, ValueError):
|
||||
self._cache = {}
|
||||
return self._cache
|
||||
|
||||
def lufs(self, path: str):
|
||||
"""``(known, lufs)``: known is False when the file hasn't been
|
||||
measured (or changed since); lufs is None for a file that was
|
||||
measured and had nothing to measure."""
|
||||
entry = self._entries().get(path)
|
||||
if entry and len(entry) == 3 and entry[:2] == _stamp(path):
|
||||
return True, entry[2]
|
||||
return False, None
|
||||
|
||||
def typical_lufs(self) -> float:
|
||||
"""What to assume for a song not measured yet: the median of what's
|
||||
been measured, once there's enough of it to mean anything."""
|
||||
values = sorted(e[2] for e in self._entries().values()
|
||||
if len(e) == 3 and e[2] is not None)
|
||||
if len(values) < 10:
|
||||
return TYPICAL_LUFS
|
||||
return values[len(values) // 2]
|
||||
|
||||
def _remember(self, path: str, lufs):
|
||||
stamp = _stamp(path)
|
||||
if stamp is not None:
|
||||
self._entries()[path] = stamp + [lufs]
|
||||
self._save_timer.start()
|
||||
|
||||
def save(self):
|
||||
if self._cache is None:
|
||||
return
|
||||
self._save_timer.stop()
|
||||
try:
|
||||
self._cache_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = self._cache_path.with_suffix(".json.tmp")
|
||||
tmp.write_text(json.dumps(self._cache))
|
||||
tmp.replace(self._cache_path)
|
||||
except OSError as exc:
|
||||
log.warning("loudness: couldn't save %s: %s", self._cache_path, exc)
|
||||
|
||||
# ---- measuring ----
|
||||
|
||||
def request(self, path: str, urgent: bool = False):
|
||||
"""Measure `path` unless it's known or already waiting. An urgent
|
||||
one (the song starting now) jumps the queue."""
|
||||
if not path or self.lufs(path)[0]:
|
||||
return
|
||||
with self._lock:
|
||||
if path in self._queued:
|
||||
if urgent and self._queue and self._queue[0] != path:
|
||||
self._queue.remove(path)
|
||||
self._queue.insert(0, path)
|
||||
return
|
||||
self._queued.add(path)
|
||||
if urgent:
|
||||
self._queue.insert(0, path)
|
||||
else:
|
||||
self._queue.append(path)
|
||||
self._wake.notify()
|
||||
if self._thread is None:
|
||||
self._thread = threading.Thread(
|
||||
target=self._run, name="loudness", daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def _run(self):
|
||||
while True:
|
||||
with self._lock:
|
||||
while not self._queue:
|
||||
self._wake.wait()
|
||||
path = self._queue.pop(0)
|
||||
lufs = self._measure(path)
|
||||
with self._lock:
|
||||
self._queued.discard(path)
|
||||
self.measured.emit(path, lufs)
|
||||
+114
-3
@@ -1,3 +1,4 @@
|
||||
import math
|
||||
import random
|
||||
import time
|
||||
from pathlib import Path
|
||||
@@ -6,6 +7,7 @@ from PyQt6.QtCore import QObject, QTimer, QUrl, pyqtSignal
|
||||
from PyQt6.QtMultimedia import (
|
||||
QMediaPlayer, QAudioOutput, QAudioBufferOutput, QMediaDevices, QAudio)
|
||||
|
||||
from lintunes import loudness
|
||||
from lintunes.eventlog import log_control
|
||||
from lintunes.models import Track
|
||||
from lintunes.storage import listen_history
|
||||
@@ -79,6 +81,24 @@ def volume_with_adjustment(level: float, adjustment: int) -> float:
|
||||
return max(0.0, min(1.0, level * (1.0 + share)))
|
||||
|
||||
|
||||
_LOG100 = math.log(100.0)
|
||||
|
||||
|
||||
def volume_with_gain(level: float, gain_db: float) -> float:
|
||||
"""The perceptual 0..1 `level` with a gain in dB on it (volume
|
||||
normalization). Done in the linear amplitude QAudioOutput plays, through
|
||||
Qt's own ``QAudio.convertVolume`` curve, so −6 dB really halves the
|
||||
signal. Can't pass full — a quiet song is only lifted while the slider
|
||||
has room above it."""
|
||||
if not gain_db or level <= 0.0:
|
||||
return max(0.0, min(1.0, level))
|
||||
linear = 1.0 if level >= 1.0 else -math.log(1.0 - level) / _LOG100
|
||||
linear *= 10.0 ** (gain_db / 20.0)
|
||||
if linear >= 1.0:
|
||||
return 1.0
|
||||
return 1.0 - math.exp(-linear * _LOG100)
|
||||
|
||||
|
||||
class PlaybackSink(QObject):
|
||||
"""One audio output engine sitting behind Player.
|
||||
|
||||
@@ -464,6 +484,17 @@ class Player(QObject):
|
||||
self._current_key = None
|
||||
self._listen: _Listen | None = None
|
||||
self._volume = 1.0 # logical 0..1; what the volume slider shows
|
||||
# Volume normalization: a LoudnessAnalyzer, injected (Player stays
|
||||
# manager-free), and the gain it implies for the playing song. A song
|
||||
# not measured yet starts at a typical song's gain and glides to its
|
||||
# own once the measurement lands, a fraction of a second later.
|
||||
self.loudness = None
|
||||
self._normalize = False
|
||||
self._gain_db = 0.0
|
||||
self._gain_target = 0.0
|
||||
self._glide = QTimer(self)
|
||||
self._glide.setInterval(self.GLIDE_STEP_MS)
|
||||
self._glide.timeout.connect(self._glide_step)
|
||||
|
||||
self._local = LocalSink(self)
|
||||
self._sink: PlaybackSink = self._local
|
||||
@@ -651,10 +682,88 @@ class Player(QObject):
|
||||
self._sink.set_volume(self._output_volume())
|
||||
|
||||
def _output_volume(self) -> float:
|
||||
"""The slider's volume with the playing track's adjustment on it."""
|
||||
return volume_with_adjustment(
|
||||
"""The slider's volume with the playing track's adjustment and the
|
||||
normalization gain on it."""
|
||||
return volume_with_gain(volume_with_adjustment(
|
||||
self._volume,
|
||||
getattr(self._current_track, "volume_adjustment", 0) or 0)
|
||||
getattr(self._current_track, "volume_adjustment", 0) or 0),
|
||||
self._gain_db)
|
||||
|
||||
# ---- normalization ----
|
||||
|
||||
GLIDE_STEP_MS = 50
|
||||
GLIDE_DB_PER_STEP = 0.5 # 10 dB a second: quick, but never a jump
|
||||
LOOKAHEAD = 2 # songs measured ahead of the playing one
|
||||
|
||||
def set_normalize(self, on: bool):
|
||||
"""Turn volume normalization on or off; applies to the playing
|
||||
song at once, gliding rather than jumping."""
|
||||
on = bool(on) and self.loudness is not None
|
||||
if on == self._normalize:
|
||||
return
|
||||
self._normalize = on
|
||||
if on:
|
||||
self.loudness.measured.connect(self._on_measured)
|
||||
else:
|
||||
try:
|
||||
self.loudness.measured.disconnect(self._on_measured)
|
||||
except TypeError:
|
||||
pass
|
||||
self._aim_gain(self._current_track, glide=True)
|
||||
if on:
|
||||
self._measure_ahead()
|
||||
|
||||
def is_normalizing(self) -> bool:
|
||||
return self._normalize
|
||||
|
||||
def _aim_gain(self, track, glide: bool):
|
||||
"""Point the gain at `track`'s measured loudness (or a typical song's
|
||||
while it's being measured). Without glide it lands at once — before
|
||||
a load, so a loud song never starts loud."""
|
||||
target = 0.0
|
||||
if self._normalize and track is not None and track.location:
|
||||
known, lufs = self.loudness.lufs(track.location)
|
||||
if not known:
|
||||
self.loudness.request(track.location, urgent=True)
|
||||
lufs = self.loudness.typical_lufs()
|
||||
target = loudness.gain_db(lufs)
|
||||
self._gain_target = target
|
||||
if glide:
|
||||
if not self._glide.isActive() and target != self._gain_db:
|
||||
self._glide.start()
|
||||
else:
|
||||
self._glide.stop()
|
||||
self._gain_db = target
|
||||
|
||||
def _glide_step(self):
|
||||
step = self.GLIDE_DB_PER_STEP
|
||||
delta = self._gain_target - self._gain_db
|
||||
self._gain_db = (self._gain_target if abs(delta) <= step
|
||||
else self._gain_db + math.copysign(step, delta))
|
||||
self._sink.set_volume(self._output_volume())
|
||||
if self._gain_db == self._gain_target:
|
||||
self._glide.stop()
|
||||
|
||||
def _on_measured(self, path: str, _lufs):
|
||||
track = self._current_track
|
||||
if track is not None and track.location == path:
|
||||
self._aim_gain(track, glide=True)
|
||||
|
||||
def _measure_ahead(self):
|
||||
"""Queue the next few songs, so they're measured before they play."""
|
||||
if not self._normalize or not self._queue:
|
||||
return
|
||||
index = self._index
|
||||
order = (self._shuffle_order if self._shuffle
|
||||
else list(range(len(self._queue))))
|
||||
try:
|
||||
at = order.index(index)
|
||||
except ValueError:
|
||||
return
|
||||
for i in order[at + 1:at + 1 + self.LOOKAHEAD]:
|
||||
track = self._resolve(self._queue[i])
|
||||
if track is not None and track.location:
|
||||
self.loudness.request(track.location)
|
||||
|
||||
def volume(self) -> float:
|
||||
"""The logical 0..1 volume last set (what the slider shows)."""
|
||||
@@ -782,7 +891,9 @@ class Player(QObject):
|
||||
self._stop_at_ms = 0
|
||||
self._counted_finish_id = None
|
||||
# Before the load, so a quieter track never starts loud.
|
||||
self._aim_gain(track, glide=False)
|
||||
self._sink.set_volume(self._output_volume())
|
||||
self._measure_ahead()
|
||||
self._sink.load(
|
||||
track, autoplay,
|
||||
track.start_time if track.start_time > 0 else 0)
|
||||
|
||||
@@ -12,6 +12,7 @@ DEFAULTS = {
|
||||
"now_playing_font": None, # None=prefer Century Gothic, ""=use app
|
||||
# default, "Family"=use app
|
||||
"volume": 0.8, # logical 0..1 master output volume
|
||||
"normalize_volume": False, # play every song at one loudness (loudness.py)
|
||||
"visualizer_mode": "on", # on (color) | dim (gray) | off — click cycle
|
||||
"visualizer_style": "bars", # bars | radial | laser (visualizer_styles)
|
||||
# The styles' tunables — ranges in visualizer_styles.LIMITS.
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
"""Round 83: volume normalization.
|
||||
|
||||
Every song is measured once (ffmpeg's ebur128, integrated LUFS), cached per
|
||||
machine, and played at TARGET_LUFS by a gain on the output volume — a DJ
|
||||
mustn't have to ride the volume between a quiet old record and a loud new
|
||||
one. A song not measured yet starts at a typical song's gain and glides to
|
||||
its own when the measurement lands.
|
||||
"""
|
||||
import math
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from PyQt6.QtMultimedia import QAudio
|
||||
|
||||
from lintunes import loudness
|
||||
from lintunes.player import volume_with_gain
|
||||
from lintunes.preferences import Preferences
|
||||
from tests.test_round11 import _FakeLastFm
|
||||
from tests.test_round29 import _player, _local_track, _FakeSink
|
||||
|
||||
SUMMARY = """[Parsed_ebur128_0 @ 0x1] Summary:
|
||||
|
||||
Integrated loudness:
|
||||
I: -9.3 LUFS
|
||||
Threshold: -19.4 LUFS
|
||||
"""
|
||||
|
||||
|
||||
class _FakeAnalyzer(loudness.LoudnessAnalyzer):
|
||||
"""Measures nothing on its own: the test says what the file measured."""
|
||||
|
||||
def __init__(self, tmp_path, known=None):
|
||||
super().__init__(cache_path=tmp_path / "loudness.json",
|
||||
measure=lambda path: None)
|
||||
self.known = dict(known or {})
|
||||
self.requested = []
|
||||
|
||||
def lufs(self, path):
|
||||
return (True, self.known[path]) if path in self.known else (False, None)
|
||||
|
||||
def request(self, path, urgent=False):
|
||||
self.requested.append((path, urgent))
|
||||
|
||||
def land(self, path, lufs):
|
||||
self.known[path] = lufs
|
||||
self.measured.emit(path, lufs)
|
||||
|
||||
|
||||
def _linear(level):
|
||||
return QAudio.convertVolume(level, QAudio.VolumeScale.LogarithmicVolumeScale,
|
||||
QAudio.VolumeScale.LinearVolumeScale)
|
||||
|
||||
|
||||
class TestMath:
|
||||
def test_parse(self):
|
||||
assert loudness.parse_ebur128(SUMMARY) == -9.3
|
||||
|
||||
def test_parse_silence_and_garbage(self):
|
||||
assert loudness.parse_ebur128(SUMMARY.replace("-9.3", "-70.0")) is None
|
||||
assert loudness.parse_ebur128("No such file") is None
|
||||
|
||||
def test_gain_is_clamped(self):
|
||||
assert loudness.gain_db(-9.0) == -5.0
|
||||
assert loudness.gain_db(-40.0) == loudness.MAX_BOOST_DB
|
||||
assert loudness.gain_db(None) == 0.0
|
||||
|
||||
def test_gain_follows_qts_own_volume_curve(self):
|
||||
# -6 dB halves the amplitude QAudioOutput actually plays.
|
||||
out = volume_with_gain(0.8, -20 * math.log10(2))
|
||||
assert _linear(out) == pytest.approx(_linear(0.8) / 2, rel=1e-3)
|
||||
|
||||
def test_no_gain_and_no_volume_change_nothing(self):
|
||||
assert volume_with_gain(0.6, 0.0) == 0.6
|
||||
assert volume_with_gain(0.0, 6.0) == 0.0
|
||||
|
||||
def test_a_boost_cant_pass_full(self):
|
||||
assert volume_with_gain(0.95, 12.0) == 1.0
|
||||
|
||||
|
||||
class TestCache:
|
||||
def test_remembers_by_size_and_mtime(self, qapp, tmp_path):
|
||||
song = tmp_path / "a.mp3"
|
||||
song.write_bytes(b"x" * 10)
|
||||
cache = tmp_path / "cache" / "loudness.json"
|
||||
analyzer = loudness.LoudnessAnalyzer(cache_path=cache)
|
||||
assert analyzer.lufs(str(song)) == (False, None)
|
||||
analyzer._remember(str(song), -11.0)
|
||||
analyzer.save()
|
||||
again = loudness.LoudnessAnalyzer(cache_path=cache)
|
||||
assert again.lufs(str(song)) == (True, -11.0)
|
||||
song.write_bytes(b"x" * 11) # retagged: measure again
|
||||
assert again.lufs(str(song)) == (False, None)
|
||||
|
||||
def test_measures_on_its_thread(self, qapp, tmp_path):
|
||||
song = tmp_path / "a.mp3"
|
||||
song.write_bytes(b"x")
|
||||
got = []
|
||||
analyzer = loudness.LoudnessAnalyzer(
|
||||
cache_path=tmp_path / "l.json", measure=lambda p: -12.5)
|
||||
analyzer.measured.connect(lambda p, v: got.append((p, v)))
|
||||
analyzer.request(str(song))
|
||||
for _ in range(200):
|
||||
qapp.processEvents()
|
||||
if got:
|
||||
break
|
||||
time.sleep(0.01)
|
||||
assert got == [(str(song), -12.5)]
|
||||
assert analyzer.lufs(str(song)) == (True, -12.5)
|
||||
|
||||
|
||||
class TestPlayer:
|
||||
def _setup(self, qapp, tmp_path, known):
|
||||
tracks = [_local_track(tmp_path, i) for i in (1, 2, 3, 4)]
|
||||
player = _player(qapp, tracks)
|
||||
player.loudness = _FakeAnalyzer(
|
||||
tmp_path, {tracks[i - 1].location: v for i, v in known.items()})
|
||||
sink = _FakeSink()
|
||||
player.set_sink(sink)
|
||||
player.set_volume(0.8)
|
||||
return player, sink, tracks
|
||||
|
||||
def test_off_by_default(self, qapp, tmp_path):
|
||||
player, sink, _ = self._setup(qapp, tmp_path, {1: -6.0})
|
||||
player.play_queue([1, 2], 0)
|
||||
assert sink.volume == pytest.approx(0.8)
|
||||
assert player.loudness.requested == []
|
||||
|
||||
def test_a_measured_song_starts_at_its_own_level(self, qapp, tmp_path):
|
||||
player, sink, _ = self._setup(qapp, tmp_path, {1: -6.0, 2: -20.0})
|
||||
player.set_normalize(True)
|
||||
player.play_queue([1, 2], 0)
|
||||
assert sink.volume == pytest.approx(volume_with_gain(0.8, -8.0))
|
||||
player.next()
|
||||
assert sink.volume == pytest.approx(volume_with_gain(0.8, 6.0))
|
||||
assert player.volume() == 0.8 # the slider is untouched
|
||||
|
||||
def test_the_next_songs_are_measured_ahead(self, qapp, tmp_path):
|
||||
player, _, tracks = self._setup(qapp, tmp_path, {})
|
||||
player.set_normalize(True)
|
||||
player.play_queue([1, 2, 3, 4], 0)
|
||||
assert player.loudness.requested == [
|
||||
(tracks[0].location, True),
|
||||
(tracks[1].location, False), (tracks[2].location, False)]
|
||||
|
||||
def test_an_unmeasured_song_glides_to_its_level(self, qapp, tmp_path):
|
||||
player, sink, tracks = self._setup(qapp, tmp_path, {})
|
||||
player.set_normalize(True)
|
||||
player.play_queue([1], 0)
|
||||
typical = loudness.gain_db(loudness.TYPICAL_LUFS)
|
||||
assert sink.volume == pytest.approx(volume_with_gain(0.8, typical))
|
||||
player.loudness.land(tracks[0].location, -24.0)
|
||||
assert player._glide.isActive()
|
||||
while player._glide.isActive():
|
||||
player._glide_step()
|
||||
assert sink.volume == pytest.approx(volume_with_gain(0.8, 10.0))
|
||||
|
||||
def test_turning_it_off_glides_back(self, qapp, tmp_path):
|
||||
player, sink, _ = self._setup(qapp, tmp_path, {1: -6.0})
|
||||
player.set_normalize(True)
|
||||
player.play_queue([1], 0)
|
||||
player.set_normalize(False)
|
||||
while player._glide.isActive():
|
||||
player._glide_step()
|
||||
assert sink.volume == pytest.approx(0.8)
|
||||
|
||||
|
||||
def test_preferences_and_menu_agree(qapp, tmp_path, monkeypatch):
|
||||
from lintunes.gui.preferences_dialog import PreferencesDialog
|
||||
monkeypatch.setattr(loudness, "ffmpeg_available", lambda: True)
|
||||
prefs = Preferences(tmp_path)
|
||||
dialog = PreferencesDialog(prefs, _FakeLastFm())
|
||||
assert not dialog._normalize_check.isChecked()
|
||||
dialog._normalize_check.setChecked(True)
|
||||
assert prefs.get("normalize_volume") is True
|
||||
prefs.set("normalize_volume", False) # Controls → Normalize Volume
|
||||
assert not dialog._normalize_check.isChecked()
|
||||
dialog.deleteLater()
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="needs ffmpeg")
|
||||
def test_real_ffmpeg_measures_a_tone(tmp_path):
|
||||
song = tmp_path / "tone.flac"
|
||||
subprocess.run(["ffmpeg", "-loglevel", "error", "-f", "lavfi", "-i",
|
||||
"sine=f=1000:d=5:sample_rate=44100",
|
||||
str(song)], check=True)
|
||||
lufs = loudness.measure_lufs(str(song))
|
||||
assert lufs is not None and -26 < lufs < -16 # a 1/8-amplitude sine
|
||||
Reference in New Issue
Block a user