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