/
/
/
1"""
2Audio settings for the Spotify Soloist engine.
3
4The engine has no CLI or WebSocket control for crossfade, loudness normalization
5or stream quality: it reads them from the classic desktop-client prefs stores in
6its data directory, at startup only. Both the Spotify Connect backend and the
7Spotify music provider's playback backend therefore write these before every
8daemon spawn.
9"""
10
11from __future__ import annotations
12
13from typing import TYPE_CHECKING, Final
14
15from music_assistant.providers.spotify_connect.base import (
16 AUDIO_QUALITY_HIGH,
17 AUDIO_QUALITY_LOSSLESS,
18 AUDIO_QUALITY_NORMAL,
19 AUDIO_QUALITY_VERY_HIGH,
20)
21
22if TYPE_CHECKING:
23 import logging
24 from pathlib import Path
25
26# The classic desktop-client prefs keys (bare key=value lines) through which the
27# engine's audio behavior is controlled. The per-user prefs override the global
28# prefs per key, so both stores are (re)written before every daemon spawn.
29# NOTE: audio.crossfade.time_v2 is in MILLISECONDS; sub-second values silently
30# disable crossfade (verified empirically), so the key is only written when
31# crossfade is enabled (>= 1000 ms).
32PREF_CROSSFADE: Final = "audio.crossfade_v2"
33PREF_CROSSFADE_TIME: Final = "audio.crossfade.time_v2"
34PREF_NORMALIZE: Final = "audio.normalize_v2"
35# The engine reads the metered variant on a metered connection and the
36# non-metered one otherwise; both are written so the tier holds either way.
37# The "migrated" marker is what makes the engine honor the non-metered key
38# instead of deriving it once from the metered one.
39PREF_QUALITY: Final = "audio.play_bitrate_enumeration"
40PREF_QUALITY_NON_METERED: Final = "audio.play_bitrate_non_metered_enumeration"
41PREF_QUALITY_MIGRATED: Final = "audio.play_bitrate_non_metered_migrated"
42
43# Quality tier -> the engine's bitrate enumeration value. Measured against
44# build 1.3.7.349 on a 4:20 track (bytes fetched for the whole file): 2 and 3
45# deliver ~96 and ~160 kbps, 4 ~320 kbps and 5 lossless FLAC (~810 kbps).
46# 5 is the ceiling â values outside 1-5 are rejected and silently fall back to
47# ~160 kbps, so an unknown tier must never reach the prefs file.
48_QUALITY_VALUES: Final[dict[str, int]] = {
49 AUDIO_QUALITY_NORMAL: 2,
50 AUDIO_QUALITY_HIGH: 3,
51 AUDIO_QUALITY_VERY_HIGH: 4,
52 AUDIO_QUALITY_LOSSLESS: 5,
53}
54
55
56def write_audio_prefs(
57 data_dir: Path,
58 logger: logging.Logger,
59 *,
60 crossfade_ms: int = 0,
61 loudness_normalization: bool = False,
62 audio_quality: str | None = None,
63) -> bool:
64 """
65 Write the given audio behavior into the engine's prefs stores (blocking).
66
67 Both the global prefs and every existing per-user prefs file are updated:
68 per-user values override the global ones per key, and a per-user file only
69 appears after an account paired â the global store covers that account's
70 first session until the next daemon (re)spawn refreshes both.
71
72 Errors are logged per store and reported back rather than raised: a caller
73 that only nudges the engine's behavior can carry on, while one that tells the
74 rest of the server what the engine is doing has to know the write landed.
75
76 :param data_dir: The daemon's data directory (holds the settings stores).
77 :param logger: Logger to report per-store write failures on.
78 :param crossfade_ms: Crossfade duration in milliseconds (0 disables crossfade).
79 :param loudness_normalization: Whether the engine normalizes loudness itself.
80 :param audio_quality: One of the AUDIO_QUALITY_* tiers, or None to leave the
81 quality the engine is already configured with untouched.
82 :return: True when every store was written.
83 """
84 managed_lines = [
85 f"{PREF_CROSSFADE}={'true' if crossfade_ms else 'false'}",
86 f"{PREF_NORMALIZE}={'true' if loudness_normalization else 'false'}",
87 ]
88 if crossfade_ms:
89 managed_lines.insert(1, f"{PREF_CROSSFADE_TIME}={crossfade_ms}")
90 # every crossfade key is dropped even when only the boolean is rewritten:
91 # leaving a stale time behind would keep a previous session's crossfade on
92 written = True
93 managed_keys = {PREF_CROSSFADE, PREF_CROSSFADE_TIME, PREF_NORMALIZE}
94 if audio_quality is not None:
95 quality = _QUALITY_VALUES.get(audio_quality, _QUALITY_VALUES[AUDIO_QUALITY_LOSSLESS])
96 managed_lines += [
97 f"{PREF_QUALITY}={quality}",
98 f"{PREF_QUALITY_NON_METERED}={quality}",
99 f"{PREF_QUALITY_MIGRATED}=true",
100 ]
101 # a caller that does not manage the quality tier leaves the engine's own
102 managed_keys |= {PREF_QUALITY, PREF_QUALITY_NON_METERED, PREF_QUALITY_MIGRATED}
103 settings_dir = data_dir / "settings"
104 prefs_files = [settings_dir / "prefs"]
105 try:
106 users_dir = settings_dir / "Users"
107 if users_dir.is_dir():
108 prefs_files += [
109 user_dir / "prefs" for user_dir in users_dir.iterdir() if user_dir.is_dir()
110 ]
111 except OSError as err:
112 # a per-user store overrides the global one per key, so one we cannot even
113 # enumerate may still be holding a stale value: report failure
114 logger.warning("Failed to list the Spotify per-user settings: %s", err)
115 written = False
116 for prefs_file in prefs_files:
117 try:
118 lines = []
119 if prefs_file.is_file():
120 lines = [
121 line
122 for line in prefs_file.read_text(encoding="utf-8").splitlines()
123 if line.split("=", 1)[0] not in managed_keys
124 ]
125 prefs_file.parent.mkdir(parents=True, exist_ok=True)
126 # the stores also carry engine-owned keys, so replace atomically
127 # (a truncated in-place write would lose those too)
128 tmp_file = prefs_file.with_suffix(".tmp")
129 tmp_file.write_text("\n".join([*lines, *managed_lines]) + "\n", encoding="utf-8")
130 tmp_file.replace(prefs_file)
131 except (OSError, UnicodeDecodeError) as err:
132 logger.warning("Failed to write the Spotify audio settings to %s: %s", prefs_file, err)
133 written = False
134 return written
135