/
/
/
1"""
2Controller for (provider specific) SyncGroup players.
3
4A SyncGroup player is a virtual player that automatically groups multiple players
5together in a sync group, where one player is the sync leader
6and the other players are synced to that leader.
7"""
8
9from __future__ import annotations
10
11import asyncio
12from copy import deepcopy
13from typing import TYPE_CHECKING, cast
14
15import shortuuid
16from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
17from music_assistant_models.constants import PLAYER_CONTROL_NONE
18from music_assistant_models.enums import (
19 ConfigEntryType,
20 PlaybackState,
21 PlayerFeature,
22 PlayerType,
23 ProviderFeature,
24)
25from music_assistant_models.errors import UnsupportedFeaturedException
26from music_assistant_models.player import DeviceInfo, PlayerMedia, PlayerSource
27from propcache import under_cached_property as cached_property
28
29from music_assistant.constants import (
30 CONF_CROSSFADE_DURATION,
31 CONF_DYNAMIC_GROUP_MEMBERS,
32 CONF_ENABLE_ICY_METADATA,
33 CONF_FLOW_MODE,
34 CONF_GROUP_MEMBERS,
35 CONF_HTTP_PROFILE,
36 CONF_OUTPUT_CODEC,
37 CONF_SAMPLE_RATES,
38 CONF_SMART_FADES_MODE,
39 SYNCGROUP_PREFIX,
40)
41from music_assistant.models.player import GroupPlayer, Player
42
43if TYPE_CHECKING:
44 from music_assistant.models.player_provider import PlayerProvider
45
46 from .player_controller import PlayerController
47
48
49SUPPORT_DYNAMIC_LEADER = {
50 # providers that support dynamic leader selection in a syncgroup
51 # meaning that if you would remove the current leader from the group,
52 # the provider will automatically select a new leader from the remaining members
53 # and the music keeps playing uninterrupted.
54 "airplay",
55 "squeezelite",
56 # TODO: Get this working with Sonos as well (need to handle range requests)
57}
58
59OPTIONAL_FEATURES = {
60 PlayerFeature.ENQUEUE,
61 PlayerFeature.GAPLESS_PLAYBACK,
62 PlayerFeature.NEXT_PREVIOUS,
63 PlayerFeature.PAUSE,
64 PlayerFeature.PLAY_ANNOUNCEMENT,
65 PlayerFeature.SEEK,
66 PlayerFeature.SELECT_SOURCE,
67 PlayerFeature.VOLUME_MUTE,
68 PlayerFeature.MULTI_DEVICE_DSP,
69}
70
71
72class SyncGroupPlayer(GroupPlayer):
73 """Helper class for a (provider specific) SyncGroup player."""
74
75 _attr_type: PlayerType = PlayerType.GROUP
76 sync_leader: Player | None = None
77 """The active sync leader player for this syncgroup."""
78
79 @cached_property
80 def is_dynamic(self) -> bool:
81 """Return if the player is a dynamic group player."""
82 return bool(self.config.get_value(CONF_DYNAMIC_GROUP_MEMBERS, False))
83
84 def __init__(
85 self,
86 provider: PlayerProvider,
87 player_id: str,
88 ) -> None:
89 """Initialize GroupPlayer instance."""
90 super().__init__(provider, player_id)
91 self._attr_name = self.config.name or self.config.default_name or f"SyncGroup {player_id}"
92 self._attr_available = True
93 self._attr_powered = False # group players are always powered off by default
94 self._attr_device_info = DeviceInfo(model="Sync Group", manufacturer=provider.name)
95 self._attr_supported_features = {
96 PlayerFeature.POWER,
97 PlayerFeature.VOLUME_SET,
98 }
99
100 async def on_config_updated(self) -> None:
101 """Handle logic when the player is loaded or updated."""
102 # Config is only available after the player was registered
103 static_members = cast("list[str]", self.config.get_value(CONF_GROUP_MEMBERS, []))
104 self._attr_static_group_members = static_members.copy()
105 if not self.powered:
106 self._attr_group_members = static_members.copy()
107 if self.is_dynamic:
108 self._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
109 else:
110 self._attr_supported_features.discard(PlayerFeature.SET_MEMBERS)
111
112 @property
113 def supported_features(self) -> set[PlayerFeature]:
114 """Return the supported features of the player."""
115 members = self.group_members
116 reference_player: Player | None = self.sync_leader or (
117 self.mass.players.get(members[0]) if members else None
118 )
119 if reference_player:
120 base_features = self._attr_supported_features.copy()
121 # add features supported by the sync leader
122 for feature in OPTIONAL_FEATURES:
123 if feature in reference_player.supported_features:
124 base_features.add(feature)
125 return base_features
126 return self._attr_supported_features
127
128 @property
129 def playback_state(self) -> PlaybackState:
130 """Return the current playback state of the player."""
131 if self.powered:
132 return self.sync_leader.playback_state if self.sync_leader else PlaybackState.IDLE
133 return PlaybackState.IDLE
134
135 @property
136 def requires_flow_mode(self) -> bool:
137 """Return if the player needs flow mode."""
138 if leader := self.sync_leader:
139 return leader.requires_flow_mode
140 return False
141
142 @property
143 def elapsed_time(self) -> float | None:
144 """Return the elapsed time in (fractional) seconds of the current track (if any)."""
145 return self.sync_leader.elapsed_time if self.sync_leader else None
146
147 @property
148 def elapsed_time_last_updated(self) -> float | None:
149 """Return when the elapsed time was last updated."""
150 return self.sync_leader.elapsed_time_last_updated if self.sync_leader else None
151
152 @property
153 def _current_media(self) -> PlayerMedia | None:
154 """Return the current media item (if any) loaded in the player."""
155 return self.sync_leader._current_media if self.sync_leader else self._attr_current_media
156
157 @property
158 def _active_source(self) -> str | None:
159 """Return the active source id (if any) of the player."""
160 return self.sync_leader._active_source if self.sync_leader else self._attr_active_source
161
162 @property
163 def _source_list(self) -> list[PlayerSource]:
164 """Return list of available (native) sources for this player."""
165 if self.sync_leader:
166 return self.sync_leader._source_list
167 return []
168
169 @property
170 def can_group_with(self) -> set[str]:
171 """
172 Return the id's of players this player can group with.
173
174 This should return set of player_id's this player can group/sync with
175 or just the provider's instance_id if all players can group with each other.
176 """
177 if self.is_dynamic and (leader := self.sync_leader):
178 return leader.can_group_with
179 if self.is_dynamic:
180 return {self.provider.instance_id}
181 return set()
182
183 async def get_config_entries(
184 self,
185 action: str | None = None,
186 values: dict[str, ConfigValueType] | None = None,
187 ) -> list[ConfigEntry]:
188 """Return all (provider/player specific) Config Entries for the given player (if any)."""
189 entries: list[ConfigEntry] = [
190 # syncgroup specific entries
191 ConfigEntry(
192 key=CONF_GROUP_MEMBERS,
193 type=ConfigEntryType.STRING,
194 multi_value=True,
195 label="Group members",
196 default_value=[],
197 description="Select all players you want to be part of this group",
198 required=False, # needed for dynamic members (which allows empty members list)
199 options=[
200 ConfigValueOption(x.display_name, x.player_id)
201 for x in self.provider.players
202 if x.type != PlayerType.GROUP
203 ],
204 ),
205 ConfigEntry(
206 key="dynamic_members",
207 type=ConfigEntryType.BOOLEAN,
208 label="Enable dynamic members",
209 description="Allow (un)joining members dynamically, so the group more or less "
210 "behaves the same like manually syncing players together, "
211 "with the main difference being that the group player will hold the queue.",
212 default_value=False,
213 required=False,
214 ),
215 ]
216 # combine base group entries with (base) player entries for this player type
217 child_player = next((x for x in self.provider.players if x.type == PlayerType.PLAYER), None)
218 if child_player:
219 allowed_conf_entries = (
220 CONF_HTTP_PROFILE,
221 CONF_ENABLE_ICY_METADATA,
222 CONF_CROSSFADE_DURATION,
223 CONF_OUTPUT_CODEC,
224 CONF_FLOW_MODE,
225 CONF_SAMPLE_RATES,
226 CONF_SMART_FADES_MODE,
227 )
228 child_config_entries = await child_player.get_config_entries()
229 entries.extend(
230 [entry for entry in child_config_entries if entry.key in allowed_conf_entries]
231 )
232 return entries
233
234 async def stop(self) -> None:
235 """Send STOP command to given player."""
236 if sync_leader := self.sync_leader:
237 await sync_leader.stop()
238
239 async def play(self) -> None:
240 """Send PLAY command to given player."""
241 if sync_leader := self.sync_leader:
242 await sync_leader.play()
243
244 async def pause(self) -> None:
245 """Send PAUSE command to given player."""
246 if sync_leader := self.sync_leader:
247 await sync_leader.pause()
248
249 async def power(self, powered: bool) -> None:
250 """Handle POWER command to group player."""
251 prev_power = self._attr_powered
252
253 # always stop at power off
254 if not powered and self.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
255 await self.stop()
256 self._attr_current_media = None
257
258 # optimistically set the group state
259 self._attr_powered = powered
260 if prev_power != powered:
261 self.update_state()
262
263 if powered:
264 # ensure static members are present when powering on
265 for static_group_member in self._attr_static_group_members:
266 member_player = self.mass.players.get(static_group_member)
267 if not member_player or not member_player.available or not member_player.enabled:
268 if static_group_member in self._attr_group_members:
269 self._attr_group_members.remove(static_group_member)
270 continue
271 if static_group_member not in self._attr_group_members:
272 # Always add static members when power(true) is called,
273 # this will ensure that static members that just became available are added
274 self._attr_group_members.append(static_group_member)
275 # Select sync leader and handle turn on
276 new_leader = self._select_sync_leader()
277 # handle TURN_ON of the group player by turning on all members
278 for member in self.mass.players.iter_group_members(
279 self, only_powered=False, active_only=False
280 ):
281 await self._handle_member_collisions(member)
282 if not member.powered and member.power_control != PLAYER_CONTROL_NONE:
283 await self.mass.players._handle_cmd_power(member.player_id, True)
284 # Set up the sync group with the new leader
285 if prev_power and new_leader == self.sync_leader:
286 # Already powered on with same leader, just re-sync members without full transition
287 await self._form_syncgroup()
288 else:
289 await self._handle_leader_transition(new_leader)
290 elif prev_power and not powered:
291 # handle TURN_OFF of the group player by dissolving group and turning off all members
292 await self._dissolve_syncgroup()
293 # turn off all group members
294 for member in self.mass.players.iter_group_members(
295 self, only_powered=True, active_only=True
296 ):
297 if member.powered and member.power_control != PLAYER_CONTROL_NONE:
298 await self.mass.players._handle_cmd_power(member.player_id, False)
299
300 if not powered:
301 # Reset to unfiltered static members list when powered off
302 # (the frontend will hide unavailable members)
303 self._attr_group_members = self._attr_static_group_members.copy()
304 # and clear the sync leader
305 self.sync_leader = None
306 self.update_state()
307
308 async def volume_set(self, volume_level: int) -> None:
309 """Send VOLUME_SET command to given player."""
310 # group volume is already handled in the player manager
311
312 async def play_media(self, media: PlayerMedia) -> None:
313 """Handle PLAY MEDIA on given player."""
314 # power on (which will also resync and add static members if needed)
315 await self.power(True)
316 # simply forward the command to the sync leader
317 if sync_leader := self.sync_leader:
318 await sync_leader.play_media(media)
319 self._attr_current_media = deepcopy(media)
320 self.update_state()
321 else:
322 raise RuntimeError("an empty group cannot play media, consider adding members first")
323
324 async def enqueue_next_media(self, media: PlayerMedia) -> None:
325 """Handle enqueuing of a next media item on the player."""
326 if sync_leader := self.sync_leader:
327 await sync_leader.enqueue_next_media(media)
328
329 async def select_source(self, source: str) -> None:
330 """
331 Handle SELECT SOURCE command on the player.
332
333 Will only be called if the PlayerFeature.SELECT_SOURCE is supported.
334
335 :param source: The source(id) to select, as defined in the source_list.
336 """
337 if sync_leader := self.sync_leader:
338 await sync_leader.select_source(source)
339 self.update_state()
340
341 async def set_members(
342 self,
343 player_ids_to_add: list[str] | None = None,
344 player_ids_to_remove: list[str] | None = None,
345 ) -> None:
346 """Handle SET_MEMBERS command on the player."""
347 if not self.is_dynamic:
348 raise UnsupportedFeaturedException(
349 f"Group {self.display_name} does not allow dynamically adding/removing members!"
350 )
351 # handle additions
352 final_players_to_add: list[str] = []
353 for player_id in player_ids_to_add or []:
354 if player_id in self._attr_group_members:
355 continue
356 if player_id == self.player_id:
357 raise UnsupportedFeaturedException(
358 f"Cannot add {self.display_name} to itself as a member!"
359 )
360 self._attr_group_members.append(player_id)
361 final_players_to_add.append(player_id)
362 # handle removals
363 final_players_to_remove: list[str] = []
364 for player_id in player_ids_to_remove or []:
365 if player_id not in self._attr_group_members:
366 continue
367 if player_id == self.player_id:
368 raise UnsupportedFeaturedException(
369 f"Cannot remove {self.display_name} from itself as a member!"
370 )
371 self._attr_group_members.remove(player_id)
372 final_players_to_remove.append(player_id)
373 self.update_state()
374 if not self.powered:
375 # Don't need to do anything else if the group is powered off
376 # The syncing will be done once powered on
377 return
378 next_leader = self._select_sync_leader()
379 prev_leader = self.sync_leader
380
381 if prev_leader and next_leader is None:
382 # Edge case: we no longer have any members in the group (and thus no leader)
383 await self._handle_leader_transition(None)
384 elif prev_leader != next_leader:
385 # Edge case: we had changed the leader (or just got one)
386 await self._handle_leader_transition(next_leader)
387 elif self.sync_leader and (player_ids_to_add or player_ids_to_remove):
388 # if the group still has the same leader, we need to (re)sync the members
389 # Handle collisions for newly added players
390 for player_id in final_players_to_add:
391 if player := self.mass.players.get(player_id):
392 await self._handle_member_collisions(player)
393
394 await self.sync_leader.set_members(
395 player_ids_to_add=final_players_to_add,
396 player_ids_to_remove=final_players_to_remove,
397 )
398
399 async def _form_syncgroup(self) -> None:
400 """Form syncgroup by syncing all (possible) members."""
401 if self.sync_leader is None:
402 # This is an empty group, leader will be selected once a member is added
403 self._attr_group_members = []
404 self.update_state()
405 return
406 # ensure the sync leader is first in the list
407 self._attr_group_members = [
408 self.sync_leader.player_id,
409 *[x for x in self._attr_group_members if x != self.sync_leader.player_id],
410 ]
411 self.update_state()
412 members_to_sync: list[str] = []
413 members_to_remove: list[str] = []
414 for member in self.mass.players.iter_group_members(self, active_only=False):
415 # Handle collisions before attempting to sync
416 await self._handle_member_collisions(member)
417
418 if member.synced_to and member.synced_to != self.sync_leader.player_id:
419 # ungroup first
420 await member.ungroup()
421 if member.player_id == self.sync_leader.player_id:
422 # skip sync leader
423 continue
424 # Always add to members_to_sync to prevent them from being removed below
425 members_to_sync.append(member.player_id)
426 for former_members in self.sync_leader.group_members:
427 if (
428 former_members not in members_to_sync
429 ) and former_members != self.sync_leader.player_id:
430 members_to_remove.append(former_members)
431 if members_to_sync or members_to_remove:
432 await self.sync_leader.set_members(members_to_sync, members_to_remove)
433
434 async def _dissolve_syncgroup(self) -> None:
435 """Dissolve the current syncgroup by ungrouping all members and restoring leader queue."""
436 if sync_leader := self.sync_leader:
437 # dissolve the temporary syncgroup from the sync leader
438 sync_children = [x for x in sync_leader.group_members if x != sync_leader.player_id]
439 if sync_children:
440 await sync_leader.set_members(player_ids_to_remove=sync_children)
441 # Reset the leaders queue since it is no longer part of this group
442 sync_leader.update_state()
443
444 async def _handle_leader_transition(self, new_leader: Player | None) -> None:
445 """Handle transition from current leader to new leader."""
446 prev_leader = self.sync_leader
447 was_playing = False
448
449 if (
450 prev_leader
451 and new_leader
452 and prev_leader != new_leader
453 and self.provider.domain in SUPPORT_DYNAMIC_LEADER
454 ):
455 # provider supports dynamic leader selection, so just remove/add members
456 await prev_leader.ungroup()
457 self.sync_leader = new_leader
458 # allow some time to propagate the changes before resyncing
459 await asyncio.sleep(2)
460 await self._form_syncgroup()
461 return
462
463 if prev_leader:
464 # Save current media and playback state for potential restart
465 was_playing = self.playback_state == PlaybackState.PLAYING
466 # Stop current playback and dissolve existing group
467 await self.stop()
468 await self._dissolve_syncgroup()
469 # allow some time to propagate the changes before resyncing
470 await asyncio.sleep(2)
471
472 # Set new leader
473 self.sync_leader = new_leader
474
475 if new_leader:
476 # form a syncgroup with the new leader
477 await self._form_syncgroup()
478
479 # Restart playback if requested and we have media to play
480 if was_playing:
481 await self.mass.players._handle_cmd_resume(self.player_id)
482 else:
483 # We have no leader anymore, send update since we stopped playback
484 self.update_state()
485
486 def _select_sync_leader(self) -> Player | None:
487 """Select the active sync leader player for a syncgroup."""
488 if self.sync_leader and self.sync_leader.player_id in self.group_members:
489 # Don't change the sync leader if we already have one
490 return self.sync_leader
491 for prefer_sync_leader in (True, False):
492 for child_player in self.mass.players.iter_group_members(self):
493 if prefer_sync_leader and child_player.synced_to:
494 # prefer the first player that already has sync children
495 continue
496 if child_player.active_group not in (
497 None,
498 self.player_id,
499 child_player.player_id,
500 ):
501 # this should not happen (because its already handled in the power on logic),
502 # but guard it just in case bad things happen
503 continue
504 return child_player
505 return None
506
507 async def _handle_member_collisions(self, member: Player) -> None:
508 """Handle collisions when adding a member to the sync group."""
509 active_groups = member.active_groups
510 for group in active_groups:
511 if group == self.player_id:
512 continue
513 # collision: child player is part another group that is already active !
514 # solve this by trying to leave the group first
515 if other_group := self.mass.players.get(group):
516 if (
517 other_group.supports_feature(PlayerFeature.SET_MEMBERS)
518 and member.player_id not in other_group.static_group_members
519 ):
520 await other_group.set_members(player_ids_to_remove=[member.player_id])
521 else:
522 # if the other group does not support SET_MEMBERS or it is a static
523 # member, we need to power it off to leave the group
524 await other_group.power(False)
525 if (
526 member.synced_to is not None
527 and self.sync_leader
528 and member.synced_to != self.sync_leader.player_id
529 and (synced_to_player := self.mass.players.get(member.synced_to))
530 and member.player_id in synced_to_player.group_members
531 ):
532 # collision: child player is synced to another player and still in that group
533 # ungroup it first
534 await synced_to_player.set_members(player_ids_to_remove=[member.player_id])
535
536
537class SyncGroupController:
538 """Controller managing SyncGroup players."""
539
540 def __init__(self, player_controller: PlayerController) -> None:
541 """Initialize SyncGroupController."""
542 self.player_controller = player_controller
543 self.mass = player_controller.mass
544
545 async def create_group_player(
546 self, provider: PlayerProvider, name: str, members: list[str], dynamic: bool = True
547 ) -> Player:
548 """
549 Create new SyncGroup Player.
550
551 :param provider: The provider to create the group player for
552 :param name: Name of the group player
553 :param members: List of player ids to add to the group
554 :param dynamic: Whether the group is dynamic (members can change)
555 """
556 # default implementation for providers that support syncing players
557 if ProviderFeature.SYNC_PLAYERS not in provider.supported_features:
558 # the frontend should already prevent this, but just in case
559 raise UnsupportedFeaturedException(
560 f"Provider {provider.name} does not support player syncing!"
561 )
562 # Create a new syncgroup player with the given members
563 members = [x for x in members if x in [y.player_id for y in provider.players]]
564 player_id = f"{SYNCGROUP_PREFIX}{shortuuid.random(8).lower()}"
565 self.mass.config.create_default_player_config(
566 player_id=player_id,
567 provider=provider.instance_id,
568 name=name,
569 enabled=True,
570 values={
571 CONF_GROUP_MEMBERS: members,
572 CONF_DYNAMIC_GROUP_MEMBERS: dynamic,
573 },
574 )
575 return await self._register_syncgroup_player(player_id, provider)
576
577 async def remove_group_player(self, player_id: str) -> None:
578 """
579 Remove a group player.
580
581 :param player_id: ID of the group player to remove.
582 """
583 # we simply permanently unregister the syncgroup player and wipe its config
584 await self.mass.players.unregister(player_id, True)
585
586 async def _register_syncgroup_player(self, player_id: str, provider: PlayerProvider) -> Player:
587 """Register a syncgroup player."""
588 syncgroup = SyncGroupPlayer(provider, player_id)
589 await self.mass.players.register_or_update(syncgroup)
590 return syncgroup
591
592 async def on_provider_loaded(self, provider: PlayerProvider) -> None:
593 """Handle logic when a provider is loaded."""
594 # register existing syncgroup players for this provider
595 for player_conf in await self.mass.config.get_player_configs(provider.instance_id):
596 if player_conf.player_id.startswith(SYNCGROUP_PREFIX):
597 await self._register_syncgroup_player(player_conf.player_id, provider)
598
599 async def on_provider_unload(self, provider: PlayerProvider) -> None:
600 """Handle logic when a provider is (about to get) unloaded."""
601 # unregister existing syncgroup players for this provider
602 for player in self.mass.players.all(
603 provider_filter=provider.instance_id, return_sync_groups=True
604 ):
605 if player.player_id.startswith(SYNCGROUP_PREFIX):
606 await self.mass.players.unregister(player.player_id, False)
607