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        if PlayerFeature.GAPLESS_PLAYBACK in player.supported_features:
673            default_entries.append(CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES)
674        # request player specific entries
675        player_entries = await player.get_config_entries()
676        players_keys = {entry.key for entry in player_entries}
677        # filter out any default entries that are already provided by the player
678        default_entries = [entry for entry in default_entries if entry.key not in players_keys]
679        return [*player_entries, *default_entries]
680
681    def _get_default_player_config_entries(self, player: Player) -> list[ConfigEntry]:
682        """
683        Return the default (generic) player config entries.
684
685        This does not return audio/protocol specific entries, those are handled elsewhere.
686        """
687        entries: list[ConfigEntry] = []
688        # default protocol-player config entries
689        if player.state.type == PlayerType.PROTOCOL:
690            # protocol players have no generic config entries
691            # only audio/protocol specific ones
692            return []
693
694        icon_entry = deepcopy(
695            CONF_ENTRY_PLAYER_ICON_GROUP
696            if player.state.type == PlayerType.GROUP
697            else CONF_ENTRY_PLAYER_ICON
698        )
699        icon_entry.default_value = player.default_icon
700        icon_entry.value = self.get_raw_player_config_value(player.player_id, CONF_ICON)
701
702        # some base entries for all player types
703        # note that these may NOT be playback/audio related
704        entries += [
705            CONF_ENTRY_TTS_PRE_ANNOUNCE,
706            ConfigEntry(
707                key=CONF_PRE_ANNOUNCE_CHIME_URL,
708                type=ConfigEntryType.STRING,
709                category="announcements",
710                required=False,
711                depends_on=CONF_ENTRY_TTS_PRE_ANNOUNCE.key,
712                depends_on_value=True,
713                validate=lambda val: validate_announcement_chime_url(cast("str", val)),
714            ),
715            # add player control entries
716            *self._create_player_control_config_entries(player),
717            # add entry to hide player in UI
718            ConfigEntry(
719                key=CONF_HIDE_IN_UI,
720                type=ConfigEntryType.BOOLEAN,
721                default_value=player.hidden_by_default,
722                category="generic",
723                advanced=False,
724            ),
725            # add entry to expose player to HA
726            ConfigEntry(
727                key=CONF_EXPOSE_PLAYER_TO_HA,
728                type=ConfigEntryType.BOOLEAN,
729                category="generic",
730                advanced=False,
731                default_value=player.expose_to_ha_by_default,
732            ),
733        ]
734        # group-player config entries
735        if player.state.type == PlayerType.GROUP:
736            entries += [
737                icon_entry,
738            ]
739            return entries
740        # normal player (or stereo pair) config entries
741        entries += [
742            icon_entry,
743            # add default entries for announce feature
744            CONF_ENTRY_ANNOUNCE_VOLUME_STRATEGY,
745            CONF_ENTRY_ANNOUNCE_VOLUME,
746            CONF_ENTRY_ANNOUNCE_VOLUME_MIN,
747            CONF_ENTRY_ANNOUNCE_VOLUME_MAX,
748            # play_media-on-self preference (only relevant to non-group players)
749            CONF_ENTRY_PLAY_MEDIA_OVERRIDES_GROUP,
750        ]
751        return entries
752
753    def _create_player_control_config_entries(self, player: Player) -> list[ConfigEntry]:
754        """Create config entries for player controls."""
755        is_group = player.state.type == PlayerType.GROUP
756        all_controls = self.mass.players.player_controls()
757        power_controls = [x for x in all_controls if x.supports_power]
758        volume_controls = [x for x in all_controls if x.supports_volume]
759        mute_controls = [x for x in all_controls if x.supports_mute]
760        auto_option = ConfigValueOption(PLAYER_CONTROL_PROTOCOL)
761        # the "native" option is always listed (disabled when the feature is unsupported) so the
762        # option set is consistent across players; the entry default skips disabled options.
763        power_options: list[ConfigValueOption] = [
764            ConfigValueOption(
765                PLAYER_CONTROL_NATIVE,
766                disabled=not player.supports_feature(PlayerFeature.POWER),
767            )
768        ]
769        has_native_volume_control = player.supports_feature(PlayerFeature.VOLUME_SET)
770        volume_options: list[ConfigValueOption] = [
771            ConfigValueOption(PLAYER_CONTROL_NATIVE, disabled=not has_native_volume_control)
772        ]
773        mute_options: list[ConfigValueOption] = [
774            ConfigValueOption(
775                PLAYER_CONTROL_NATIVE,
776                disabled=not player.supports_feature(PlayerFeature.VOLUME_MUTE),
777            )
778        ]
779        # add player protocols as volume controls if native player has no volume control
780        for linked_protocol in player.linked_output_protocols:
781            if has_native_volume_control:
782                break
783            protocol_player = self.mass.players.get_player(linked_protocol.output_protocol_id)
784            if not protocol_player or not protocol_player.available_for_playback:
785                continue
786            if protocol_player.supports_feature(PlayerFeature.VOLUME_SET):
787                if auto_option not in volume_options:
788                    volume_options.append(auto_option)
789                if linked_protocol.protocol_domain in ("chromecast", "dlna"):
790                    # for chromecast/dlna we can use the protocol player for volume control
791                    # even if the protocol player is not the active protocol
792                    volume_options.append(
793                        ConfigValueOption(
794                            protocol_player.player_id, title=protocol_player.provider.name
795                        )
796                    )
797            if protocol_player.supports_feature(PlayerFeature.VOLUME_MUTE):
798                if auto_option not in mute_options:
799                    mute_options.append(auto_option)
800                if linked_protocol.protocol_domain in ("chromecast", "dlna"):
801                    # for chromecast/dlna we can use the protocol player for volume control
802                    # even if the protocol player is not the active protocol
803                    mute_options.append(
804                        ConfigValueOption(
805                            protocol_player.player_id, title=protocol_player.provider.name
806                        )
807                    )
808
809        # append none+fake options
810        power_options += [
811            ConfigValueOption(PLAYER_CONTROL_NONE),
812            ConfigValueOption(PLAYER_CONTROL_FAKE),
813        ]
814        volume_options += [
815            ConfigValueOption(PLAYER_CONTROL_NONE),
816        ]
817        mute_options.append(ConfigValueOption(PLAYER_CONTROL_NONE))
818        # fake mute drives the volume control, so offer it when the player has any
819        # usable volume path (native or via a linked protocol player)
820        if player.supports_feature(PlayerFeature.VOLUME_SET) or auto_option in volume_options:
821            mute_options.append(ConfigValueOption(PLAYER_CONTROL_FAKE))
822
823        # return final config entries for all options
824        return [
825            # Power control config entry
826            ConfigEntry(
827                key=CONF_POWER_CONTROL,
828                type=ConfigEntryType.STRING,
829                default_value=_first_enabled_control_value(power_options),
830                required=False,
831                options=[
832                    *power_options,
833                    *(ConfigValueOption(x.id, title=x.name) for x in power_controls),
834                ],
835                category="player_controls",
836            ),
837            # Volume control config entry
838            ConfigEntry(
839                key=CONF_VOLUME_CONTROL,
840                type=ConfigEntryType.STRING,
841                default_value=_first_enabled_control_value(volume_options),
842                required=True,
843                options=[
844                    *volume_options,
845                    *(ConfigValueOption(x.id, title=x.name) for x in volume_controls),
846                ],
847                category="player_controls",
848            ),
849            # Mute control config entry
850            ConfigEntry(
851                key=CONF_MUTE_CONTROL,
852                type=ConfigEntryType.STRING,
853                default_value=_first_enabled_control_value(mute_options),
854                required=True,
855                options=[
856                    *mute_options,
857                    *[ConfigValueOption(x.id, title=x.name) for x in mute_controls],
858                ],
859                category="player_controls",
860            ),
861            # Volume limit entries
862            CONF_ENTRY_MIN_VOLUME,
863            CONF_ENTRY_MAX_VOLUME,
864            # auto-play on power on — only meaningful for individual players.
865            # For group players, power on/off is purely a "capture members"
866            # toggle (Fake control) and auto-starting playback there causes
867            # surprise playback when the user just wanted to pin the group.
868            *([] if is_group else [CONF_ENTRY_AUTO_PLAY]),
869        ]
870
871    async def _create_output_protocol_config_entries(  # noqa: PLR0915
872        self,
873        player: Player,
874    ) -> list[ConfigEntry]:
875        """
876        Create the output protocol config entries for a player.
877
878        The preferred output protocol entry is always returned, listing outputs that can not
879        be selected right now as disabled options and hidden altogether when the player has
880        at most one output. The settings of each output whose provider is loaded follow.
881
882        :param player: The player to create the output protocol config entries for.
883        """
884        all_entries: list[ConfigEntry] = []
885        output_protocols = player.output_protocols
886
887        # Resolve derived-transport edges (e.g. a Sendspin bridge riding on the
888        # AirPlay protocol) so derived protocols render as dependent on their base.
889        base_protocols: dict[str, OutputProtocol] = {}
890        for protocol in output_protocols:
891            if protocol.is_native:
892                continue
893            underlying_id = self.get_raw_player_config_value(
894                protocol.output_protocol_id, CONF_UNDERLYING_PLAYER_ID
895            )
896            if not underlying_id:
897                continue
898            if base_protocol := next(
899                (p for p in output_protocols if p.output_protocol_id == underlying_id), None
900            ):
901                base_protocols[protocol.output_protocol_id] = base_protocol
902
903        # Build options from all output protocols, sorted by priority
904        options: list[ConfigValueOption] = []
905
906        # Add each output protocol as an option, sorted by priority. An output that can not be
907        # used right now is offered disabled with the reason why, rather than left out entirely:
908        # that keeps the device's outputs recognizable and explains what to do about it.
909        has_native = False
910        for protocol in sorted(output_protocols, key=lambda p: p.priority):
911            protocol_name = self._get_protocol_display_name(protocol.protocol_domain)
912            # Use "native" for native playback,
913            # otherwise use the protocol output id (=player id)
914            if protocol.is_native:
915                title = f"{protocol_name} (native)"
916            elif base_protocol := base_protocols.get(protocol.output_protocol_id):
917                title = (
918                    f"{protocol_name} "
919                    f"(over {self._get_protocol_display_name(base_protocol.protocol_domain)})"
920                )
921            else:
922                title = protocol_name
923            value = "native" if protocol.is_native else protocol.output_protocol_id
924            options.append(
925                ConfigValueOption(
926                    value,
927                    title=title,
928                    disabled=not protocol.available,
929                    # the option's value is a player id, so its reason is keyed by this slug
930                    translation_key=(
931                        None
932                        if protocol.available
933                        else self._output_protocol_unavailable_reason(player, protocol)
934                    ),
935                )
936            )
937            # never default to an output that can not be selected
938            has_native = has_native or (protocol.is_native and protocol.available)
939
940        if has_native:
941            default_value = "native"
942        else:
943            # Without a native output the entry default stays "auto": runtime selection
944            # honours the player's default_output_protocol_domain (e.g. DLNA-first for a
945            # LinkPlay shell) with plain priority fallback, so the stored config default
946            # must not depend on which linked protocols happen to be available right now.
947            options.append(ConfigValueOption("auto"))
948            default_value = "auto"
949
950        all_entries.append(
951            ConfigEntry(
952                key=CONF_PREFERRED_OUTPUT_PROTOCOL,
953                type=ConfigEntryType.STRING,
954                default_value=default_value,
955                required=True,
956                options=options,
957                category="protocol_general",
958                requires_reload=False,
959                hidden=len(output_protocols) <= 1,
960            )
961        )
962
963        # Add config entries for all protocol players/outputs
964        for protocol in output_protocols:
965            domain = protocol.protocol_domain
966            protocol_name = self._get_protocol_display_name(domain)
967            protocol_player_enabled = self.get_raw_player_config_value(
968                protocol.output_protocol_id, CONF_ENABLED, True
969            )
970            provider_available = self.mass.get_provider(protocol.protocol_domain) is not None
971            if not provider_available:
972                # protocol provider is not available, skip adding entries
973                continue
974            protocol_prefix = f"{protocol.output_protocol_id}{CONF_PROTOCOL_KEY_SPLITTER}"
975            protocol_enabled_key = f"{protocol_prefix}enabled"
976            protocol_category = f"{CONF_PROTOCOL_CATEGORY_PREFIX}_{domain}"
977            category_translation_key = "protocol_output_settings"
978            category_translation_params = [protocol_name]
979            # Derived protocols (e.g. Sendspin over AirPlay) render with their base
980            # protocol in the label and follow its enabled state.
981            base_protocol = base_protocols.get(protocol.output_protocol_id)
982            base_name: str | None = None
983            base_enabled = True
984            if base_protocol is not None:
985                base_name = self._get_protocol_display_name(base_protocol.protocol_domain)
986                base_raw_conf = self.get(f"{CONF_PLAYERS}/{base_protocol.output_protocol_id}") or {}
987                base_enabled = bool(base_raw_conf.get("enabled", True))
988                category_translation_key = "protocol_output_settings_via"
989                category_translation_params = [protocol_name, base_name]
990            if not protocol.is_native:
991                all_entries.append(
992                    ConfigEntry(
993                        key=protocol_enabled_key,
994                        type=ConfigEntryType.BOOLEAN,
995                        # the key is per-protocol (dynamic), so pin a static catalog key
996                        translation_key="protocol_enable_via" if base_name else "protocol_enable",
997                        translation_params=[base_name] if base_name else None,
998                        # a derived protocol cannot be active while its base is disabled
999                        value=bool(protocol_player_enabled) and base_enabled,
1000                        read_only=not base_enabled,
1001                        default_value=True,
1002                        category=protocol_category,
1003                        category_translation_key=category_translation_key,
1004                        category_translation_params=category_translation_params,
1005                        requires_reload=False,
1006                    )
1007                )
1008            if protocol.is_native:
1009                # add protocol-specific entries from native player
1010                protocol_entries = await self._get_player_config_entries(player)
1011                for proto_entry in protocol_entries:
1012                    # deep copy to avoid mutating shared/constant ConfigEntry objects
1013                    entry = deepcopy(proto_entry)
1014                    entry.category = protocol_category
1015                    entry.category_translation_key = category_translation_key
1016                    entry.category_translation_params = category_translation_params
1017                    all_entries.append(entry)
1018
1019            elif protocol_player := self.mass.players.get_player(protocol.output_protocol_id):
1020                # we grab the config entries from the protocol player
1021                # and then prefix them to avoid key collisions
1022                protocol_entries = await self._get_player_config_entries(protocol_player)
1023                protocol_entry_keys = {entry.key for entry in protocol_entries}
1024                for proto_entry in protocol_entries:
1025                    # deep copy to avoid mutating shared/constant ConfigEntry objects
1026                    entry = deepcopy(proto_entry)
1027                    entry.category = protocol_category
1028                    entry.category_translation_key = category_translation_key
1029                    entry.category_translation_params = category_translation_params
1030                    # the key gets prefixed below to avoid collisions; pin the catalog key to the
1031                    # original (bare) slug and the protocol's own provider so the label still
1032                    # resolves against provider.<domain>.config_entries.<original_key>
1033                    entry.translation_key = entry.translation_key or entry.key
1034                    entry.translation_owner = protocol_player.translation_owner
1035                    entry.key = f"{protocol_prefix}{entry.key}"
1036                    if entry.depends_on in protocol_entry_keys:
1037                        # the entry it depends on is copied into this same block, so follow it
1038                        # to its prefixed key and keep the value condition that goes with it
1039                        entry.depends_on = f"{protocol_prefix}{entry.depends_on}"
1040                    else:
1041                        # nothing of its own to depend on, so gate it on the protocol toggle.
1042                        # any value condition belonged to the original key and must not carry
1043                        # over, or it gets compared against the toggle's boolean instead.
1044                        entry.depends_on = protocol_enabled_key
1045                        entry.depends_on_value = None
1046                        entry.depends_on_value_not = None
1047                    entry.action = f"{protocol_prefix}{entry.action}" if entry.action else None
1048                    all_entries.append(entry)
1049
1050        return all_entries
1051
1052    async def _update_output_protocol_config(
1053        self, values: dict[str, ConfigValueType]
1054    ) -> dict[str, ConfigValueType]:
1055        """
1056        Update output protocol related config for a player based on config values.
1057
1058        Returns updated values dict with output protocol related entries removed.
1059        """
1060        protocol_values: dict[str, dict[str, ConfigValueType]] = {}
1061        for key, value in list(values.items()):
1062            if CONF_PROTOCOL_KEY_SPLITTER not in key:
1063                continue
1064            # extract protocol player id and actual key
1065            protocol_player_id, actual_key = key.split(CONF_PROTOCOL_KEY_SPLITTER)
1066            if protocol_player_id not in protocol_values:
1067                protocol_values[protocol_player_id] = {}
1068            protocol_values[protocol_player_id][actual_key] = value
1069            # remove from main values dict
1070            del values[key]
1071        for protocol_player_id, proto_values in protocol_values.items():
1072            await self.save_player_config(protocol_player_id, proto_values)
1073            if proto_values.get(CONF_ENABLED):
1074                # wait max 10 seconds for protocol to become available
1075                for _ in range(10):
1076                    protocol_player = self.mass.players.get_player(protocol_player_id)
1077                    if protocol_player is not None:
1078                        break
1079                    await asyncio.sleep(1)
1080            # wait max 10 seconds for protocol
1081        return values
1082
1083    async def _get_output_protocol_config_values(
1084        self,
1085        entries: list[ConfigEntry],
1086    ) -> dict[str, ConfigValueType]:
1087        """Extract output protocol related config values for given (parent) player entries."""
1088        values: dict[str, ConfigValueType] = {}
1089        for entry in entries:
1090            if CONF_PROTOCOL_KEY_SPLITTER not in entry.key:
1091                continue
1092            protocol_player_id, actual_key = entry.key.split(CONF_PROTOCOL_KEY_SPLITTER)
1093            stored_value = self.get_raw_player_config_value(protocol_player_id, actual_key)
1094            if stored_value is None:
1095                continue
1096            values[entry.key] = stored_value
1097        return values
1098
1099    def _output_protocol_unavailable_reason(self, player: Player, protocol: OutputProtocol) -> str:
1100        """
1101        Return the translation slug telling why an output protocol can not be selected.
1102
1103        :param player: The player the output protocol belongs to.
1104        :param protocol: The output protocol that is currently unavailable.
1105        """
1106        if protocol.is_native:
1107            return "needs_setup" if player.needs_setup else "unavailable"
1108        if not self.get_raw_player_config_value(protocol.output_protocol_id, CONF_ENABLED, True):
1109            return "turned_off"
1110        protocol_player = self.mass.players.get_player(protocol.output_protocol_id)
1111        if protocol_player is not None and protocol_player.needs_setup:
1112            return "needs_setup"
1113        return "unavailable"
1114
1115    def _get_protocol_display_name(self, protocol_domain: str) -> str:
1116        """Return the display name for a protocol domain."""
1117        if provider_manifest := self.mass.get_provider_manifest(protocol_domain):
1118            return provider_manifest.name
1119        return protocol_domain.upper()
1120