/
/
/
1"""Player queue configuration handling for the ConfigController."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, cast, overload
6
7from music_assistant_models.auth import Scope
8from music_assistant_models.config_entries import (
9 ConfigEntry,
10 ConfigValueOption,
11 ConfigValueType,
12 PlayerQueueConfig,
13)
14from music_assistant_models.enums import PlaybackState
15
16from music_assistant.constants import CONF_PLAYER_QUEUES, CONF_VALUE_GLOBAL
17from music_assistant.controllers.config.constants import (
18 PLAYER_QUEUE_CONFIG_OWNER,
19 _ConfigValueT,
20)
21from music_assistant.controllers.config.helpers import _with_translation_owner
22from music_assistant.controllers.player_queues.constants import CONF_AUTOPLAY_PLAYLIST
23from music_assistant.helpers.api import api_command
24
25if TYPE_CHECKING:
26 from music_assistant import MusicAssistant
27
28
29class PlayerQueueConfigMixin:
30 """Mixin providing player queue configuration handling for the ConfigController."""
31
32 # Type hints for attributes/methods provided by the class this mixin is used with
33 if TYPE_CHECKING:
34 mass: MusicAssistant
35
36 def get(self, key: str, default: Any = None) -> Any: ... # noqa: D102
37
38 def set(self, key: str, value: Any) -> None: ... # noqa: D102
39
40 @api_command("config/player_queues", required_scope=Scope.CONFIG_PLAYERS_READ)
41 def get_player_queue_configs(self) -> list[PlayerQueueConfig]:
42 """Return all (stored) queue configurations."""
43 return [
44 self._parse_player_queue_config(queue_id, raw_conf)
45 for queue_id, raw_conf in list(self.get(CONF_PLAYER_QUEUES, {}).items())
46 ]
47
48 @api_command("config/player_queues/get", required_scope=Scope.CONFIG_PLAYERS_READ)
49 async def get_player_queue_config_for_api(self, queue_id: str) -> PlayerQueueConfig:
50 """Return (full) configuration for a single queue, with dynamic options populated."""
51 # The frontend renders the queue settings form straight from this config, so the
52 # autoplay playlist dropdown options must be resolved here (the sync core is used by
53 # the streaming hot path and must stay free of library lookups).
54 config = self.get_player_queue_config(queue_id)
55 if (autoplay_playlist := config.values.get(CONF_AUTOPLAY_PLAYLIST)) is not None:
56 autoplay_playlist.options = await self._library_playlist_options()
57 return config
58
59 def get_player_queue_config(self, queue_id: str) -> PlayerQueueConfig:
60 """Return (full) configuration for a single queue."""
61 raw_conf = self.get(f"{CONF_PLAYER_QUEUES}/{queue_id}") or {"queue_id": queue_id}
62 return self._parse_player_queue_config(queue_id, raw_conf)
63
64 @api_command("config/player_queues/get_entries", required_scope=Scope.CONFIG_PLAYERS_READ)
65 async def get_player_queue_config_entries(
66 self,
67 queue_id: str,
68 action: str | None = None,
69 values: dict[str, ConfigValueType] | None = None,
70 ) -> list[ConfigEntry]:
71 """Return all Config Entries to configure a queue."""
72 entries = self.mass.player_queues.get_queue_config_entries(
73 playlist_options=await self._library_playlist_options()
74 )
75 return _with_translation_owner(entries, PLAYER_QUEUE_CONFIG_OWNER)
76
77 @api_command("config/player_queues/get_value", required_scope=Scope.CONFIG_PLAYERS_READ)
78 def get_player_queue_config_value(self, queue_id: str, key: str) -> ConfigValueType:
79 """Return single config(entry) value for a queue."""
80 return self.get_player_queue_config(queue_id).get_value(key)
81
82 if TYPE_CHECKING:
83
84 @overload
85 def get_raw_player_queue_config_value(
86 self, queue_id: str, key: str, default: _ConfigValueT
87 ) -> _ConfigValueT: ...
88
89 @overload
90 def get_raw_player_queue_config_value(
91 self, queue_id: str, key: str, default: None = None
92 ) -> ConfigValueType | None: ...
93
94 def get_raw_player_queue_config_value(
95 self, queue_id: str, key: str, default: ConfigValueType = None
96 ) -> ConfigValueType:
97 """
98 Return (raw) single config(entry) value for a queue.
99
100 Returns the stored value as-is (no validation), or the given default when not stored.
101 """
102 return cast(
103 "ConfigValueType",
104 self.get(f"{CONF_PLAYER_QUEUES}/{queue_id}/values/{key}", default),
105 )
106
107 def get_effective_player_queue_config_value(
108 self, queue_id: str, key: str, default: ConfigValueType = None
109 ) -> ConfigValueType:
110 """
111 Return the effective queue config value, following the global (queue controller) value.
112
113 A per-queue value of "global" (or unset) resolves to the matching value on the Player Queues
114 core controller, so a queue can either follow the global default or override it â mirroring
115 the log_level "GLOBAL" pattern.
116
117 :param queue_id: The queue to read the value for.
118 :param key: The config key.
119 :param default: The global (queue-controller) default to fall back to when the value is
120 stored on neither the queue nor the core config.
121 """
122 value = self.get_raw_player_queue_config_value(queue_id, key, CONF_VALUE_GLOBAL)
123 if value in (CONF_VALUE_GLOBAL, None):
124 # self is the ConfigController; go via mass.config for the typed (overloaded) accessor
125 return self.mass.config.get_raw_core_config_value(CONF_PLAYER_QUEUES, key, default)
126 return value
127
128 @api_command("config/player_queues/save", required_scope=Scope.CONFIG_PLAYERS_WRITE)
129 async def save_player_queue_config(
130 self, queue_id: str, values: dict[str, ConfigValueType]
131 ) -> PlayerQueueConfig:
132 """Save/update PlayerQueueConfig."""
133 config = self.get_player_queue_config(queue_id)
134 changed_keys = config.update(values)
135 conf_key = f"{CONF_PLAYER_QUEUES}/{queue_id}"
136 if not changed_keys and self.get(conf_key) is not None:
137 # no changes
138 return config
139 self.set(conf_key, config.to_raw())
140 if changed_keys and (queue := self.mass.player_queues.get(queue_id)):
141 # refresh derived queue state (e.g. the effective smart-fades indicator) and notify
142 # clients so they don't see a stale value until the next unrelated queue update
143 queue.smart_fades_active = self.mass.streams.is_smart_fades_active(queue)
144 queue.smart_shuffle_active = self.mass.player_queues.is_smart_shuffle_active(queue)
145 self.mass.player_queues.signal_update(queue_id)
146 # apply immediately: restart playback if a changed setting requires a reload
147 requires_restart = any(
148 v.requires_reload
149 for v in config.values.values()
150 if f"values/{v.key}" in changed_keys
151 )
152 if requires_restart and queue.state == PlaybackState.PLAYING:
153 await self.mass.player_queues.stop(queue_id)
154 self.mass.call_later(1, self.mass.player_queues.resume, queue_id, False)
155 return self.get_player_queue_config(queue_id)
156
157 def _parse_player_queue_config(
158 self, queue_id: str, raw_conf: dict[str, Any]
159 ) -> PlayerQueueConfig:
160 """Parse a (raw) queue config dict into a PlayerQueueConfig with the current entries."""
161 raw_conf = {**raw_conf, "queue_id": queue_id}
162 entries = self.mass.player_queues.get_queue_config_entries()
163 return cast("PlayerQueueConfig", PlayerQueueConfig.parse(entries, raw_conf))
164
165 async def _library_playlist_options(self) -> list[ConfigValueOption]:
166 """Return the library playlists as selectable config options (for autoplay playlist mode)."""
167 return [
168 ConfigValueOption(playlist.uri, title=playlist.name)
169 async for playlist in self.mass.music.playlists.iter_library_items()
170 ]
171