music-assistant-server

1"""Universal Player Group Provider implementation."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7import shortuuid
8from music_assistant_models.enums import PlayerType
9
10from music_assistant.constants import CONF_DYNAMIC_GROUP_MEMBERS, CONF_GROUP_MEMBERS
11from music_assistant.models.player_provider import PlayerProvider
12
13from .constants import UGP_PREFIX
14from .player import UniversalGroupPlayer
15
16if TYPE_CHECKING:
17    from music_assistant_models.config_entries import ConfigEntry
18
19    from music_assistant.models.player import Player
20
21
22class UniversalGroupProvider(PlayerProvider):
23    """Universal Group Player Provider."""
24
25    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
26        """Return Config entries to setup this provider."""
27        # nothing to configure (for now)
28        return ()
29
30    async def create_group_player(
31        self, name: str, members: list[str], dynamic: bool = True
32    ) -> Player:
33        """Create new Universal Group Player."""
34        # filter out members that are not registered players
35        # TODO: do we want to filter out groups here to prevent nested groups?
36        members = [x for x in members if x in [y.player_id for y in self.mass.players]]
37        # generate a new player_id for the group player
38        player_id = f"{UGP_PREFIX}{shortuuid.random(8).lower()}"
39        self.mass.config.create_default_player_config(
40            player_id=player_id,
41            provider=self.instance_id,
42            player_type=PlayerType.GROUP,
43            name=name,
44            enabled=True,
45            values={
46                CONF_GROUP_MEMBERS: members,
47                CONF_DYNAMIC_GROUP_MEMBERS: dynamic,
48            },
49        )
50        return await self._register_player(player_id)
51
52    async def remove_group_player(self, player_id: str) -> None:
53        """
54        Remove a group player.
55
56        Only called for providers that support REMOVE_GROUP_PLAYER feature.
57
58        :param player_id: ID of the group player to remove.
59        """
60        # we simply permanently unregister the player and wipe its config
61        await self.mass.players.unregister(player_id, True)
62
63    async def discover_players(self) -> None:
64        """Discover players."""
65        for player_conf in await self.mass.config.get_player_configs(self.instance_id):
66            if player_conf.player_id.startswith(UGP_PREFIX):
67                await self._register_player(player_conf.player_id)
68
69    async def _register_player(self, player_id: str) -> Player:
70        """Register a universal group player."""
71        group = UniversalGroupPlayer(self, player_id)
72        await self.mass.players.register_or_update(group)
73        return group
74