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