music-assistant-server

1"""Player configuration handling for the ConfigController."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from copy import deepcopy
8from typing import TYPE_CHECKING, Any, Literal, cast, overload
9
10from music_assistant_models.auth import Scope
11from music_assistant_models.config_entries import (
12    ConfigActionResult,
13    ConfigEntry,
14    ConfigValueOption,
15    ConfigValueType,
16    PlayerConfig,
17)
18from music_assistant_models.constants import (
19    PLAYER_CONTROL_FAKE,
20    PLAYER_CONTROL_NATIVE,
21    PLAYER_CONTROL_NONE,
22)
23from music_assistant_models.enums import (
24    ConfigEntryType,
25    EventType,
26    PlayerFeature,
27    PlayerType,
28)
29from music_assistant_models.errors import (
30    ActionUnavailable,
31    UnsupportedFeaturedException,
32)
33
34from music_assistant.constants import (
35    CONF_ENABLED,
36    CONF_ENTRY_ANNOUNCE_VOLUME,
37    CONF_ENTRY_ANNOUNCE_VOLUME_MAX,
38    CONF_ENTRY_ANNOUNCE_VOLUME_MIN,
39    CONF_ENTRY_ANNOUNCE_VOLUME_STRATEGY,
40    CONF_ENTRY_AUTO_PLAY,
41    CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES,
42    CONF_ENTRY_ENABLE_ICY_METADATA,
43    CONF_ENTRY_FLOW_MODE,
44    CONF_ENTRY_FLOW_MODE_SAMPLE_RATE,
45    CONF_ENTRY_HTTP_PROFILE,
46    CONF_ENTRY_MAX_VOLUME,
47    CONF_ENTRY_MIN_VOLUME,
48    CONF_ENTRY_OUTPUT_CHANNELS,
49    CONF_ENTRY_OUTPUT_CODEC,
50    CONF_ENTRY_PLAY_MEDIA_OVERRIDES_GROUP,
51    CONF_ENTRY_PLAYER_ICON,
52    CONF_ENTRY_PLAYER_ICON_GROUP,
53    CONF_ENTRY_SAMPLE_RATES,
54    CONF_ENTRY_TTS_PRE_ANNOUNCE,
55    CONF_EXPOSE_PLAYER_TO_HA,
56    CONF_HIDE_IN_UI,
57    CONF_ICON,
58    CONF_MUTE_CONTROL,
59    CONF_PLAYERS,
60    CONF_POWER_CONTROL,
61    CONF_PRE_ANNOUNCE_CHIME_URL,
62    CONF_PREFERRED_OUTPUT_PROTOCOL,
63    CONF_PROTOCOL_CATEGORY_PREFIX,
64    CONF_PROTOCOL_KEY_SPLITTER,
65    CONF_UNDERLYING_PLAYER_ID,
66    CONF_VOLUME_CONTROL,
67    NON_HTTP_PROVIDERS,
68    PLAYER_CONTROL_PROTOCOL,
69)
70from music_assistant.controllers.config.constants import BASE_KEYS, _ConfigValueT
71from music_assistant.controllers.config.helpers import _with_translation_owner
72from music_assistant.helpers.api import api_command
73from music_assistant.helpers.util import validate_announcement_chime_url
74from music_assistant.providers.sync_group.constants import SGP_PREFIX
75from music_assistant.providers.universal_group.constants import UGP_PREFIX
76
77if TYPE_CHECKING:
78    from music_assistant_models.player import OutputProtocol
79
80    from music_assistant import MusicAssistant
81    from music_assistant.models.player import Player
82
83
84LOGGER = logging.getLogger(__name__)
85
86
87def _first_enabled_control_value(options: list[ConfigValueOption]) -> ConfigValueType:
88    """
89    Return the value of the first selectable option, so a disabled option is never the default.
90
91    Player-control selects always list the "native" option (shown disabled when the feature is
92    unsupported); this picks the first enabled option as the entry default and falls back to the
93    always-present "none" control.
94
95    :param options: The control entry's options, in display order.
96    """
97    for option in options:
98        if not option.disabled:
99            return option.value
100    return PLAYER_CONTROL_NONE
101
102
103def _reconcile_player_icon_value(
104    submitted_values: dict[str, ConfigValueType],
105    stored_values: dict[str, Any],
106    new_values: dict[str, ConfigValueType],
107) -> bool:
108    """Persist explicit icon selections while keeping None as automatic selection."""
109    stored_icon = stored_values.get(CONF_ICON)
110    has_explicit_icon = isinstance(stored_icon, str) and bool(stored_icon)
111    if CONF_ICON not in submitted_values:
112        if has_explicit_icon:
113            new_values[CONF_ICON] = stored_icon
114        else:
115            new_values.pop(CONF_ICON, None)
116        return False
117
118    submitted_icon = submitted_values[CONF_ICON]
119    if isinstance(submitted_icon, str) and submitted_icon:
120        new_values[CONF_ICON] = submitted_icon
121        return not has_explicit_icon or submitted_icon != stored_icon
122
123    new_values.pop(CONF_ICON, None)
124    return has_explicit_icon
125
126
127def _apply_raw_player_icon_value(
128    config: PlayerConfig,
129    raw_values: dict[str, Any],
130) -> None:
131    """Apply the stored icon value to a parsed player config."""
132    if icon_entry := config.values.get(CONF_ICON):
133        stored_icon = raw_values.get(CONF_ICON)
134        icon_entry.value = stored_icon if isinstance(stored_icon, str) and stored_icon else None
135
136
137class PlayerConfigMixin:
138    """Mixin providing player configuration handling for the ConfigController."""
139
140    # Type hints for attributes/methods provided by the class this mixin is used with
141    if TYPE_CHECKING:
142        mass: MusicAssistant
143
144        def get(self, key: str, default: Any = None) -> Any: ...  # noqa: D102
145
146        def set(self, key: str, value: Any) -> None: ...  # noqa: D102
147
148        def remove(self, key: str) -> None: ...  # noqa: D102
149
150    @api_command("config/players", required_scope=Scope.CONFIG_PLAYERS_READ)
151    async def get_player_configs(
152        self,
153        provider: str | None = None,
154        include_values: bool = False,
155        include_unavailable: bool = True,
156        include_disabled: bool = True,
157    ) -> list[PlayerConfig]:
158        """Return all known player configurations, optionally filtered by provider id."""
159        result: list[PlayerConfig] = []
160        for key, raw_conf in list(self.get(CONF_PLAYERS, {}).items()):
161            # guard against malformed entries that lost their base keys
162            # (can happen via race between delete_player_config and a stale player
163            # update writing back a nested sub-key, which recreates a partial dict).
164            if not isinstance(raw_conf, dict) or "player_id" not in raw_conf:
165                LOGGER.warning("Removing malformed player config entry %s (missing player_id)", key)
166                self.remove(f"{CONF_PLAYERS}/{key}")
167                continue
168            # optional provider filter
169            if provider is not None and raw_conf.get("provider") != provider:
170                continue
171            # filter out unavailable players
172            # (unless disabled, otherwise there is no way to re-enable them)
173            # note that we only check for missing players in the player controller,
174            # and we do allow players that are temporary unavailable
175            # (player.state.available = false) because this can also mean that the
176            # player needs additional configuration such as airplay devices that need pairing.
177            player = self.mass.players.get_player(raw_conf["player_id"], False)
178            if not include_unavailable and player is None and raw_conf.get("enabled", True):
179                continue
180            # filter out protocol players
181            # their configuration is handled differently as part of their parent player
182            if raw_conf.get("player_type") == PlayerType.PROTOCOL or (
183                player and player.state.type == PlayerType.PROTOCOL
184            ):
185                continue
186            # filter out disabled players
187            if not include_disabled and not raw_conf.get("enabled", True):
188                continue
189            if include_values:
190                result.append(await self.get_player_config(raw_conf["player_id"]))
191            else:
192                summary_conf = deepcopy(raw_conf)
193                summary_conf["default_name"] = (
194                    player.state.name if player else summary_conf.get("default_name")
195                )
196                summary_conf["available"] = player.state.available if player else False
197                result.append(cast("PlayerConfig", PlayerConfig.parse([], summary_conf)))
198        return result
199
200    @api_command("config/players/get", required_scope=Scope.CONFIG_PLAYERS_READ)
201    async def get_player_config(
202        self,
203        player_id: str,
204    ) -> PlayerConfig:
205        """Return (full) configuration for a single player."""
206        raw_conf: dict[str, Any]
207        if raw_conf := self.get(f"{CONF_PLAYERS}/{player_id}"):
208            raw_conf = deepcopy(raw_conf)
209            # protocol-prefixed entries are virtual mirrors of the linked protocol player's own
210            # config (the canonical store). Drop any that linger in this player's persisted values
211            # so a stale copy can never shadow the protocol player's current value; the live values
212            # are merged back in from the protocol player(s) below.
213            if stored_values := raw_conf.get("values"):
214                for key in [key for key in stored_values if CONF_PROTOCOL_KEY_SPLITTER in key]:
215                    del stored_values[key]
216            if player := self.mass.players.get_player(player_id, False):
217                raw_conf["default_name"] = player.state.name
218                raw_conf["provider"] = player.provider.instance_id
219                config_entries = await self.get_player_config_entries(
220                    player_id,
221                )
222                # also grab (raw) values for protocol outputs
223                if protocol_values := await self._get_output_protocol_config_values(config_entries):
224                    if "values" not in raw_conf:
225                        raw_conf["values"] = {}
226                    raw_conf["values"].update(protocol_values)
227            else:
228                # handle unavailable player and/or provider
229                config_entries = []
230                raw_conf["available"] = False
231                raw_conf["default_name"] = (
232                    raw_conf.get("default_name") or raw_conf.get("player_id") or player_id
233                )
234                raw_conf.setdefault("player_id", player_id)
235
236            conf = cast("PlayerConfig", PlayerConfig.parse(config_entries, raw_conf))
237            _apply_raw_player_icon_value(conf, raw_conf.get("values", {}))
238            # parse() stamps every entry with this player's owner; injected protocol entries
239            # belong to their own protocol provider, so restore that owner for string resolution.
240            for entry in conf.values.values():
241                if CONF_PROTOCOL_KEY_SPLITTER not in entry.key:
242                    continue
243                protocol_player_id = entry.key.split(CONF_PROTOCOL_KEY_SPLITTER, 1)[0]
244                if protocol_player := self.mass.players.get_player(protocol_player_id, False):
245                    entry.translation_owner = protocol_player.translation_owner
246            return conf
247        msg = f"No config found for player id {player_id}"
248        raise KeyError(msg)
249
250    @api_command("config/players/get_entries", required_scope=Scope.CONFIG_PLAYERS_READ)
251    async def get_player_config_entries(self, player_id: str) -> list[ConfigEntry]:
252        """
253        Return Config entries to configure a player.
254
255        :param player_id: id of an existing player instance.
256        """
257        if not (player := self.mass.players.get_player(player_id, False)):
258            msg = f"Player {player_id} not found"
259            raise KeyError(msg)
260
261        default_entries: list[ConfigEntry]
262        player_entries: list[ConfigEntry]
263        if player.state.type == PlayerType.PROTOCOL:
264            default_entries = []
265            player_entries = await self._get_player_config_entries(player)
266        else:
267            # get default entries which are common for all (non protocol)players
268            default_entries = self._get_default_player_config_entries(player)
269
270            # get player(protocol) specific entries
271            # this basically injects virtual config entries for each protocol output
272            # this feels maybe a bit of a hack to do it this way but it keeps the UI logic simple
273            # and maximizes api client compatibility because you can configure the whole player
274            # including its protocols from a single config endpoint without needing special handling
275            # for protocol players in the UI/api clients
276            if protocol_entries := await self._create_output_protocol_config_entries(player):
277                player_entries = protocol_entries
278                if not any(protocol.is_native for protocol in player.output_protocols):
279                    # A control-only player (e.g. a device that delegates playback to a
280                    # linked DLNA protocol player) has no native output protocol, so the
281                    # block above never injects the player's own entries. Append them here
282                    # so it keeps its own config surface, skipping keys the protocol
283                    # entries already cover.
284                    protocol_keys = {entry.key for entry in protocol_entries}
285                    player_entries = [
286                        *protocol_entries,
287                        *[
288                            entry
289                            for entry in await self._get_player_config_entries(player)
290                            if entry.key not in protocol_keys
291                        ],
292                    ]
293            else:
294                player_entries = await self._get_player_config_entries(player)
295
296        player_entries_keys = {entry.key for entry in player_entries}
297        all_entries = [
298            # ignore default entries that were overridden by the player specific ones
299            *[x for x in default_entries if x.key not in player_entries_keys],
300            *player_entries,
301        ]
302        return _with_translation_owner(all_entries, player.translation_owner)
303
304    @api_command("config/players/invoke_action", required_scope=Scope.CONFIG_PLAYERS_WRITE)
305    async def invoke_player_config_action(
306        self, player_id: str, action: str
307    ) -> list[ConfigEntry] | ConfigActionResult:
308        """
309        Run a one-shot action button from a player's config.
310
311        A protocol-prefixed action (``<protocol_player_id>||protocol||<action>``) is routed to
312        the linked protocol player; the parent player's entries are then re-rendered so the
313        injected protocol entries pick up any state change. A ``ConfigActionResult`` holds the
314        outcome to report to the user; an empty list means the action ran with nothing to
315        report; a non-empty list holds the parent player's entries the config form should
316        re-render with.
317
318        :param player_id: The player whose config surface holds the action.
319        :param action: The action id of the pressed button (may be protocol-prefixed).
320        """
321        if not (player := self.mass.players.get_player(player_id, False)):
322            msg = f"Player {player_id} not found"
323            raise KeyError(msg)
324        if CONF_PROTOCOL_KEY_SPLITTER in action:
325            protocol_player_id, protocol_action = action.split(CONF_PROTOCOL_KEY_SPLITTER, 1)
326            if not (target := self.mass.players.get_player(protocol_player_id, False)):
327                msg = f"Player {protocol_player_id} not found"
328                raise KeyError(msg)
329            result = await target.handle_config_action(protocol_action)
330        else:
331            target = player
332            result = await player.handle_config_action(action)
333        if result is None:
334            return []
335        if isinstance(result, ConfigActionResult):
336            # the strings belong to the provider that handled the action, which for a
337            # protocol-prefixed action is the protocol player's, not the host player's
338            result.translation_owner = result.translation_owner or target.translation_owner
339            return result
340        # re-render the full (parent) player entries so injected protocol entries refresh
341        return await self.get_player_config_entries(player_id)
342
343    @overload
344    async def get_player_config_value(
345        self,
346        player_id: str,
347        key: str,
348        unpack_splitted_values: Literal[True],
349        *,
350        default: ConfigValueType = ...,
351        return_type: type[_ConfigValueT] | None = ...,
352    ) -> tuple[str, ...] | list[tuple[str, ...]]: ...
353
354    @overload
355    async def get_player_config_value(
356        self,
357        player_id: str,
358        key: str,
359        unpack_splitted_values: Literal[False] = False,
360        *,
361        default: _ConfigValueT,
362        return_type: type[_ConfigValueT] = ...,
363    ) -> _ConfigValueT: ...
364
365    @overload
366    async def get_player_config_value(
367        self,
368        player_id: str,
369        key: str,
370        unpack_splitted_values: Literal[False] = False,
371        *,
372        default: ConfigValueType = ...,
373        return_type: type[_ConfigValueT] = ...,
374    ) -> _ConfigValueT: ...
375
376    @overload
377    async def get_player_config_value(
378        self,
379        player_id: str,
380        key: str,
381        unpack_splitted_values: Literal[False] = False,
382        *,
383        default: ConfigValueType = ...,
384        return_type: None = ...,
385    ) -> ConfigValueType: ...
386
387    @api_command("config/players/get_value", required_scope=Scope.CONFIG_PLAYERS_READ)
388    async def get_player_config_value(
389        self,
390        player_id: str,
391        key: str,
392        unpack_splitted_values: bool = False,
393        *,
394        default: ConfigValueType = None,
395        return_type: type[_ConfigValueT | ConfigValueType] | None = None,
396    ) -> _ConfigValueT | ConfigValueType | tuple[str, ...] | list[tuple[str, ...]]:
397        """
398        Return single configentry value for a player.
399
400        :param player_id: The player ID.
401        :param key: The config key to retrieve.
402        :param unpack_splitted_values: Whether to unpack multi-value config entries.
403        :param default: Optional default value to return if key is not found.
404        :param return_type: Optional type hint for type inference (e.g., str, int, bool).
405            Note: This parameter is used purely for static type checking and does not
406            perform runtime type validation. Callers are responsible for ensuring the
407            specified type matches the actual config value type.
408        """
409        # prefer stored value so we don't have to retrieve all config entries every time
410        if (raw_value := self.get_raw_player_config_value(player_id, key)) is not None:
411            if not unpack_splitted_values:
412                return raw_value
413        conf = await self.get_player_config(player_id)
414        if key not in conf.values:
415            if default is not None:
416                return default
417            msg = f"Config key {key} not found for player {player_id}"
418            raise KeyError(msg)
419        if unpack_splitted_values:
420            return conf.values[key].get_splitted_values()
421        return (
422            conf.values[key].value
423            if conf.values[key].value is not None
424            else conf.values[key].default_value
425        )
426
427    if TYPE_CHECKING:
428        # Overload for when default is provided - return type matches default type
429        @overload
430        def get_raw_player_config_value(
431            self, player_id: str, key: str, default: _ConfigValueT
432        ) -> _ConfigValueT: ...
433
434        # Overload for when no default is provided - return ConfigValueType | None
435        @overload
436        def get_raw_player_config_value(
437            self, player_id: str, key: str, default: None = None
438        ) -> ConfigValueType | None: ...
439
440    def get_raw_player_config_value(
441        self, player_id: str, key: str, default: ConfigValueType = None
442    ) -> ConfigValueType:
443        """
444        Return (raw) single configentry value for a player.
445
446        Note that this only returns the stored value without any validation or default.
447        """
448        return cast(
449            "ConfigValueType",
450            self.get(
451                f"{CONF_PLAYERS}/{player_id}/values/{key}",
452                self.get(f"{CONF_PLAYERS}/{player_id}/{key}", default),
453            ),
454        )
455
456    def get_base_player_config(self, player_id: str, provider: str) -> PlayerConfig:
457        """
458        Return base PlayerConfig for a player.
459
460        This is used to get the base config for a player, without any provider specific values,
461        for initialization purposes.
462        """
463        if not (raw_conf := self.get(f"{CONF_PLAYERS}/{player_id}")):
464            raw_conf = {
465                "player_id": player_id,
466                "provider": provider,
467            }
468        return cast("PlayerConfig", PlayerConfig.parse([], raw_conf))
469
470    @api_command("config/players/save", required_scope=Scope.CONFIG_PLAYERS_WRITE)
471    async def save_player_config(
472        self, player_id: str, values: dict[str, ConfigValueType]
473    ) -> PlayerConfig:
474        """Save/update PlayerConfig."""
475        values = await self._update_output_protocol_config(values)
476        conf_key = f"{CONF_PLAYERS}/{player_id}"
477        existing_raw = self.get(conf_key) or {}
478        existing_values = existing_raw.get("values", {})
479        if values.get(CONF_ICON) is None and CONF_ICON not in existing_values:
480            values = {key: value for key, value in values.items() if key != CONF_ICON}
481        config = await self.get_player_config(player_id)
482        changed_keys = config.update(values)
483        new_raw = config.to_raw()
484        new_values = new_raw.get("values", {})
485        # Preserve values from storage that don't have config entries in current context.
486        config_entry_keys = set(config.values.keys())
487        for key, value in existing_values.items():
488            if key not in new_values and key not in config_entry_keys:
489                new_values[key] = value
490        # never persist protocol-prefixed (virtual) entries on this player; the linked protocol
491        # player is the canonical store (handled by _update_output_protocol_config). Storing a copy
492        # here would shadow the protocol player's value once it is reset back to its default.
493        new_values = {
494            key: value for key, value in new_values.items() if CONF_PROTOCOL_KEY_SPLITTER not in key
495        }
496        if _reconcile_player_icon_value(values, existing_values, new_values):
497            changed_keys.add(f"values/{CONF_ICON}")
498        _apply_raw_player_icon_value(config, new_values)
499        if not changed_keys:
500            # no changes
501            return config
502        # store updated config first (to prevent issues with enabling/disabling players)
503        new_raw["values"] = new_values
504        self.set(conf_key, new_raw)
505        try:
506            # validate/handle the update in the player manager
507            await self.mass.players.on_player_config_change(config, changed_keys)
508        except Exception:
509            # rollback on error - use existing_raw to preserve all values
510            self.set(conf_key, existing_raw)
511            raise
512        # send config updated event
513        self.mass.signal_event(
514            EventType.PLAYER_CONFIG_UPDATED,
515            object_id=config.player_id,
516            data=config,
517        )
518        # return full player config (just in case)
519        return await self.get_player_config(player_id)
520
521    @api_command("config/players/remove", required_scope=Scope.CONFIG_PLAYERS_WRITE)
522    async def remove_player_config(self, player_id: str) -> None:
523        """Remove PlayerConfig."""
524        conf_key = f"{CONF_PLAYERS}/{player_id}"
525        player_config = self.get(conf_key)
526        if not player_config:
527            msg = f"Player configuration for {player_id} does not exist"
528            raise KeyError(msg)
529        if self.mass.players.get_player(player_id):
530            try:
531                await self.mass.players.remove(player_id)
532            except UnsupportedFeaturedException:
533                # removing a player config while it is active is not allowed
534                # unless the provider reports it has the remove_player feature
535                raise ActionUnavailable("Can not remove config for an active player!")
536            # tell the player manager to remove the player if its lingering around
537            # set permanent to false otherwise we end up in an infinite loop
538            await self.mass.players.unregister(player_id, permanent=False)
539        # all of the above passed, so wipe the config (incl. DSP and linked protocol players)
540        self.mass.players.delete_player_config(player_id)
541
542    def set_player_default_name(self, player_id: str, default_name: str) -> None:
543        """Set (or update) the default name for a player."""
544        # skip if the player config root no longer exists, otherwise the
545        # nested set would resurrect a partial entry (missing player_id etc).
546        if not self.get(f"{CONF_PLAYERS}/{player_id}"):
547            return
548        conf_key = f"{CONF_PLAYERS}/{player_id}/default_name"
549        self.set(conf_key, default_name)
550
551    def set_player_type(self, player_id: str, player_type: PlayerType) -> None:
552        """Set (or update) the type for a player."""
553        # skip if the player config root no longer exists, otherwise the
554        # nested set would resurrect a partial entry (missing player_id etc).
555        if not self.get(f"{CONF_PLAYERS}/{player_id}"):
556            return
557        conf_key = f"{CONF_PLAYERS}/{player_id}/player_type"
558        self.set(conf_key, player_type)
559
560    def create_default_player_config(
561        self,
562        player_id: str,
563        provider: str,
564        player_type: PlayerType,
565        name: str | None = None,
566        enabled: bool = True,
567        values: dict[str, ConfigValueType] | None = None,
568    ) -> None:
569        """
570        Create default/empty PlayerConfig.
571
572        This is meant as helper to create default configs when a player is registered.
573        Called by the player manager on player register.
574        """
575        # return early if the config already exists
576        if existing_conf := self.get(f"{CONF_PLAYERS}/{player_id}"):
577            # update default name if needed
578            if name and name != existing_conf.get("default_name"):
579                self.set(f"{CONF_PLAYERS}/{player_id}/default_name", name)
580            # deliberately do NOT update player_type here: this is called from
581            # Player.__init__ where the type can still be a transient class default.
582            # Genuine type changes are persisted by update_state after registration.
583            return
584        # config does not yet exist, create a default one
585        conf_key = f"{CONF_PLAYERS}/{player_id}"
586        default_conf = PlayerConfig(
587            values={},
588            provider=provider,
589            player_id=player_id,
590            enabled=enabled,
591            name=name,
592            default_name=name,
593            player_type=player_type,
594        )
595        default_conf_raw = default_conf.to_raw()
596        if values is not None:
597            default_conf_raw["values"] = values
598        self.set(
599            conf_key,
600            default_conf_raw,
601        )
602
603    def set_raw_player_config_value(self, player_id: str, key: str, value: ConfigValueType) -> None:
604        """
605        Set (raw) single config(entry) value for a player.
606
607        Note that this only stores the (raw) value without any validation or default.
608        """
609        if not self.get(f"{CONF_PLAYERS}/{player_id}"):
610            # only allow setting raw values if main entry exists
611            msg = f"Invalid player_id: {player_id}"
612            raise KeyError(msg)
613        if key in BASE_KEYS:
614            self.set(f"{CONF_PLAYERS}/{player_id}/{key}", value)
615        else:
616            self.set(f"{CONF_PLAYERS}/{player_id}/values/{key}", value)
617            # also update the player's in-place config copy so object-local
618            # value reads stay in sync with raw writes
619            if (player := self.mass.players.get_player(player_id, False)) and (
620                entry := player.config.values.get(key)
621            ):
622                entry.value = value
623
624    async def _get_player_config_entries(
625        self,
626        player: Player,
627    ) -> list[ConfigEntry]:
628        """
629        Return Player(protocol) specific config entries, without any default entries.
630
631        In general this returns entries that are specific to this provider/player type only,
632        and includes audio related entries that are not part of the default set.
633
634        :param player: the player instance
635        """
636        default_entries: list[ConfigEntry]
637        is_dedicated_group_player = player.state.type in (
638            PlayerType.GROUP,
639            PlayerType.STEREO_PAIR,
640        ) and not player.player_id.startswith((UGP_PREFIX, SGP_PREFIX))
641        is_http_based_player_protocol = player.provider.domain not in NON_HTTP_PROVIDERS
642        if player.state.type == PlayerType.GROUP and not is_dedicated_group_player:
643            # no audio related entries for universal group players or sync group players
644            default_entries = []
645        elif PlayerFeature.PLAY_MEDIA not in player.supported_features:
646            # no audio related entries for players that do not support play_media
647            default_entries = []
648        else:
649            # default output/audio related entries
650            default_entries = [
651                # output channel is always configurable per player(protocol)
652                CONF_ENTRY_OUTPUT_CHANNELS
653            ]
654            if is_http_based_player_protocol:
655                # for http based players we can add the http streaming related entries
656                default_entries += [
657                    CONF_ENTRY_OUTPUT_CODEC,
658                    CONF_ENTRY_HTTP_PROFILE,
659                    CONF_ENTRY_ENABLE_ICY_METADATA,
660                ]
661                # only inject the sample-rates config when the player can't declare its rates itself
662                if not player.declares_supported_sample_rates:
663                    default_entries.append(CONF_ENTRY_SAMPLE_RATES)
664                # add flow mode entry for http-based players that do not already enforce it
665                if not player.requires_flow_mode:
666                    default_entries.append(CONF_ENTRY_FLOW_MODE)
667                default_entries.append(CONF_ENTRY_FLOW_MODE_SAMPLE_RATE)
668        if PlayerFeature.GAPLESS_PLAYBACK in player.supported_features:
669            default_entries.append(CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES)
670        # request player specific entries
671        player_entries = await player.get_config_entries()
672        players_keys = {entry.key for entry in player_entries}
673        # filter out any default entries that are already provided by the player
674        default_entries = [entry for entry in default_entries if entry.key not in players_keys]
675        return [*player_entries, *default_entries]
676
677    def _get_default_player_config_entries(self, player: Player) -> list[ConfigEntry]:
678        """
679        Return the default (generic) player config entries.
680
681        This does not return audio/protocol specific entries, those are handled elsewhere.
682        """
683        entries: list[ConfigEntry] = []
684        # default protocol-player config entries
685        if player.state.type == PlayerType.PROTOCOL:
686            # protocol players have no generic config entries
687            # only audio/protocol specific ones
688            return []
689
690        icon_entry = deepcopy(
691            CONF_ENTRY_PLAYER_ICON_GROUP
692            if player.state.type == PlayerType.GROUP
693            else CONF_ENTRY_PLAYER_ICON
694        )
695        icon_entry.default_value = player.default_icon
696        icon_entry.value = self.get_raw_player_config_value(player.player_id, CONF_ICON)
697
698        # some base entries for all player types
699        # note that these may NOT be playback/audio related
700        entries += [
701            CONF_ENTRY_TTS_PRE_ANNOUNCE,
702            ConfigEntry(
703                key=CONF_PRE_ANNOUNCE_CHIME_URL,
704                type=ConfigEntryType.STRING,
705                category="announcements",
706                required=False,
707                depends_on=CONF_ENTRY_TTS_PRE_ANNOUNCE.key,
708                depends_on_value=True,
709                validate=lambda val: validate_announcement_chime_url(cast("str", val)),
710            ),
711            # add player control entries
712            *self._create_player_control_config_entries(player),
713            # add entry to hide player in UI
714            ConfigEntry(
715                key=CONF_HIDE_IN_UI,
716                type=ConfigEntryType.BOOLEAN,
717                default_value=player.hidden_by_default,
718                category="generic",
719                advanced=False,
720            ),
721            # add entry to expose player to HA
722            ConfigEntry(
723                key=CONF_EXPOSE_PLAYER_TO_HA,
724                type=ConfigEntryType.BOOLEAN,
725                category="generic",
726                advanced=False,
727                default_value=player.expose_to_ha_by_default,
728            ),
729        ]
730        # group-player config entries
731        if player.state.type == PlayerType.GROUP:
732            entries += [
733                icon_entry,
734            ]
735            return entries
736        # normal player (or stereo pair) config entries
737        entries += [
738            icon_entry,
739            # add default entries for announce feature
740            CONF_ENTRY_ANNOUNCE_VOLUME_STRATEGY,
741            CONF_ENTRY_ANNOUNCE_VOLUME,
742            CONF_ENTRY_ANNOUNCE_VOLUME_MIN,
743            CONF_ENTRY_ANNOUNCE_VOLUME_MAX,
744            # play_media-on-self preference (only relevant to non-group players)
745            CONF_ENTRY_PLAY_MEDIA_OVERRIDES_GROUP,
746        ]
747        return entries
748
749    def _create_player_control_config_entries(self, player: Player) -> list[ConfigEntry]:
750        """Create config entries for player controls."""
751        is_group = player.state.type == PlayerType.GROUP
752        all_controls = self.mass.players.player_controls()
753        power_controls = [x for x in all_controls if x.supports_power]
754        volume_controls = [x for x in all_controls if x.supports_volume]
755        mute_controls = [x for x in all_controls if x.supports_mute]
756        auto_option = ConfigValueOption(PLAYER_CONTROL_PROTOCOL)
757        # the "native" option is always listed (disabled when the feature is unsupported) so the
758        # option set is consistent across players; the entry default skips disabled options.
759        power_options: list[ConfigValueOption] = [
760            ConfigValueOption(
761                PLAYER_CONTROL_NATIVE,
762                disabled=not player.supports_feature(PlayerFeature.POWER),
763            )
764        ]
765        has_native_volume_control = player.supports_feature(PlayerFeature.VOLUME_SET)
766        volume_options: list[ConfigValueOption] = [
767            ConfigValueOption(PLAYER_CONTROL_NATIVE, disabled=not has_native_volume_control)
768        ]
769        mute_options: list[ConfigValueOption] = [
770            ConfigValueOption(
771                PLAYER_CONTROL_NATIVE,
772                disabled=not player.supports_feature(PlayerFeature.VOLUME_MUTE),
773            )
774        ]
775        # add player protocols as volume controls if native player has no volume control
776        for linked_protocol in player.linked_output_protocols:
777            if has_native_volume_control:
778                break
779            protocol_player = self.mass.players.get_player(linked_protocol.output_protocol_id)
780            if not protocol_player or not protocol_player.available_for_playback:
781                continue
782            if protocol_player.supports_feature(PlayerFeature.VOLUME_SET):
783                if auto_option not in volume_options:
784                    volume_options.append(auto_option)
785                if linked_protocol.protocol_domain in ("chromecast", "dlna"):
786                    # for chromecast/dlna we can use the protocol player for volume control
787                    # even if the protocol player is not the active protocol
788                    volume_options.append(
789                        ConfigValueOption(
790                            protocol_player.player_id, title=protocol_player.provider.name
791                        )
792                    )
793            if protocol_player.supports_feature(PlayerFeature.VOLUME_MUTE):
794                if auto_option not in mute_options:
795                    mute_options.append(auto_option)
796                if linked_protocol.protocol_domain in ("chromecast", "dlna"):
797                    # for chromecast/dlna we can use the protocol player for volume control
798                    # even if the protocol player is not the active protocol
799                    mute_options.append(
800                        ConfigValueOption(
801                            protocol_player.player_id, title=protocol_player.provider.name
802                        )
803                    )
804
805        # append none+fake options
806        power_options += [
807            ConfigValueOption(PLAYER_CONTROL_NONE),
808            ConfigValueOption(PLAYER_CONTROL_FAKE),
809        ]
810        volume_options += [
811            ConfigValueOption(PLAYER_CONTROL_NONE),
812        ]
813        mute_options.append(ConfigValueOption(PLAYER_CONTROL_NONE))
814        # fake mute drives the volume control, so offer it when the player has any
815        # usable volume path (native or via a linked protocol player)
816        if player.supports_feature(PlayerFeature.VOLUME_SET) or auto_option in volume_options:
817            mute_options.append(ConfigValueOption(PLAYER_CONTROL_FAKE))
818
819        # return final config entries for all options
820        return [
821            # Power control config entry
822            ConfigEntry(
823                key=CONF_POWER_CONTROL,
824                type=ConfigEntryType.STRING,
825                default_value=_first_enabled_control_value(power_options),
826                required=False,
827                options=[
828                    *power_options,
829                    *(ConfigValueOption(x.id, title=x.name) for x in power_controls),
830                ],
831                category="player_controls",
832            ),
833            # Volume control config entry
834            ConfigEntry(
835                key=CONF_VOLUME_CONTROL,
836                type=ConfigEntryType.STRING,
837                default_value=_first_enabled_control_value(volume_options),
838                required=True,
839                options=[
840                    *volume_options,
841                    *(ConfigValueOption(x.id, title=x.name) for x in volume_controls),
842                ],
843                category="player_controls",
844            ),
845            # Mute control config entry
846            ConfigEntry(
847                key=CONF_MUTE_CONTROL,
848                type=ConfigEntryType.STRING,
849                default_value=_first_enabled_control_value(mute_options),
850                required=True,
851                options=[
852                    *mute_options,
853                    *[ConfigValueOption(x.id, title=x.name) for x in mute_controls],
854                ],
855                category="player_controls",
856            ),
857            # Volume limit entries
858            CONF_ENTRY_MIN_VOLUME,
859            CONF_ENTRY_MAX_VOLUME,
860            # auto-play on power on — only meaningful for individual players.
861            # For group players, power on/off is purely a "capture members"
862            # toggle (Fake control) and auto-starting playback there causes
863            # surprise playback when the user just wanted to pin the group.
864            *([] if is_group else [CONF_ENTRY_AUTO_PLAY]),
865        ]
866
867    async def _create_output_protocol_config_entries(  # noqa: PLR0915
868        self,
869        player: Player,
870    ) -> list[ConfigEntry]:
871        """
872        Create the output protocol config entries for a player.
873
874        The preferred output protocol entry is always returned, listing outputs that can not
875        be selected right now as disabled options and hidden altogether when the player has
876        at most one output. The settings of each output whose provider is loaded follow.
877
878        :param player: The player to create the output protocol config entries for.
879        """
880        all_entries: list[ConfigEntry] = []
881        output_protocols = player.output_protocols
882
883        # Resolve derived-transport edges (e.g. a Sendspin bridge riding on the
884        # AirPlay protocol) so derived protocols render as dependent on their base.
885        base_protocols: dict[str, OutputProtocol] = {}
886        for protocol in output_protocols:
887            if protocol.is_native:
888                continue
889            underlying_id = self.get_raw_player_config_value(
890                protocol.output_protocol_id, CONF_UNDERLYING_PLAYER_ID
891            )
892            if not underlying_id:
893                continue
894            if base_protocol := next(
895                (p for p in output_protocols if p.output_protocol_id == underlying_id), None
896            ):
897                base_protocols[protocol.output_protocol_id] = base_protocol
898
899        # Build options from all output protocols, sorted by priority
900        options: list[ConfigValueOption] = []
901
902        # Add each output protocol as an option, sorted by priority. An output that can not be
903        # used right now is offered disabled with the reason why, rather than left out entirely:
904        # that keeps the device's outputs recognizable and explains what to do about it.
905        has_native = False
906        for protocol in sorted(output_protocols, key=lambda p: p.priority):
907            protocol_name = self._get_protocol_display_name(protocol.protocol_domain)
908            # Use "native" for native playback,
909            # otherwise use the protocol output id (=player id)
910            if protocol.is_native:
911                title = f"{protocol_name} (native)"
912            elif base_protocol := base_protocols.get(protocol.output_protocol_id):
913                title = (
914                    f"{protocol_name} "
915                    f"(over {self._get_protocol_display_name(base_protocol.protocol_domain)})"
916                )
917            else:
918                title = protocol_name
919            value = "native" if protocol.is_native else protocol.output_protocol_id
920            options.append(
921                ConfigValueOption(
922                    value,
923                    title=title,
924                    disabled=not protocol.available,
925                    # the option's value is a player id, so its reason is keyed by this slug
926                    translation_key=(
927                        None
928                        if protocol.available
929                        else self._output_protocol_unavailable_reason(player, protocol)
930                    ),
931                )
932            )
933            # never default to an output that can not be selected
934            has_native = has_native or (protocol.is_native and protocol.available)
935
936        if has_native:
937            default_value = "native"
938        else:
939            default_value = "auto"
940            options.append(ConfigValueOption("auto"))
941
942        all_entries.append(
943            ConfigEntry(
944                key=CONF_PREFERRED_OUTPUT_PROTOCOL,
945                type=ConfigEntryType.STRING,
946                default_value=default_value,
947                required=True,
948                options=options,
949                category="protocol_general",
950                requires_reload=False,
951                hidden=len(output_protocols) <= 1,
952            )
953        )
954
955        # Add config entries for all protocol players/outputs
956        for protocol in output_protocols:
957            domain = protocol.protocol_domain
958            protocol_name = self._get_protocol_display_name(domain)
959            protocol_player_enabled = self.get_raw_player_config_value(
960                protocol.output_protocol_id, CONF_ENABLED, True
961            )
962            provider_available = self.mass.get_provider(protocol.protocol_domain) is not None
963            if not provider_available:
964                # protocol provider is not available, skip adding entries
965                continue
966            protocol_prefix = f"{protocol.output_protocol_id}{CONF_PROTOCOL_KEY_SPLITTER}"
967            protocol_enabled_key = f"{protocol_prefix}enabled"
968            protocol_category = f"{CONF_PROTOCOL_CATEGORY_PREFIX}_{domain}"
969            category_translation_key = "protocol_output_settings"
970            category_translation_params = [protocol_name]
971            # Derived protocols (e.g. Sendspin over AirPlay) render with their base
972            # protocol in the label and follow its enabled state.
973            base_protocol = base_protocols.get(protocol.output_protocol_id)
974            base_name: str | None = None
975            base_enabled = True
976            if base_protocol is not None:
977                base_name = self._get_protocol_display_name(base_protocol.protocol_domain)
978                base_raw_conf = self.get(f"{CONF_PLAYERS}/{base_protocol.output_protocol_id}") or {}
979                base_enabled = bool(base_raw_conf.get("enabled", True))
980                category_translation_key = "protocol_output_settings_via"
981                category_translation_params = [protocol_name, base_name]
982            if not protocol.is_native:
983                all_entries.append(
984                    ConfigEntry(
985                        key=protocol_enabled_key,
986                        type=ConfigEntryType.BOOLEAN,
987                        # the key is per-protocol (dynamic), so pin a static catalog key
988                        translation_key="protocol_enable_via" if base_name else "protocol_enable",
989                        translation_params=[base_name] if base_name else None,
990                        # a derived protocol cannot be active while its base is disabled
991                        value=bool(protocol_player_enabled) and base_enabled,
992                        read_only=not base_enabled,
993                        default_value=True,
994                        category=protocol_category,
995                        category_translation_key=category_translation_key,
996                        category_translation_params=category_translation_params,
997                        requires_reload=False,
998                    )
999                )
1000            if protocol.is_native:
1001                # add protocol-specific entries from native player
1002                protocol_entries = await self._get_player_config_entries(player)
1003                for proto_entry in protocol_entries:
1004                    # deep copy to avoid mutating shared/constant ConfigEntry objects
1005                    entry = deepcopy(proto_entry)
1006                    entry.category = protocol_category
1007                    entry.category_translation_key = category_translation_key
1008                    entry.category_translation_params = category_translation_params
1009                    all_entries.append(entry)
1010
1011            elif protocol_player := self.mass.players.get_player(protocol.output_protocol_id):
1012                # we grab the config entries from the protocol player
1013                # and then prefix them to avoid key collisions
1014                protocol_entries = await self._get_player_config_entries(protocol_player)
1015                protocol_entry_keys = {entry.key for entry in protocol_entries}
1016                for proto_entry in protocol_entries:
1017                    # deep copy to avoid mutating shared/constant ConfigEntry objects
1018                    entry = deepcopy(proto_entry)
1019                    entry.category = protocol_category
1020                    entry.category_translation_key = category_translation_key
1021                    entry.category_translation_params = category_translation_params
1022                    # the key gets prefixed below to avoid collisions; pin the catalog key to the
1023                    # original (bare) slug and the protocol's own provider so the label still
1024                    # resolves against provider.<domain>.config_entries.<original_key>
1025                    entry.translation_key = entry.translation_key or entry.key
1026                    entry.translation_owner = protocol_player.translation_owner
1027                    entry.key = f"{protocol_prefix}{entry.key}"
1028                    if entry.depends_on in protocol_entry_keys:
1029                        # the entry it depends on is copied into this same block, so follow it
1030                        # to its prefixed key and keep the value condition that goes with it
1031                        entry.depends_on = f"{protocol_prefix}{entry.depends_on}"
1032                    else:
1033                        # nothing of its own to depend on, so gate it on the protocol toggle.
1034                        # any value condition belonged to the original key and must not carry
1035                        # over, or it gets compared against the toggle's boolean instead.
1036                        entry.depends_on = protocol_enabled_key
1037                        entry.depends_on_value = None
1038                        entry.depends_on_value_not = None
1039                    entry.action = f"{protocol_prefix}{entry.action}" if entry.action else None
1040                    all_entries.append(entry)
1041
1042        return all_entries
1043
1044    async def _update_output_protocol_config(
1045        self, values: dict[str, ConfigValueType]
1046    ) -> dict[str, ConfigValueType]:
1047        """
1048        Update output protocol related config for a player based on config values.
1049
1050        Returns updated values dict with output protocol related entries removed.
1051        """
1052        protocol_values: dict[str, dict[str, ConfigValueType]] = {}
1053        for key, value in list(values.items()):
1054            if CONF_PROTOCOL_KEY_SPLITTER not in key:
1055                continue
1056            # extract protocol player id and actual key
1057            protocol_player_id, actual_key = key.split(CONF_PROTOCOL_KEY_SPLITTER)
1058            if protocol_player_id not in protocol_values:
1059                protocol_values[protocol_player_id] = {}
1060            protocol_values[protocol_player_id][actual_key] = value
1061            # remove from main values dict
1062            del values[key]
1063        for protocol_player_id, proto_values in protocol_values.items():
1064            await self.save_player_config(protocol_player_id, proto_values)
1065            if proto_values.get(CONF_ENABLED):
1066                # wait max 10 seconds for protocol to become available
1067                for _ in range(10):
1068                    protocol_player = self.mass.players.get_player(protocol_player_id)
1069                    if protocol_player is not None:
1070                        break
1071                    await asyncio.sleep(1)
1072            # wait max 10 seconds for protocol
1073        return values
1074
1075    async def _get_output_protocol_config_values(
1076        self,
1077        entries: list[ConfigEntry],
1078    ) -> dict[str, ConfigValueType]:
1079        """Extract output protocol related config values for given (parent) player entries."""
1080        values: dict[str, ConfigValueType] = {}
1081        for entry in entries:
1082            if CONF_PROTOCOL_KEY_SPLITTER not in entry.key:
1083                continue
1084            protocol_player_id, actual_key = entry.key.split(CONF_PROTOCOL_KEY_SPLITTER)
1085            stored_value = self.get_raw_player_config_value(protocol_player_id, actual_key)
1086            if stored_value is None:
1087                continue
1088            values[entry.key] = stored_value
1089        return values
1090
1091    def _output_protocol_unavailable_reason(self, player: Player, protocol: OutputProtocol) -> str:
1092        """
1093        Return the translation slug telling why an output protocol can not be selected.
1094
1095        :param player: The player the output protocol belongs to.
1096        :param protocol: The output protocol that is currently unavailable.
1097        """
1098        if protocol.is_native:
1099            return "needs_setup" if player.needs_setup else "unavailable"
1100        if not self.get_raw_player_config_value(protocol.output_protocol_id, CONF_ENABLED, True):
1101            return "turned_off"
1102        protocol_player = self.mass.players.get_player(protocol.output_protocol_id)
1103        if protocol_player is not None and protocol_player.needs_setup:
1104            return "needs_setup"
1105        return "unavailable"
1106
1107    def _get_protocol_display_name(self, protocol_domain: str) -> str:
1108        """Return the display name for a protocol domain."""
1109        if provider_manifest := self.mass.get_provider_manifest(protocol_domain):
1110            return provider_manifest.name
1111        return protocol_domain.upper()
1112