music-assistant-server

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