music-assistant-server

1"""Sync Group Player 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 SGP_PREFIX
14from .player import SyncGroupPlayer
15
16if TYPE_CHECKING:
17    from music_assistant.models.player import Player
18
19
20class SyncGroupProvider(PlayerProvider):
21    """Sync Group Player Provider."""
22
23    async def create_group_player(
24        self, name: str, members: list[str], dynamic: bool = True
25    ) -> Player:
26        """
27        Create new Sync Group Player.
28
29        :param name: Name of the group player.
30        :param members: List of player ids to add to the group.
31        :param dynamic: Whether the group is dynamic (members can change).
32        """
33        # validation to ensure all members are compatible (can_group_with check)
34        members = [x for x in members if x in [y.player_id for y in self.mass.players]]
35        final_members: list[str] = []
36        can_group_with: set[str] = set()
37        for member_id in members:
38            member = self.mass.players.get_player(member_id)
39            if member is None or not member.available:
40                continue
41            if not can_group_with:
42                # first member, add all its compatible players to the can_group_with set
43                can_group_with = set(member.state.can_group_with)
44            elif member_id not in can_group_with:
45                # member is not compatible with the current group, skip it
46                continue
47            final_members.append(member_id)
48        # generate a new player_id for the group player
49        player_id = f"{SGP_PREFIX}{shortuuid.random(8).lower()}"
50        self.mass.config.create_default_player_config(
51            player_id=player_id,
52            provider=self.instance_id,
53            player_type=PlayerType.GROUP,
54            name=name,
55            enabled=True,
56            values={
57                CONF_GROUP_MEMBERS: final_members,
58                CONF_DYNAMIC_GROUP_MEMBERS: dynamic,
59            },
60        )
61        return await self._register_player(player_id)
62
63    async def remove_group_player(self, player_id: str) -> None:
64        """
65        Remove a group player.
66
67        :param player_id: ID of the group player to remove.
68        """
69        # we simply permanently unregister the player and wipe its config
70        await self.mass.players.unregister(player_id, True)
71
72    async def discover_players(self) -> None:
73        """Discover players."""
74        for player_conf in await self.mass.config.get_player_configs(self.instance_id):
75            if player_conf.player_id.startswith(SGP_PREFIX):
76                await self._register_player(player_conf.player_id)
77
78    async def _register_player(self, player_id: str) -> Player:
79        """Register a sync group player."""
80        group = SyncGroupPlayer(self, player_id)
81        await self.mass.players.register_or_update(group)
82        return group
83