/
/
/
1"""DSP configuration handling for the ConfigController."""
2
3from __future__ import annotations
4
5import asyncio
6import base64
7import binascii
8import os
9from contextlib import suppress
10from copy import deepcopy
11from typing import TYPE_CHECKING, Any, Final
12
13import aiofiles
14import shortuuid
15from music_assistant_models.auth import Scope
16from music_assistant_models.dsp import ConvolutionFilter, DSPConfig, DSPConfigPreset
17from music_assistant_models.enums import EventType
18from music_assistant_models.errors import InvalidDataError
19
20from music_assistant.constants import (
21 CONF_PLAYER_DSP,
22 CONF_PLAYER_DSP_IRS,
23 CONF_PLAYER_DSP_PRESETS,
24 DSP_IR_ID_RE,
25 DSP_IRS_DIRNAME,
26)
27from music_assistant.helpers.api import api_command
28from music_assistant.helpers.process import check_output
29from music_assistant.helpers.security import is_safe_path
30from music_assistant.helpers.tags import async_parse_tags
31
32if TYPE_CHECKING:
33 from music_assistant import MusicAssistant
34# cap the accepted upload so a caller cannot exhaust memory or disk; IRs are small
35MAX_IR_BYTES: Final = 10 * 1024 * 1024
36# afir rejects anything over 30 seconds at playback time, so cap it at upload
37MAX_IR_SECONDS: Final = 10.0
38# bound the transcode so a crafted upload cannot leave ffmpeg running forever
39IR_TRANSCODE_TIMEOUT: Final = 60.0
40# a mono IR is applied to both channels and a stereo one channel per side, but afir
41# would silently downmix anything wider (e.g. a four channel true stereo IR) to match
42# the stream, blending the measurements instead of applying them
43MAX_IR_CHANNELS: Final = 2
44
45
46class DSPConfigMixin:
47 """Mixin providing DSP configuration handling for the ConfigController."""
48
49 # Type hints for attributes/methods provided by the class this mixin is used with
50 if TYPE_CHECKING:
51 mass: MusicAssistant
52
53 def get(self, key: str, default: Any = None) -> Any: ... # noqa: D102
54
55 def set(self, key: str, value: Any) -> None: ... # noqa: D102
56
57 def remove(self, key: str) -> None: ... # noqa: D102
58
59 @api_command("config/players/dsp/get", required_scope=Scope.CONFIG_PLAYERS_READ)
60 def get_player_dsp_config(self, player_id: str) -> DSPConfig:
61 """
62 Return the DSP Configuration for a player.
63
64 In case the player does not have a DSP configuration, a default one is returned.
65 """
66 if raw_conf := self.get(f"{CONF_PLAYER_DSP}/{player_id}"):
67 return DSPConfig.from_dict(raw_conf)
68 # return default DSP config
69 dsp_config = DSPConfig()
70 # The DSP config does not do anything by default, so we disable it
71 dsp_config.enabled = False
72 return dsp_config
73
74 @api_command("config/players/dsp/save", required_scope=Scope.CONFIG_PLAYERS_WRITE)
75 async def save_dsp_config(self, player_id: str, config: DSPConfig) -> DSPConfig:
76 """
77 Save/update DSPConfig for a player.
78
79 This method will validate the config and apply it to the player.
80 """
81 config = deepcopy(config)
82 config.preset_id = None
83 return await self._save_dsp_config(player_id, config)
84
85 @api_command("config/players/dsp/apply_preset", required_scope=Scope.CONFIG_PLAYERS_WRITE)
86 async def apply_dsp_preset(self, player_id: str, preset_id: str) -> DSPConfig:
87 """
88 Apply a persisted DSP preset to a player.
89
90 :param player_id: Player that should use the preset.
91 :param preset_id: Preset identifier to apply.
92 """
93 if (preset := self._get_dsp_preset(preset_id)) is None:
94 msg = f"DSP preset {preset_id} not found"
95 raise KeyError(msg)
96 config = deepcopy(preset.config)
97 config.preset_id = preset_id
98 return await self._save_dsp_config(player_id, config)
99
100 @api_command("config/dsp_presets/get", required_scope=Scope.CONFIG_PLAYERS_READ)
101 async def get_dsp_presets(self) -> list[DSPConfigPreset]:
102 """Return all user-defined DSP presets."""
103 raw_presets = self.get(CONF_PLAYER_DSP_PRESETS, {})
104 return [DSPConfigPreset.from_dict(preset) for preset in raw_presets.values()]
105
106 @api_command("config/dsp_presets/save", required_scope=Scope.CONFIG_PLAYERS_WRITE)
107 async def save_dsp_presets(self, preset: DSPConfigPreset) -> DSPConfigPreset:
108 """
109 Save/update a user-defined DSP presets.
110
111 This method will validate the config before saving it to the persistent storage.
112 """
113 preset = deepcopy(preset)
114 preset.config.preset_id = None
115 preset.validate()
116 self._validate_dsp_ir_refs(preset.config)
117
118 previous = self._get_dsp_preset(preset.preset_id) if preset.preset_id else None
119 if preset.preset_id is None:
120 # Generate a new preset_id if it does not exist
121 preset.preset_id = shortuuid.random(8).lower()
122
123 # Save the preset to the persistent storage
124 self.set(f"{CONF_PLAYER_DSP_PRESETS}/preset_{preset.preset_id}", preset.to_dict())
125 if previous:
126 previous_config = deepcopy(previous.config)
127 previous_config.preset_id = None
128 if previous_config != preset.config:
129 self._clear_dsp_preset_assignments(preset.preset_id)
130
131 all_presets = await self.get_dsp_presets()
132
133 self.mass.signal_event(
134 EventType.DSP_PRESETS_UPDATED,
135 data=all_presets,
136 )
137
138 return preset
139
140 @api_command("config/dsp_presets/remove", required_scope=Scope.CONFIG_PLAYERS_WRITE)
141 async def remove_dsp_preset(self, preset_id: str) -> None:
142 """Remove a user-defined DSP preset."""
143 self.remove(f"{CONF_PLAYER_DSP_PRESETS}/preset_{preset_id}")
144 self._clear_dsp_preset_assignments(preset_id)
145
146 all_presets = await self.get_dsp_presets()
147
148 self.mass.signal_event(
149 EventType.DSP_PRESETS_UPDATED,
150 data=all_presets,
151 )
152
153 @api_command("config/dsp_irs/list", required_scope=Scope.CONFIG_PLAYERS_READ)
154 def get_dsp_irs(self) -> list[dict[str, Any]]:
155 """Return the metadata for all stored convolution impulse responses."""
156 stored: dict[str, dict[str, Any]] = self.get(CONF_PLAYER_DSP_IRS, {})
157 return list(stored.values())
158
159 @api_command("config/dsp_irs/upload", required_scope=Scope.CONFIG_PLAYERS_WRITE)
160 async def upload_dsp_ir(self, name: str, data: str) -> dict[str, Any]:
161 """
162 Store a convolution impulse response and return its metadata record.
163
164 :param name: Display name for the impulse response.
165 :param data: Base64 encoded contents of the audio file to store.
166 """
167 # base64 inflates by roughly 4/3, so bound the encoded input before decoding
168 # to cap memory, then confirm the decoded size against the real limit
169 if len(data) > MAX_IR_BYTES // 3 * 4 + 8:
170 raise InvalidDataError("Impulse response file exceeds the size limit")
171 try:
172 raw = base64.b64decode(data, validate=True)
173 except binascii.Error as err:
174 raise InvalidDataError("Impulse response data is not valid base64") from err
175 if len(raw) > MAX_IR_BYTES:
176 raise InvalidDataError("Impulse response file exceeds the size limit")
177
178 ir_dir = self._dsp_irs_dir()
179 await asyncio.to_thread(os.makedirs, ir_dir, exist_ok=True)
180 ir_id = shortuuid.random(8).lower()
181 upload_path = os.path.join(ir_dir, f".{ir_id}.upload")
182 ir_path = self._dsp_ir_path(ir_id)
183
184 async with aiofiles.open(upload_path, "wb") as ir_file:
185 await ir_file.write(raw)
186 try:
187 # transcoding to wav both validates the upload is decodable audio and
188 # gives afir a format it can always read back at playback time
189 returncode, output = await check_output(
190 "ffmpeg",
191 "-hide_banner",
192 "-loglevel",
193 "error",
194 "-y",
195 # restrict to the file protocol so a crafted upload (e.g. a playlist
196 # or concat script) cannot make ffmpeg reach out over the network
197 "-protocol_whitelist",
198 "file",
199 "-i",
200 upload_path,
201 # bound the decoded output so a small but very long input cannot fill
202 # the disk before the duration check below runs. the extra second keeps
203 # an over-long response above the cap, so that check still rejects it
204 "-t",
205 str(MAX_IR_SECONDS + 1),
206 "-c:a",
207 "pcm_f32le",
208 ir_path,
209 timeout=IR_TRANSCODE_TIMEOUT,
210 )
211 if returncode != 0:
212 msg = output.decode(errors="replace").strip()
213 raise InvalidDataError(f"Uploaded file is not valid audio: {msg}")
214 tags = await async_parse_tags(ir_path)
215 if tags.channels > MAX_IR_CHANNELS:
216 raise InvalidDataError(
217 f"Impulse response has {tags.channels} channels: "
218 "only mono and stereo files are supported"
219 )
220 if tags.duration is not None and tags.duration > MAX_IR_SECONDS:
221 raise InvalidDataError(
222 f"Impulse response is {tags.duration:.1f} seconds long: "
223 f"the limit is {MAX_IR_SECONDS:.0f} seconds"
224 )
225 except TimeoutError:
226 await self._remove_file(ir_path)
227 raise InvalidDataError("Impulse response took too long to convert") from None
228 except BaseException:
229 # drop the IR if any upload error happened
230 await self._remove_file(ir_path)
231 raise
232 finally:
233 await self._remove_file(upload_path)
234
235 record = {
236 "ir_id": ir_id,
237 "name": name,
238 "sample_rate": tags.sample_rate,
239 "channels": tags.channels,
240 "duration": tags.duration,
241 }
242 stored: dict[str, dict[str, Any]] = self.get(CONF_PLAYER_DSP_IRS, {})
243 stored[ir_id] = record
244 self.set(CONF_PLAYER_DSP_IRS, stored)
245
246 self.mass.signal_event(
247 EventType.DSP_IRS_UPDATED,
248 data=self.get_dsp_irs(),
249 )
250
251 return record
252
253 @api_command("config/dsp_irs/remove", required_scope=Scope.CONFIG_PLAYERS_WRITE)
254 async def remove_dsp_ir(self, ir_id: str) -> None:
255 """Remove a stored convolution impulse response by its identifier."""
256 stored: dict[str, dict[str, Any]] = self.get(CONF_PLAYER_DSP_IRS, {})
257 known = stored.pop(ir_id, None) is not None
258 if known:
259 self.set(CONF_PLAYER_DSP_IRS, stored)
260 removed_file = False
261 try:
262 ir_path = self._dsp_ir_path(ir_id)
263 except InvalidDataError:
264 # an unusable stored id has no file to remove, but the record is gone
265 if not known:
266 raise
267 else:
268 removed_file = await self._remove_file(ir_path)
269 cleared = await self._clear_dsp_ir_assignments(ir_id)
270 if not (known or removed_file or cleared):
271 return
272
273 self.mass.signal_event(
274 EventType.DSP_IRS_UPDATED,
275 data=self.get_dsp_irs(),
276 )
277
278 def _get_dsp_preset(self, preset_id: str | None) -> DSPConfigPreset | None:
279 """Return a DSP preset by identifier."""
280 if preset_id is None:
281 return None
282 if raw_preset := self.get(f"{CONF_PLAYER_DSP_PRESETS}/preset_{preset_id}"):
283 return DSPConfigPreset.from_dict(raw_preset)
284 return None
285
286 async def _save_dsp_config(self, player_id: str, config: DSPConfig) -> DSPConfig:
287 """Persist and apply a validated DSP configuration."""
288 config.validate()
289 self._validate_dsp_ir_refs(config)
290 previous = self.get_player_dsp_config(player_id)
291 self.set(f"{CONF_PLAYER_DSP}/{player_id}", config.to_dict())
292 if previous.enabled or config.enabled:
293 await self.mass.players.on_player_dsp_change(player_id)
294 elif previous.preset_id != config.preset_id:
295 self.mass.streams.audio_processing.update_player_dsp_preset(
296 player_id,
297 config.preset_id,
298 )
299 self.mass.signal_event(
300 EventType.PLAYER_DSP_CONFIG_UPDATED,
301 object_id=player_id,
302 data=config,
303 )
304 return config
305
306 def _clear_dsp_preset_assignments(self, preset_id: str) -> None:
307 """Clear a preset selection without changing player DSP values."""
308 raw_configs: dict[str, dict[str, Any]] = self.get(CONF_PLAYER_DSP, {})
309 for player_id, raw_config in tuple(raw_configs.items()):
310 config = DSPConfig.from_dict(raw_config)
311 if config.preset_id != preset_id:
312 continue
313 config.preset_id = None
314 self.set(f"{CONF_PLAYER_DSP}/{player_id}", config.to_dict())
315 self.mass.streams.audio_processing.update_player_dsp_preset(player_id, None)
316 self.mass.signal_event(
317 EventType.PLAYER_DSP_CONFIG_UPDATED,
318 object_id=player_id,
319 data=config,
320 )
321
322 async def _clear_dsp_ir_assignments(self, ir_id: str) -> bool:
323 """
324 Blank a removed impulse response from any player config or preset using it.
325
326 :param ir_id: The impulse response identifier to clear.
327 :return: True if any config or preset was changed.
328 """
329 cleared = False
330 raw_configs: dict[str, dict[str, Any]] = self.get(CONF_PLAYER_DSP, {})
331 for player_id, raw_config in tuple(raw_configs.items()):
332 config = DSPConfig.from_dict(raw_config)
333 if not _blank_convolution_ir(config, ir_id):
334 continue
335 self.set(f"{CONF_PLAYER_DSP}/{player_id}", config.to_dict())
336 cleared = True
337 if config.enabled:
338 # dropping the convolution changes what the player should hear, so the
339 # stream has to be rebuilt the same way a saved config change does it
340 await self.mass.players.on_player_dsp_change(player_id)
341 self.mass.signal_event(
342 EventType.PLAYER_DSP_CONFIG_UPDATED,
343 object_id=player_id,
344 data=config,
345 )
346 raw_presets: dict[str, dict[str, Any]] = self.get(CONF_PLAYER_DSP_PRESETS, {})
347 presets_changed = False
348 for preset_key, raw_preset in tuple(raw_presets.items()):
349 preset = DSPConfigPreset.from_dict(raw_preset)
350 if not _blank_convolution_ir(preset.config, ir_id):
351 continue
352 self.set(f"{CONF_PLAYER_DSP_PRESETS}/{preset_key}", preset.to_dict())
353 cleared = True
354 presets_changed = True
355
356 if presets_changed:
357 self.mass.signal_event(
358 EventType.DSP_PRESETS_UPDATED,
359 data=await self.get_dsp_presets(),
360 )
361 return cleared
362
363 def _validate_dsp_ir_refs(self, config: DSPConfig) -> None:
364 """
365 Reject a config naming an impulse response this server does not hold.
366
367 :param config: The DSP configuration to check.
368 """
369 stored: dict[str, dict[str, Any]] = self.get(CONF_PLAYER_DSP_IRS, {})
370 for dsp_filter in config.filters:
371 # an empty id is the "none selected yet" value, which stays allowed
372 if not isinstance(dsp_filter, ConvolutionFilter) or not dsp_filter.ir_id:
373 continue
374 if dsp_filter.ir_id not in stored:
375 raise InvalidDataError(f"Unknown impulse response: {dsp_filter.ir_id!r}")
376
377 def _dsp_irs_dir(self) -> str:
378 """Return the directory holding convolution impulse response files."""
379 return os.path.join(self.mass.storage_path, DSP_IRS_DIRNAME)
380
381 def _dsp_ir_path(self, ir_id: str) -> str:
382 """
383 Return the on-disk path for an impulse response, rejecting unsafe ids.
384
385 :param ir_id: The impulse response identifier to resolve.
386 """
387 ir_dir = self._dsp_irs_dir()
388 ir_path = os.path.join(ir_dir, f"{ir_id}.wav")
389 if not DSP_IR_ID_RE.match(ir_id) or not is_safe_path(ir_path, ir_dir):
390 raise InvalidDataError(f"Invalid impulse response id: {ir_id!r}")
391 return ir_path
392
393 async def _remove_file(self, path: str) -> bool:
394 """Delete a file if it exists, reporting whether it was there."""
395 with suppress(FileNotFoundError):
396 await asyncio.to_thread(os.remove, path)
397 return True
398 return False
399
400
401def _blank_convolution_ir(config: DSPConfig, ir_id: str) -> bool:
402 """
403 Blank any convolution filter in the config that references ir_id, in place.
404
405 :param config: The DSP configuration to update.
406 :param ir_id: The impulse response identifier to clear.
407 :return: True if the config was changed.
408 """
409 cleared = False
410 for dsp_filter in config.filters:
411 if isinstance(dsp_filter, ConvolutionFilter) and dsp_filter.ir_id == ir_id:
412 dsp_filter.ir_id = ""
413 cleared = True
414 return cleared
415