/
/
/
1"""Native multiroom topology coordinator shared by both WiiM backends."""
2
3from __future__ import annotations
4
5import asyncio
6import time
7from contextlib import AsyncExitStack
8from enum import StrEnum
9from typing import TYPE_CHECKING, Any, cast
10
11from music_assistant_models.errors import PlayerCommandFailed
12from pywiim import WiiMError
13from wiim.exceptions import WiimException
14
15from .constants import BACKEND_GENERIC, BACKEND_OFFICIAL
16from .helpers import linkplay_group_compatible, match_slave_uuid_to_player_id
17
18if TYPE_CHECKING:
19 from pywiim import WiiMClient
20 from pywiim.models import DeviceInfo
21
22 from .linkplay_player import LinkPlayPlayer
23 from .player import WiimPlayer
24 from .provider import WiimProvider
25
26 # Either backend's player; both expose the small grouping surface used here.
27 type NativePlayer = WiimPlayer | LinkPlayPlayer
28
29# A leader's slave list is only re-read this often on the ordinary player poll; forced
30# refreshes (grouping commands, RenderingControl slave events, moves) bypass the TTL so a
31# real change is never delayed by it.
32SLAVE_TTL = 60.0
33
34# A device accepts a grouping command before its role has propagated, so membership is
35# polled up to this long before a still-mismatched role is declared a no-op.
36VERIFY_MAX_WAIT = 10.0
37VERIFY_POLL_INTERVAL = 1.0
38
39# Registration/availability announcements are coalesced into a single IO-free reconcile so
40# a burst of them heals discovery-order misses just once.
41RECONCILE_DEBOUNCE = 1.0
42RECONCILE_TASK_ID = "wiim_native_reconcile"
43REPUBLISH_TASK_ID = "wiim_native_republish"
44
45_NATIVE_BACKENDS = (BACKEND_OFFICIAL, BACKEND_GENERIC)
46
47
48class NativeGroupRole(StrEnum):
49 """A player's role in the native multiroom topology."""
50
51 LEADER = "leader"
52 FOLLOWER = "follower"
53 STANDALONE = "standalone"
54
55
56class NativeGroupCoordinator:
57 """
58 The single native topology authority across both WiiM backends.
59
60 It owns which players are leaders, followers or standalone and who belongs to which
61 group; the official SDK and the linked protocol players remain the playback/state
62 authorities. Topology is rebuilt from the raw slave-uuid lists that leaders report
63 (read over a low-level command client) and resolved against the currently registered
64 players, so a discovery-order miss self-heals on the next reconcile.
65 """
66
67 def __init__(self, provider: WiimProvider) -> None:
68 """Initialize the coordinator for a provider instance."""
69 self._provider = provider
70 self._mass = provider.mass
71 self._lock = asyncio.Lock()
72 # serializes grouping commands so concurrent moves targeting the same member from
73 # different leaders cannot interleave their check/mutate/verify phases
74 self._command_lock = asyncio.Lock()
75 # leader player_id -> raw slave uuids from its last successful topology read
76 self._raw_slaves: dict[str, list[str]] = {}
77 self._refreshed_at: dict[str, float] = {}
78 # per-leader lock so overlapping refreshes cannot apply out of order
79 self._refresh_locks: dict[str, asyncio.Lock] = {}
80 # players that report themselves as a follower (of a possibly unknown leader)
81 self._self_follower: set[str] = set()
82 # atomically rebuilt indexes; readers see a consistent snapshot between awaits
83 self._members: dict[str, list[str]] = {}
84 self._reverse: dict[str, str] = {}
85 self._role: dict[str, NativeGroupRole] = {}
86
87 # --- Topology queries (lock-free, read the last rebuilt snapshot) ---
88
89 def role_of(self, player_id: str) -> NativeGroupRole:
90 """Return the native topology role of a player."""
91 return self._role.get(player_id, NativeGroupRole.STANDALONE)
92
93 def members_of(self, player_id: str) -> list[str]:
94 """
95 Return the group member ids for a leader (leader first), else an empty list.
96
97 :param player_id: The player whose managed members to return.
98 """
99 return list(self._members.get(player_id, ()))
100
101 def leader_of(self, player_id: str) -> str | None:
102 """Return the leader a follower belongs to, or None when not a follower."""
103 return self._reverse.get(player_id)
104
105 def is_unknown_leader_follower(self, player_id: str) -> bool:
106 """
107 Return whether a player follows a leader MA has not discovered.
108
109 Such a device belongs to an externally-created group MA cannot see the leader of,
110 so it can be neither cleanly detached nor safely regrouped and must have grouping
111 withdrawn entirely.
112
113 :param player_id: The player to check.
114 """
115 return self._is_unknown_leader_follower(player_id)
116
117 def can_group_with(self, player: NativePlayer) -> set[str]:
118 """
119 Return every reachable peer of either backend this player may group with.
120
121 Core applies the final grouping filter and auto-ungroups a known follower before
122 regrouping it. A device whose own native grouping API is unreachable offers no native
123 peers (grouping via a linked protocol is decided separately by core), and a follower
124 of a leader MA has NOT discovered is excluded (it cannot be cleanly moved). Two generic
125 devices are only paired when they share a known, matching router-based multiroom
126 generation, so the UI never offers a generic pair that the command path would then
127 reject. Cross-backend generation cannot be known without a device read, so those pairs
128 stay candidates and fail closed at command time if incompatible.
129
130 :param player: The player requesting its grouping candidates.
131 """
132 if not player.native_available or self._is_unknown_leader_follower(player.player_id):
133 # the requesting device cannot lead a native group right now (its grouping API is
134 # unreachable, or it follows a group MA has not discovered and cannot be moved),
135 # so it offers no native candidates of its own.
136 return set()
137 return {
138 peer.player_id
139 for peer in self._native_players()
140 if peer.native_available
141 and peer.player_id != player.player_id
142 and not self._is_unknown_leader_follower(peer.player_id)
143 and self._offerable_pair(player, peer)
144 }
145
146 # --- Topology feeders ---
147
148 def set_self_role(self, player_id: str, is_follower: bool) -> bool:
149 """
150 Record a player's own report of whether it is a follower.
151
152 This is the fallback for a device following a leader MA has not discovered: the
153 leader never lists it here, but the device itself knows it is grouped. The
154 recorded value only takes effect on the next reconcile.
155
156 :param player_id: The player reporting its own role.
157 :param is_follower: Whether the device reports itself as a follower.
158 :return: Whether this changed the player's recorded self-follower state.
159 """
160 if is_follower:
161 if player_id in self._self_follower:
162 return False
163 self._self_follower.add(player_id)
164 return True
165 if player_id not in self._self_follower:
166 return False
167 self._self_follower.discard(player_id)
168 return True
169
170 def set_leader_slaves(
171 self, player_id: str, raw_slave_uuids: list[str], observed_at: float | None = None
172 ) -> bool:
173 """
174 Record a leader's raw slave-uuid list from a topology read.
175
176 :param player_id: The leader that reported the slave list.
177 :param raw_slave_uuids: The slave uuids exactly as the device reported them.
178 :param observed_at: The monotonic time the read that produced this list *started*.
179 The single-owner tie-break ranks claims by this, so a longer-running older read
180 that completes late cannot outrank a newer read; defaults to now for a list set
181 directly (e.g. cleared on a confirmed join), which is a genuine "now" observation.
182 :return: Whether the recorded list differs from the previously cached one.
183 """
184 changed = self._raw_slaves.get(player_id) != raw_slave_uuids
185 self._raw_slaves[player_id] = raw_slave_uuids
186 self._refreshed_at[player_id] = time.monotonic() if observed_at is None else observed_at
187 return changed
188
189 async def refresh_leader(self, player: NativePlayer, *, force: bool = False) -> bool:
190 """
191 Re-read a player's live group state over its command client and reconcile.
192
193 A live read yields both the device's own role (the self-follower signal) and, when
194 it leads, its slave list. Reads for one player are serialized so overlapping
195 refreshes cannot apply an older response after a newer one, and a failed read keeps
196 the previous state so a temporarily unreachable device loses neither its members
197 nor its self-role.
198
199 :param player: The player whose live group state to read.
200 :param force: Read now even if the slow TTL has not elapsed.
201 :return: Whether a fresh group state was actually read and applied.
202 """
203 if not player.native_ip or not player.native_available:
204 # no reachable address to command yet (never seen or currently unavailable);
205 # keep the cached topology instead of doing a request that will only time out.
206 return False
207 lock = self._refresh_locks.setdefault(player.player_id, asyncio.Lock())
208 async with lock:
209 # re-check the TTL inside the lock: a concurrent refresh may have just run
210 if not force and (
211 time.monotonic() - self._refreshed_at.get(player.player_id, 0.0) < SLAVE_TTL
212 ):
213 return False
214 client = player.make_command_client()
215 # stamp the read by when it STARTED, so a longer-running older observation that
216 # completes after a newer one cannot win the single-owner tie-break and move a
217 # follower back to a stale leader.
218 observed_at = time.monotonic()
219 try:
220 info = await client.get_device_group_info()
221 if info.role == "slave":
222 is_follower, raw = True, []
223 else:
224 # a non-slave is a leader or solo, but get_device_group_info both
225 # swallows a failed slave-list read (reporting solo) and can report a
226 # master with empty slave uuids when it derived them from status ip
227 # strings; read the authoritative getSlaveList so members resolve and a
228 # transient failure retains the cached topology (handled below).
229 is_follower = False
230 raw = self._normalize_slave_uuids(await client.get_slaves_info())
231 except WiiMError as err:
232 self._provider.logger.debug(
233 "Failed to read group state for %s: %s", player.player_id, err
234 )
235 return False
236 # the player may have been unregistered/replaced during the await; never apply
237 # a stale device response to a different registered instance
238 if self._mass.players.get_player(player.player_id) is not player:
239 return False
240 # reuse the capabilities this read detected so a later fresh command client (the
241 # official backend builds one per call) does not re-probe the device
242 player.store_command_capabilities(client.capabilities)
243 self.set_self_role(player.player_id, is_follower)
244 self.set_leader_slaves(player.player_id, raw, observed_at=observed_at)
245 await self.reconcile()
246 return True
247
248 def unregister(self, player_id: str) -> None:
249 """Drop a permanently removed player from the topology cache."""
250 self._raw_slaves.pop(player_id, None)
251 self._refreshed_at.pop(player_id, None)
252 self._refresh_locks.pop(player_id, None)
253 self._self_follower.discard(player_id)
254
255 def schedule_reconcile(self) -> None:
256 """Schedule a debounced reconcile, coalescing bursts into one rebuild."""
257 self._mass.call_later(RECONCILE_DEBOUNCE, self.reconcile, task_id=RECONCILE_TASK_ID)
258
259 def schedule_republish(self) -> None:
260 """
261 Schedule every native player to re-publish its state, coalescing bursts.
262
263 A player's ``can_group_with`` depends on its peers' native availability and
264 compatibility, but a peer's health/metadata transition does not change the topology
265 and so is not picked up by :meth:`reconcile`. This forces every native player to
266 recompute and re-publish, so no peer keeps a stale candidate set that would offer a
267 native command the coordinator then rejects.
268 """
269 self._mass.call_later(RECONCILE_DEBOUNCE, self._republish_all, task_id=REPUBLISH_TASK_ID)
270
271 async def reconcile(self) -> None:
272 """Rebuild the topology indexes from the cached slave lists and notify changes."""
273 async with self._lock:
274 self._rebuild()
275
276 # --- Grouping commands ---
277
278 async def set_members(
279 self,
280 leader: NativePlayer,
281 player_ids_to_add: list[str] | None,
282 player_ids_to_remove: list[str] | None,
283 ) -> None:
284 """
285 Add or remove native group members for a leader, spanning both backends.
286
287 Same-backend official operations keep the SDK path; every other combination joins
288 or removes the follower over the low-level LinkPlay client. Every operation is
289 verified against the leader's own live slave list (not the follower's role, which
290 is unreachable for a legacy Wi-Fi Direct follower) and raises on a no-op.
291
292 :param leader: The player the grouping command was issued on.
293 :param player_ids_to_add: Player ids to join to this leader's group.
294 :param player_ids_to_remove: Player ids to remove from this leader's group.
295 """
296 add_ids = player_ids_to_add or []
297 remove_ids = player_ids_to_remove or []
298 # reject a contradictory request before any work: core can forward the same id in
299 # both lists when the parent membership is stale but the child still reports synced_to,
300 # and joining then immediately removing it is never the intent.
301 if len(set(add_ids)) != len(add_ids) or len(set(remove_ids)) != len(remove_ids):
302 raise PlayerCommandFailed(
303 f"Duplicate member id in grouping request on {leader.player_id}"
304 )
305 if conflicting := set(add_ids) & set(remove_ids):
306 raise PlayerCommandFailed(
307 f"Cannot add and remove the same member on {leader.player_id}: {conflicting}"
308 )
309 # core only locks the command's own leader, so serialize here: a concurrent move of
310 # the same member under a different leader must not interleave with this command's
311 # live check, mutation and verification.
312 async with self._command_lock, AsyncExitStack() as stack:
313 add_members = [self._require_member(mid) for mid in add_ids]
314 remove_members = [self._require_member(mid) for mid in remove_ids]
315 held: set[str] = set()
316 # Phase 1: lock the leader and the explicit targets before reading their live
317 # state, so a concurrent address rebuild cannot swap a generic client mid-read
318 # and have the command act on a stale-address topology.
319 await self._acquire_rebuild_locks(stack, [leader, *add_members, *remove_members], held)
320 # authoritative preflight, now under lock: read the leader's and every addition's
321 # live group and fail the whole batch if a fresh read cannot be obtained, so a
322 # device that externally became a follower (or gained followers) since its last
323 # poll is validated on live state, never repurposed from a stale cache.
324 for player in (leader, *add_members):
325 if not await self.refresh_leader(player, force=True):
326 raise PlayerCommandFailed(f"Cannot read the group state of {player.player_id}")
327 # Phase 2: joining a member dissolves the followers that read just revealed, so
328 # lock them too before any mutation. The recursive dissolve never re-takes locks.
329 followers: list[NativePlayer] = []
330 for member in (*add_members, *remove_members):
331 for follower_id in self.members_of(member.player_id):
332 if follower_id == member.player_id:
333 continue
334 follower = self._mass.players.get_player(follower_id)
335 if getattr(follower, "linkplay_backend", None) in _NATIVE_BACKENDS:
336 followers.append(cast("NativePlayer", follower))
337 await self._acquire_rebuild_locks(stack, followers, held)
338 await self._apply_members(leader, add_members, remove_members)
339
340 # --- Private helpers ---
341
342 async def _acquire_rebuild_locks(
343 self, stack: AsyncExitStack, players: list[NativePlayer], held: set[str]
344 ) -> None:
345 """
346 Acquire the not-yet-held rebuild locks of the given players in a deterministic order.
347
348 Acquiring in sorted player-id order and skipping already-held devices keeps the
349 overall order stable across the command's two locking phases; the command lock
350 serializes whole commands, so this can never deadlock against another command, and
351 an address rebuild only ever holds a single device's lock.
352
353 :param stack: The exit stack that releases every acquired lock when the command ends.
354 :param players: The players whose rebuild locks to acquire.
355 :param held: The set of player ids whose locks are already held, updated in place.
356 """
357 for player in sorted(players, key=lambda candidate: candidate.player_id):
358 if player.player_id in held:
359 continue
360 held.add(player.player_id)
361 if (lock := self._rebuild_lock_for(player)) is not None:
362 await stack.enter_async_context(lock)
363
364 async def _apply_members(
365 self,
366 leader: NativePlayer,
367 add_members: list[NativePlayer],
368 remove_members: list[NativePlayer],
369 ) -> None:
370 """Validate the whole batch, then run the grouping operations (caller holds locks)."""
371 try:
372 # full validation before any hardware mutation, so an unknown, unreachable or
373 # generation-incompatible target fails the whole request before any speaker has
374 # joined or left, rather than leaving a partially mutated group behind.
375 self._guard_regroupable(leader)
376 # a low-level (generic or mixed) join needs the leader's device info for the
377 # compatibility gate and the join itself; read it once for the whole batch and
378 # reuse it, rather than re-fetching it for every member.
379 leader_info: DeviceInfo | None = None
380 if any(not self._is_official_pair(leader, member) for member in add_members):
381 leader_info = await self._device_info(leader)
382 join_plan: dict[str, tuple[DeviceInfo | None, tuple[NativePlayer, ...]] | None] = {}
383 for member in add_members:
384 self._guard_regroupable(member)
385 join_plan[member.player_id] = await self._prevalidate_join(
386 leader, member, leader_info
387 )
388 for member in remove_members:
389 self._prevalidate_leave(leader, member)
390 for member in add_members:
391 await self._join(leader, member, join_plan[member.player_id])
392 for member in remove_members:
393 await self._leave(leader, member)
394 finally:
395 await self.refresh_leader(leader, force=True)
396
397 def _rebuild_lock_for(self, player: NativePlayer) -> asyncio.Lock | None:
398 """Return the per-device rebuild lock that grouping must hold, if the backend has one."""
399 return getattr(player, "grouping_rebuild_lock", None)
400
401 def _guard_regroupable(self, player: NativePlayer) -> None:
402 """
403 Assert a player can take part in an MA-driven grouping command right now.
404
405 A device that follows a leader MA has not discovered belongs to an external group
406 that cannot be cleanly detached, so it can be neither a leader nor a member here.
407 Re-checked at command time because core may have cached the request (or it may have
408 waited on the command lock) since the candidate set was computed.
409
410 :param player: The leader or member to validate.
411 """
412 if self._is_unknown_leader_follower(player.player_id):
413 raise PlayerCommandFailed(
414 f"Cannot regroup {player.player_id}: it follows a group MA has not discovered"
415 )
416
417 @staticmethod
418 def _is_official_pair(leader: NativePlayer, member: NativePlayer) -> bool:
419 """Return whether both players are official WiiM devices (the SDK grouping path)."""
420 return (
421 leader.linkplay_backend == BACKEND_OFFICIAL
422 and member.linkplay_backend == BACKEND_OFFICIAL
423 )
424
425 async def _prevalidate_join(
426 self,
427 leader: NativePlayer,
428 member: NativePlayer,
429 leader_info: DeviceInfo | None,
430 ) -> tuple[DeviceInfo | None, tuple[NativePlayer, ...]] | None:
431 """
432 Validate an addition before any mutation and return the plan the join needs.
433
434 Same-backend official joins are validated by the SDK itself (returns ``None``);
435 every other combination must reach both devices and share a known, matching
436 router-based multiroom generation. The member's own live group was read before the
437 locks were taken, so it is validated here too: it must not still lead any slave MA
438 cannot dissolve (it would be orphaned once the member becomes a follower). The
439 managed followers to dissolve are captured so execution needs no further read.
440
441 :param leader: The leader the member would join.
442 :param member: The member that would join the leader.
443 :param leader_info: The leader's device info, read once for the whole batch.
444 """
445 if self._is_official_pair(leader, member):
446 return None
447 if not leader.native_ip:
448 raise PlayerCommandFailed(f"Cannot group {member.player_id}: leader address unknown")
449 member_info = await self._device_info(member)
450 if not linkplay_group_compatible(leader_info, member_info):
451 raise PlayerCommandFailed(
452 f"Cannot group {member.player_id} with {leader.player_id}: incompatible or "
453 "legacy Wi-Fi Direct LinkPlay multiroom"
454 )
455 if self._leads_unmanaged_slave(member):
456 raise PlayerCommandFailed(
457 f"Cannot group {member.player_id}: it still leads slaves MA cannot dissolve"
458 )
459 followers = tuple(
460 self._require_member(follower_id)
461 for follower_id in self.members_of(member.player_id)
462 if follower_id != member.player_id
463 )
464 return leader_info, followers
465
466 def _leads_unmanaged_slave(self, member: NativePlayer) -> bool:
467 """Return whether the member's live slave list holds a slave MA cannot resolve."""
468 registered = {player.player_id for player in self._native_players()}
469 return any(
470 match_slave_uuid_to_player_id(uuid, registered) is None
471 for uuid in self._raw_slaves.get(member.player_id, ())
472 )
473
474 def _prevalidate_leave(self, leader: NativePlayer, member: NativePlayer) -> None:
475 """
476 Validate a removal before any mutation, so a bad target fails the whole batch.
477
478 Only a member the leader still owns is actually detached; one the freshly refreshed
479 leader no longer lists (moved away or absent) is an idempotent skip in :meth:`_leave`,
480 so its reachability is irrelevant and must not abort the batch. For a member still
481 owned, a same-backend generic follower leaves over its own client, so an unreachable
482 one cannot be detached and fails up front rather than half-way through the batch.
483 Cross-backend and official removals go through the always-reachable leader (SDK
484 ungroup or leader-side kick), so the follower's own reachability is not required.
485
486 :param leader: The leader the member would leave.
487 :param member: The member that would leave the leader.
488 """
489 if member.player_id not in self.members_of(leader.player_id):
490 return
491 if (
492 leader.linkplay_backend == BACKEND_GENERIC
493 and member.linkplay_backend == BACKEND_GENERIC
494 and not member.native_available
495 ):
496 raise PlayerCommandFailed(f"{member.player_id} is not reachable to leave its group")
497
498 def _native_players(self) -> list[NativePlayer]:
499 """Return this provider's players that belong to either native backend."""
500 return [
501 cast("NativePlayer", player)
502 for player in self._provider.players
503 if getattr(player, "linkplay_backend", None) in _NATIVE_BACKENDS
504 ]
505
506 def _republish_all(self) -> None:
507 """Re-publish the state of every registered native player."""
508 for player in self._native_players():
509 player.on_native_group_update()
510
511 def _is_unknown_leader_follower(self, player_id: str) -> bool:
512 """Return whether a player follows a leader MA has not discovered."""
513 return (
514 self.role_of(player_id) == NativeGroupRole.FOLLOWER
515 and self.leader_of(player_id) is None
516 )
517
518 def _offerable_pair(self, player: NativePlayer, peer: NativePlayer) -> bool:
519 """
520 Return whether two players may be offered as a grouping pair in the UI.
521
522 Two generic devices are only offered when their cached generations are both known,
523 router-based and matching; any pair involving an official device is offered and
524 validated at command time, where the live device generation is read.
525
526 :param player: The player requesting candidates.
527 :param peer: The candidate peer being considered.
528 """
529 if player.linkplay_backend == BACKEND_GENERIC and peer.linkplay_backend == BACKEND_GENERIC:
530 return linkplay_group_compatible(
531 getattr(player, "cached_device_info", None),
532 getattr(peer, "cached_device_info", None),
533 )
534 return True
535
536 def _rebuild(self) -> None:
537 """Recompute roles and membership, then push state to every changed player."""
538 registered: dict[str, NativePlayer] = {
539 player.player_id: player for player in self._native_players()
540 }
541 # a fresh topology read for an unregistered player can never arrive, so its cached
542 # slave list (and self-reported role) is stale forever: drop it here.
543 for stale_id in [pid for pid in self._raw_slaves if pid not in registered]:
544 self.unregister(stale_id)
545 for stale_id in [pid for pid in self._self_follower if pid not in registered]:
546 self._self_follower.discard(stale_id)
547
548 candidate_members: dict[str, list[str]] = {}
549 for leader_id, raw in self._raw_slaves.items():
550 resolved = [
551 member_id
552 for uuid in raw
553 if (member_id := match_slave_uuid_to_player_id(uuid, registered)) is not None
554 and member_id != leader_id
555 ]
556 if resolved:
557 candidate_members[leader_id] = resolved
558
559 # a follower can be claimed by only one leader: while a moved device's old leader
560 # still has it in its (not-yet-refreshed) cached list, the most recently refreshed
561 # leader wins so one player never shows in two groups. Ties break on the leader id
562 # so the winner is fully deterministic.
563 owner: dict[str, str] = {}
564 for leader_id, resolved in candidate_members.items():
565 leader_key = (self._refreshed_at.get(leader_id, 0.0), leader_id)
566 for member_id in resolved:
567 incumbent = owner.get(member_id)
568 if incumbent is None or leader_key > (
569 self._refreshed_at.get(incumbent, 0.0),
570 incumbent,
571 ):
572 owner[member_id] = leader_id
573
574 # a device that is itself another leader's follower cannot also be a leader; its own
575 # (stale) slave list is ignored so nested/ghost groups never form.
576 members: dict[str, list[str]] = {}
577 for leader_id, resolved in candidate_members.items():
578 if leader_id in owner:
579 continue
580 owned = [member_id for member_id in resolved if owner.get(member_id) == leader_id]
581 if owned:
582 members[leader_id] = [leader_id, *owned]
583 reverse: dict[str, str] = {
584 member_id: leader_id
585 for leader_id, group in members.items()
586 for member_id in group
587 if member_id != leader_id
588 }
589 role: dict[str, NativeGroupRole] = {}
590 for player_id in registered:
591 if player_id in members:
592 role[player_id] = NativeGroupRole.LEADER
593 elif player_id in reverse:
594 role[player_id] = NativeGroupRole.FOLLOWER
595 elif player_id in self._self_follower:
596 # following a leader MA has not discovered: a follower with no known leader
597 # (leader_of stays None), still suppressed and not groupable.
598 role[player_id] = NativeGroupRole.FOLLOWER
599 elif any(
600 match_slave_uuid_to_player_id(uuid, registered) is None
601 for uuid in self._raw_slaves.get(player_id, ())
602 ):
603 # leads a native group whose slaves MA cannot resolve to registered players:
604 # it manages no members here (members_of stays empty), but it is NOT
605 # standalone, so it is never offered for a second (linked-protocol) group on
606 # top of its existing hardware group — which matters most during an API outage
607 # when that native group cannot be dissolved. A stale entry that still resolves
608 # to a registered player (a follower that moved away) is not treated as leading.
609 role[player_id] = NativeGroupRole.LEADER
610 else:
611 role[player_id] = NativeGroupRole.STANDALONE
612
613 old_members, old_reverse, old_role = self._members, self._reverse, self._role
614 old_unknown = {
615 player_id
616 for player_id in registered
617 if old_role.get(player_id) == NativeGroupRole.FOLLOWER and player_id not in old_reverse
618 }
619 new_unknown = {
620 player_id
621 for player_id in registered
622 if role[player_id] == NativeGroupRole.FOLLOWER and player_id not in reverse
623 }
624 # a change in which players follow an undiscovered leader flips every peer's
625 # can_group_with (those followers are excluded there), so all players must
626 # re-publish; otherwise only those whose own role/membership actually changed.
627 if old_unknown != new_unknown:
628 changed_ids = list(registered)
629 else:
630 changed_ids = [
631 player_id
632 for player_id in registered
633 if old_role.get(player_id, NativeGroupRole.STANDALONE) != role[player_id]
634 or old_members.get(player_id, []) != members.get(player_id, [])
635 or old_reverse.get(player_id) != reverse.get(player_id)
636 ]
637 # atomic swap: readers between awaits always see one consistent snapshot
638 self._members, self._reverse, self._role = members, reverse, role
639 # notify members-publishers (a leader in the old OR new snapshot) before followers:
640 # a follower's synced_to scans its leaders' cached group_members, so an old leader
641 # shedding it and a new leader gaining it must both refresh first.
642 changed_ids.sort(
643 key=lambda player_id: not (old_members.get(player_id) or members.get(player_id))
644 )
645 for player_id in changed_ids:
646 registered[player_id].on_native_group_update()
647
648 def _require_member(self, player_id: str) -> NativePlayer:
649 """
650 Return the registered native player for a grouping target.
651
652 :param player_id: The member id supplied to the grouping command.
653 """
654 member = self._mass.players.get_player(player_id)
655 if getattr(member, "linkplay_backend", None) not in _NATIVE_BACKENDS:
656 raise PlayerCommandFailed(f"Cannot group unknown or unsupported player {player_id}")
657 return cast("NativePlayer", member)
658
659 async def _join(
660 self,
661 leader: NativePlayer,
662 member: NativePlayer,
663 plan: tuple[DeviceInfo | None, tuple[NativePlayer, ...]] | None,
664 ) -> None:
665 """Join a member to a leader over the correct backend path and verify it."""
666 if self._is_official_pair(leader, member):
667 leader_udn = cast("WiimPlayer", leader).native_device_udn
668 member_udn = cast("WiimPlayer", member).native_device_udn
669 try:
670 await self._provider.wiim_controller.async_join_group(leader_udn, [member_udn])
671 except WiimException as err:
672 raise PlayerCommandFailed(f"Failed to group {member.player_id}: {err}") from err
673 else:
674 assert plan is not None # a low-level join is always prevalidated
675 await self._low_level_join(leader, member, plan[0], plan[1])
676 await self._verify_membership(leader, member, joined=True)
677 # the member is now a follower and manages no members of its own; drop any slave
678 # list it cached while it was a leader so a later leave cannot resurrect a ghost
679 # group, and record its follower role so it stays suppressed even if its leader is
680 # unregistered before the member's next live read (whose TTL was just advanced).
681 self.set_leader_slaves(member.player_id, [])
682 self.set_self_role(member.player_id, True)
683
684 async def _low_level_join(
685 self,
686 leader: NativePlayer,
687 member: NativePlayer,
688 leader_info: DeviceInfo | None,
689 followers: tuple[NativePlayer, ...],
690 ) -> None:
691 """Join a member to a leader over the low-level LinkPlay client (generic or mixed)."""
692 if not (leader_ip := leader.native_ip):
693 raise PlayerCommandFailed(f"Cannot group {member.player_id}: leader address unknown")
694 # the low-level join_slave (unlike the official SDK) does not disband a member that
695 # leads its own group, so dissolve the followers prevalidation captured from its live
696 # group first: otherwise they are orphaned once the member becomes a follower itself.
697 if followers:
698 await self._apply_members(member, [], list(followers))
699 try:
700 await member.make_command_client().join_slave(leader_ip, master_device_info=leader_info)
701 except WiiMError as err:
702 raise PlayerCommandFailed(f"Failed to group {member.player_id}: {err}") from err
703
704 async def _leave(self, leader: NativePlayer, member: NativePlayer) -> None:
705 """Remove a member from a leader over the correct backend path and verify it."""
706 # confirm, from the leader's own live slave list, that the member is currently this
707 # leader's follower. A member that is solo or grouped under a different leader is not
708 # detached (idempotent), so another leader's follower is safe.
709 if not await self.refresh_leader(leader, force=True):
710 raise PlayerCommandFailed(
711 f"Could not read the group state of leader {leader.player_id}"
712 )
713 if member.player_id not in self.members_of(leader.player_id):
714 return
715 if (
716 leader.linkplay_backend == BACKEND_OFFICIAL
717 and member.linkplay_backend == BACKEND_OFFICIAL
718 ):
719 await self._official_detach(leader, member)
720 elif (
721 leader.linkplay_backend == BACKEND_GENERIC
722 and member.linkplay_backend == BACKEND_GENERIC
723 ):
724 # a same-backend generic follower always sits on the router network, so it can
725 # leave itself.
726 try:
727 await member.make_command_client().leave_group()
728 except WiiMError as err:
729 raise PlayerCommandFailed(f"Failed to ungroup {member.player_id}: {err}") from err
730 else:
731 # a cross-backend follower is kicked from the leader, which is always reachable
732 # and knows its address even on a private Wi-Fi Direct network.
733 await self._leader_side_detach(leader, member)
734 await self._verify_membership(leader, member, joined=False)
735 # the member is confirmed no longer this leader's follower; clear any stale
736 # self-follower flag from its last live read so its state and grouping unblock now
737 # instead of on its next poll (a concurrently discovered new leader still wins,
738 # because the reverse index outranks the self-follower flag on reconcile).
739 if self.set_self_role(member.player_id, False):
740 await self.reconcile()
741
742 async def _official_detach(self, leader: NativePlayer, member: NativePlayer) -> None:
743 """Ungroup an official follower via the SDK, falling back to a leader-side kick."""
744 member_udn = cast("WiimPlayer", member).native_device_udn
745 try:
746 await self._provider.wiim_controller.async_ungroup_device(member_udn)
747 except WiimException as err:
748 raise PlayerCommandFailed(f"Failed to ungroup {member.player_id}: {err}") from err
749 # the SDK ungroups from its own managed cache and can silently no-op on a
750 # recovered/external group; if the leader still lists the follower, kick it from the
751 # leader (as the mixed path does).
752 if await self.refresh_leader(leader, force=True) and member.player_id in self.members_of(
753 leader.player_id
754 ):
755 await self._leader_side_detach(leader, member)
756
757 async def _leader_side_detach(self, leader: NativePlayer, member: NativePlayer) -> None:
758 """
759 Remove a follower by kicking it from the leader.
760
761 The leader is always reachable and knows the follower's address (even the private
762 Wi-Fi Direct one), so kicking from the leader works where a leave sent to the
763 follower's own LAN address would not.
764
765 :param leader: The leader the follower belongs to.
766 :param member: The follower to remove.
767 """
768 client = leader.make_command_client()
769 if not (slave_ip := await self._leader_slave_ip(client, member)):
770 raise PlayerCommandFailed(
771 f"Could not resolve {member.player_id}'s address on leader {leader.player_id}"
772 )
773 # a freshly built command client has no cached group role, so prime it as master
774 # (state-only, no request) to satisfy kick_slave's master guard; membership was
775 # already confirmed from the leader's live slave list above.
776 await client.create_group()
777 try:
778 await client.kick_slave(slave_ip)
779 except WiiMError as err:
780 raise PlayerCommandFailed(f"Failed to ungroup {member.player_id}: {err}") from err
781
782 async def _leader_slave_ip(self, client: WiiMClient, member: NativePlayer) -> str | None:
783 """
784 Resolve a follower's address from the leader's own slave list.
785
786 Returns ``None`` only when the list is read but does not (yet) contain the member;
787 a failed read raises a chained typed error rather than masquerading as "no address".
788 """
789 try:
790 slaves = await client.get_slaves_info()
791 except WiiMError as err:
792 raise PlayerCommandFailed(
793 f"Failed to read the leader's slave list while resolving {member.player_id}: {err}"
794 ) from err
795 for slave in slaves:
796 if (
797 isinstance(slave, dict)
798 and match_slave_uuid_to_player_id(slave.get("uuid"), (member.player_id,))
799 == member.player_id
800 ):
801 ip = slave.get("ip")
802 return ip if isinstance(ip, str) and ip else None
803 return None
804
805 @staticmethod
806 def _normalize_slave_uuids(slaves_info: list[dict[str, Any]]) -> list[str]:
807 """Extract the slave uuids from a raw getSlaveList response (addressable entries)."""
808 return [
809 (slave.get("uuid") or "").replace("uuid:", "")
810 for slave in slaves_info
811 if isinstance(slave, dict) and slave.get("ip")
812 ]
813
814 async def _verify_membership(
815 self, leader: NativePlayer, member: NativePlayer, *, joined: bool
816 ) -> None:
817 """
818 Confirm a grouping op took effect against the leader's own live slave list.
819
820 Verification is leader-side because a legacy Wi-Fi Direct follower moves onto the
821 leader's private network and is unreachable from the LAN, while the leader always
822 reports its slaves. A device accepts a grouping command before its role has
823 propagated, so the leader's list is polled until it matches (or a bounded timeout
824 elapses); a failed read fails closed.
825
826 :param leader: The leader the command targeted.
827 :param member: The player whose membership was expected to change.
828 :param joined: Whether the member was expected to join (True) or leave.
829 """
830 deadline = time.monotonic() + VERIFY_MAX_WAIT
831 while True:
832 if not await self.refresh_leader(leader, force=True):
833 raise PlayerCommandFailed(
834 f"Could not confirm grouping of {member.player_id}: "
835 f"no fresh topology for leader {leader.player_id}"
836 )
837 if (member.player_id in self.members_of(leader.player_id)) == joined:
838 return
839 if time.monotonic() >= deadline:
840 break
841 await asyncio.sleep(VERIFY_POLL_INTERVAL)
842 verb = "join" if joined else "leave"
843 raise PlayerCommandFailed(
844 f"{member.player_id} did not {verb} the group led by {leader.player_id}"
845 )
846
847 async def _device_info(self, player: NativePlayer) -> DeviceInfo:
848 """
849 Read a player's device info for the compatibility gate.
850
851 A failed read raises a chained typed error rather than being collapsed into a
852 "None" that the compatibility gate would then misreport as an incompatible or
853 legacy device, hiding the real API failure from the caller.
854
855 :param player: The player whose device info to read.
856 """
857 try:
858 return await player.make_command_client().get_device_info_model()
859 except WiiMError as err:
860 raise PlayerCommandFailed(
861 f"Could not read device info for {player.player_id}: {err}"
862 ) from err
863