/
/
/
1"""AirPlay Player implementations."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import ipaddress
8import time
9from typing import TYPE_CHECKING, cast
10
11from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
12from music_assistant_models.enums import (
13 ConfigEntryType,
14 ContentType,
15 CrossfadeMode,
16 IdentifierType,
17 MediaType,
18 PlaybackState,
19 PlayerFeature,
20 PlayerType,
21)
22from music_assistant_models.errors import PlayerCommandFailed
23from music_assistant_models.media_items import AudioFormat
24
25from music_assistant.controllers.streams.audio import overlay_active
26from music_assistant.helpers.util import get_primary_ip_address_from_zeroconf, is_valid_mac_address
27from music_assistant.models.player import DeviceInfo, Player, PlayerMedia
28from music_assistant.models.setup_flow import AbortFlow
29
30from . import announce
31from .constants import (
32 AIRPLAY_DEFAULT_PORT,
33 AIRPLAY_DISCOVERY_TYPE,
34 AIRPLAY_HIRES_AUDIO_FORMATS,
35 AIRPLAY_HIRES_SAMPLE_RATES,
36 AIRPLAY_PCM_FORMAT,
37 AIRPLAY_REJOIN_ATTEMPT_DELAYS,
38 AIRPLAY_VOLUME_ECHO_GRACE_S,
39 BASE_PLAYER_FEATURES,
40 CONF_AIRPLAY_CREDENTIALS,
41 CONF_BUFFER_DEPTH,
42 CONF_ENABLE_HIRES,
43 CONF_ENCRYPTION,
44 CONF_ENTRY_SYNC_ADJUST_AIRPLAY,
45 CONF_IGNORE_VOLUME,
46 CONF_PAIR_NOW,
47 CONF_PAIRING_PASSWORD,
48 CONF_PAIRING_PIN,
49 CONF_PASSWORD,
50 CONF_PASSWORD_INVALID,
51 CONF_RAOP_CREDENTIALS,
52 CONF_STORED_VOLUME,
53 CONF_STREAMING_MODE,
54 FALLBACK_VOLUME,
55 LEGACY_PAIRING_BIT,
56 PAIRING_PIN_FORMAT,
57 PASSWORD_BIT,
58 PIN_REQUIRED,
59 RAOP_DEFAULT_PORT,
60 RAOP_DISCOVERY_TYPE,
61 STREAMING_MODE_AP2_COMPAT,
62 STREAMING_MODE_AP2_NTP,
63 STREAMING_MODE_AP2_PTP,
64 STREAMING_MODE_AUTO,
65 STREAMING_MODE_RAOP,
66 StreamingProtocol,
67)
68from .helpers import (
69 default_buffer_depth,
70 default_hires_enabled,
71 get_decoded_property,
72 is_apple_device,
73 is_macos_device,
74 parse_airplay_features,
75 player_id_to_mac_address,
76 supports_airplay2,
77)
78from .stream_session import AirPlayStreamSession
79
80if TYPE_CHECKING:
81 from zeroconf.asyncio import AsyncServiceInfo
82
83 from music_assistant.models.setup_flow import SetupSession
84
85 from .pairing import AirPlayPairing
86 from .provider import AirPlayProvider
87 from .stream import AirPlayStream
88
89# Docker bridge subnet, sometimes wrongly advertised via mDNS by containerized devices.
90_DOCKER_SUBNET = ipaddress.ip_network("172.16.0.0/12")
91
92
93class AirPlayPlayer(Player):
94 """Base implementation shared by all AirPlay players."""
95
96 def __init__(
97 self,
98 provider: AirPlayProvider,
99 player_id: str,
100 raop_discovery_info: AsyncServiceInfo | None,
101 airplay_discovery_info: AsyncServiceInfo | None,
102 address: str,
103 display_name: str,
104 manufacturer: str,
105 model: str,
106 initial_volume: int = FALLBACK_VOLUME,
107 ) -> None:
108 """Initialize AirPlayPlayer."""
109 self.raop_discovery_info = raop_discovery_info
110 self.airplay_discovery_info = airplay_discovery_info
111 # Audio formats the receiver advertises, learned from its /info response;
112 # zero until that lands (or when the device publishes no format tables).
113 self.advertised_audio_formats = 0
114 self._attr_enabled_by_default = not is_macos_device(manufacturer, model)
115 super().__init__(provider, player_id)
116 self.address = address
117 self.stream: AirPlayStream | None = None
118 # Serializes the two paths that can put a cliairplay process on this
119 # receiver (the native stream session and the Sendspin bridge), from the
120 # moment either decides to displace what is published until it publishes
121 # its own stream. Two processes on one receiver reset each other's RTSP
122 # channel and both sessions die. Always taken INSIDE self._lock, never
123 # around it: play_media holds self._lock across the whole session start,
124 # which takes this lock for every member.
125 self.stream_spawn_lock = asyncio.Lock()
126 self.last_command_sent = 0.0
127 self._volume_reports_ignored_until = 0.0
128 self._lock = asyncio.Lock()
129 self._transitioning = False # Set during stream replacement to ignore stale DACP messages
130 self._rejoin_task: asyncio.Task[None] | None = None
131 # Set (static) player attributes
132 self._attr_name = display_name
133 self._attr_available = True
134 mac_address = player_id_to_mac_address(player_id)
135 self._attr_device_info = DeviceInfo(
136 model=model,
137 manufacturer=manufacturer,
138 )
139 # Only add MAC address if it's valid (not 00:00:00:00:00:00)
140 if is_valid_mac_address(mac_address):
141 self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, mac_address)
142 self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, address)
143 self._attr_device_info.add_identifier(IdentifierType.AIRPLAY_ID, player_id)
144 self._attr_volume_level = initial_volume
145 self._attr_can_group_with = {provider.instance_id}
146
147 @property
148 def protocol(self) -> StreamingProtocol:
149 """Get the streaming protocol to use/prefer for this player."""
150 # AirPlay 2 whenever the device can speak it and RAOP is not being forced;
151 # RAOP for legacy receivers (or when the RAOP streaming mode is set).
152 if self._is_airplay2_capable and self.streaming_mode != STREAMING_MODE_RAOP:
153 return StreamingProtocol.AIRPLAY2
154 return StreamingProtocol.RAOP
155
156 @property
157 def streaming_mode(self) -> str:
158 """
159 Return the effective per-player streaming mode.
160
161 Automatic unless the (advanced) streaming-mode setting pins a lane the
162 device actually offers; a stored value the device no longer advertises
163 falls back to Automatic rather than forcing an impossible route.
164 """
165 value = str(self.config.get_value(CONF_STREAMING_MODE, STREAMING_MODE_AUTO))
166 offered = {option.value for option in self.streaming_mode_options}
167 return value if value in offered else STREAMING_MODE_AUTO
168
169 @property
170 def streaming_mode_options(self) -> list[ConfigValueOption]:
171 """
172 Return the streaming-mode options this device can actually offer.
173
174 Every option is an escape from the automatic AirPlay 2 route, gated on
175 the device's own advertisements: the AirPlay 2 lanes need AirPlay 2
176 capability (PTP timing additionally needs the SupportsPTP bit), and
177 legacy RAOP needs an advertised _raop service to fall back to. A
178 RAOP-only device has no alternative lane and keeps Automatic only,
179 which hides the entry entirely. Apple receivers get every lane except
180 NTP timing â they render silence on an NTP-timed realtime stream
181 (hardware-measured). Of their lanes, the compatibility flow and
182 legacy RAOP are the escapes for networks where the PTP ports are
183 blocked; pinning PTP is an explicit choice of the normal lane.
184 """
185 options = [ConfigValueOption(STREAMING_MODE_AUTO, "Automatic (recommended)")]
186 if not self._is_airplay2_capable:
187 return options
188 apple = is_apple_device(self.device_info.manufacturer, self.device_info.model)
189 features = parse_airplay_features(self._advertised_features)
190 if (features >> 41) & 1:
191 options.append(ConfigValueOption(STREAMING_MODE_AP2_PTP, "AirPlay 2 - PTP timing"))
192 if not apple:
193 options.append(ConfigValueOption(STREAMING_MODE_AP2_NTP, "AirPlay 2 - NTP timing"))
194 options.append(
195 ConfigValueOption(STREAMING_MODE_AP2_COMPAT, "AirPlay 2 - compatibility mode")
196 )
197 if self.raop_discovery_info is not None:
198 options.append(ConfigValueOption(STREAMING_MODE_RAOP, "AirPlay 1 (RAOP)"))
199 return options
200
201 @property
202 def protocol_override(self) -> StreamingProtocol | None:
203 """
204 Return the user-forced streaming protocol, or None for automatic selection.
205
206 Only the RAOP streaming mode forces the protocol outright; the AirPlay 2
207 modes stay on the AirPlay 2 protocol and pin the flow/timing through the
208 binary's --protocol/--timing arguments instead. Otherwise the cliairplay
209 binary resolves the route itself from the mDNS TXT records (--protocol
210 auto) and the ``protocol`` property above only reflects MA's own planning
211 heuristic (timing, ports).
212 """
213 if self.streaming_mode == STREAMING_MODE_RAOP:
214 return StreamingProtocol.RAOP
215 return None
216
217 @property
218 def hires_playback_enabled(self) -> bool:
219 """Return if 24-bit hi-res playback is possible and enabled for this player."""
220 # 24-bit only works over the AirPlay 2 flow, so a device that streams RAOP
221 # (a legacy receiver, or the force-RAOP escape hatch) stays on the 16-bit
222 # base whatever it advertises.
223 return (
224 bool(self.advertised_audio_formats & AIRPLAY_HIRES_AUDIO_FORMATS)
225 and self.protocol == StreamingProtocol.AIRPLAY2
226 # the compat lane is 16-bit only, so hi-res stands down while the pin is active
227 and self.streaming_mode != STREAMING_MODE_AP2_COMPAT
228 and bool(self.config.get_value(CONF_ENABLE_HIRES, self._hires_default_enabled))
229 )
230
231 @property
232 def supported_sample_rates(self) -> list[tuple[int, int]]:
233 """Return the (sample_rate, bit_depth) pairs this player natively supports."""
234 if self.hires_playback_enabled:
235 return AIRPLAY_HIRES_SAMPLE_RATES
236 return [(AIRPLAY_PCM_FORMAT.sample_rate, AIRPLAY_PCM_FORMAT.bit_depth)]
237
238 @property
239 def needs_setup(self) -> bool:
240 """Return if the player needs setup."""
241 # A stored password satisfies password protection on its own (the binary
242 # authenticates with it directly; stored credentials are only its
243 # fallback), so the password side is fully covered by the check above.
244 if self.needs_password_setup:
245 return True
246 if self._requires_pin_pairing():
247 # Credentials for either protocol keep the player usable: the binary
248 # picks the best route for the credentials it has. Re-running the setup
249 # flow from the player settings offers replacing a stored pairing.
250 if not (
251 self.get_setup_value(CONF_AIRPLAY_CREDENTIALS)
252 or self.get_setup_value(CONF_RAOP_CREDENTIALS)
253 ):
254 return True
255 return False
256
257 @property
258 def setup_reason(self) -> str | None:
259 """Return why the player needs setup, or None when it is ready to use."""
260 if not self.needs_setup:
261 return None
262 return "password_required" if self.needs_password_setup else "pairing_required"
263
264 @property
265 def password_required(self) -> bool:
266 """Return if the device announces that it is password protected."""
267 # Two announcement forms, verified against live devices (including Apple
268 # TVs, which raise the password bit only while a password is actually
269 # set): receivers publish the password bit in sf/flags and/or the classic
270 # pw boolean. Enforcement can also exist WITHOUT any announcement (stale
271 # TXT after the password was enabled); that case is caught at connect
272 # time via password_invalid.
273 if self._get_flags() & PASSWORD_BIT:
274 return True
275 if raop_info := self.raop_discovery_info:
276 return (raop_info.decoded_properties.get("pw") or "").lower() == "true"
277 return False
278
279 @property
280 def password_invalid(self) -> bool:
281 """Return if the device rejected the stored password on its last connect."""
282 return bool(
283 self.mass.config.get_raw_player_config_value(
284 self.player_id, CONF_PASSWORD_INVALID, False
285 )
286 )
287
288 @property
289 def needs_password_setup(self) -> bool:
290 """Return if the device password still has to be entered through the setup flow."""
291 # The password is only ever entered through the setup flow, so both a
292 # device that announces password protection without one stored and a
293 # password the device rejected must send the user back into that flow.
294 if self.password_invalid:
295 return True
296 return self.password_required and not self.config.get_value(CONF_PASSWORD)
297
298 def set_password_invalid(self, invalid: bool) -> None:
299 """
300 Persist (or clear) the marker that the device rejected the stored password.
301
302 :param invalid: True when the device rejected the password, False once a
303 connect succeeded or a new password was stored.
304 """
305 if self.password_invalid == invalid:
306 # keeps a successful connect from writing the config on every stream
307 return
308 self.mass.config.set_raw_player_config_value(self.player_id, CONF_PASSWORD_INVALID, invalid)
309 # needs_setup/setup_reason are part of the player's own state inputs, so a
310 # plain update publishes the (dis)appeared setup action to the clients.
311 self.update_state()
312
313 @property
314 def requires_flow_mode(self) -> bool:
315 """Return if the player requires flow mode."""
316 return True
317
318 @property
319 def supported_features(self) -> set[PlayerFeature]:
320 """Return the supported features of this player."""
321 # PAUSE is always advertised, including while synced. This keeps the AirPlay
322 # player itself as the pause control target so pause() can decide what to do:
323 # a true pause for a single player, or a full session stop for a sync group
324 # (see pause()). If PAUSE were dropped while grouped, the players controller
325 # could fall through to a linked native player's pause (e.g. a Sonos acting as
326 # an AirPlay receiver), which only pauses the sync leader while the other
327 # members keep playing.
328 features = {*BASE_PLAYER_FEATURES, PlayerFeature.PAUSE}
329 # An announcement is mixed into the audio the player is already rendering, so
330 # the feature is only offered while there is live playback to mix into. Without
331 # it the players controller plays the announcement its own way, which leaves
332 # the device to whatever else may be streaming to it.
333 if not self.has_live_audio:
334 features.discard(PlayerFeature.PLAY_ANNOUNCEMENT)
335 return features
336
337 @property
338 def has_live_audio(self) -> bool:
339 """Return True if the player is rendering audio an announcement can mix into."""
340 if self.playback_state != PlaybackState.PLAYING:
341 return False
342 if self.stream is None or self.stream.superseded:
343 # A stream handed to a teardown stays published until its process is
344 # off the receiver, and a clip mixed into it dies with it.
345 return False
346 return self.stream.running and self.stream.connected
347
348 @property
349 def applies_announcement_volume(self) -> bool:
350 """Return True: the announcement volume is applied around the mixed clip."""
351 return True
352
353 @property
354 def can_group_with(self) -> set[str]:
355 """
356 Return player IDs this player can group with.
357
358 RAOP and AP2 players can group with other RAOP and/or AP2 players.
359 """
360 prov = cast("AirPlayProvider", self.provider)
361 return {
362 p.player_id for p in prov.get_players() if p.available and p.player_id != self.player_id
363 }
364
365 @property
366 def native_grouping_requires_own_stream(self) -> bool:
367 """Return True: members are attached to this player's own stream session."""
368 return True
369
370 @property
371 def live_session_members(self) -> list[str]:
372 """Return the id's of the players the running stream session feeds."""
373 # group membership is bookkeeping that outlives the session: a member can be
374 # dropped from the session (write failures) or never make it in (a refused
375 # late join) while still being listed as part of the group, and without a
376 # session there is nobody to render with at all
377 if self.stream and self.stream.running and self.stream.session:
378 return [x.player_id for x in self.stream.session.sync_clients]
379 return []
380
381 async def get_config_entries(self) -> list[ConfigEntry]:
382 """Return all (provider/player specific) Config Entries for the given player (if any)."""
383 # Pairing/credentials are no longer config entries: they are collected by the
384 # interactive setup flow (run_setup_flow) and stored in the player's setup_data.
385 base_entries: list[ConfigEntry] = []
386
387 # Effective RAOP state from the current (stored) streaming mode, so the
388 # RAOP-only entries show/hide consistently with it.
389 is_raop = self.protocol == StreamingProtocol.RAOP
390
391 # Streaming-mode escape hatch: a per-device pin of the protocol/timing
392 # lane for receivers whose automatic route misbehaves. Only offered
393 # when the device actually has a lane to choose (Apple receivers are
394 # always native AirPlay 2 with PTP and get no entry).
395 mode_options = self.streaming_mode_options
396 if len(mode_options) > 1:
397 base_entries.append(
398 ConfigEntry(
399 key=CONF_STREAMING_MODE,
400 type=ConfigEntryType.STRING,
401 options=mode_options,
402 default_value=STREAMING_MODE_AUTO,
403 category="protocol_generic",
404 advanced=True,
405 )
406 )
407
408 # 24-bit toggle, shown only when the device advertises 24-bit support
409 # (per-device default: see default_hires_enabled). Hidden rather than
410 # omitted when it does not: the formats are probed async after
411 # registration, and an entry absent from the registration-time config
412 # parse would drop the user's stored value until the next config save.
413 base_entries.append(
414 ConfigEntry(
415 key=CONF_ENABLE_HIRES,
416 type=ConfigEntryType.BOOLEAN,
417 default_value=self._hires_default_enabled,
418 hidden=not self.advertised_audio_formats & AIRPLAY_HIRES_AUDIO_FORMATS,
419 category="protocol_generic",
420 requires_reload=True,
421 )
422 )
423
424 # Regular AirPlay config entries
425 base_entries += [
426 CONF_ENTRY_SYNC_ADJUST_AIRPLAY,
427 ConfigEntry(
428 key=CONF_ENCRYPTION,
429 type=ConfigEntryType.BOOLEAN,
430 default_value=True,
431 hidden=not is_raop,
432 category="protocol_generic",
433 advanced=True,
434 ),
435 ConfigEntry(
436 key=CONF_PASSWORD,
437 type=ConfigEntryType.SECURE_STRING,
438 default_value=None,
439 required=False,
440 # Storage (and encryption) vehicle only: the device password is
441 # entered through the setup flow, which is also what a wrong
442 # password sends the user back to. A hidden entry keeps its stored
443 # value across config saves (the frontend never submits it).
444 hidden=True,
445 category="protocol_generic",
446 advanced=True,
447 ),
448 ConfigEntry(
449 key=CONF_IGNORE_VOLUME,
450 type=ConfigEntryType.BOOLEAN,
451 default_value=False,
452 category="protocol_generic",
453 advanced=True,
454 ),
455 # Receiver-queue depth presets. The range reaches past the standard
456 # 2 s receiver buffer because that figure is only what the binary
457 # assumes for a device that reports no window of its own, and the
458 # deepest starving devices ask for more than the assumption. The
459 # default comes from the device-family table, and Automatic resolves
460 # through that same table at stream time, so selecting it never
461 # downgrades an affected device.
462 ConfigEntry(
463 key=CONF_BUFFER_DEPTH,
464 type=ConfigEntryType.INTEGER,
465 options=[
466 ConfigValueOption(0),
467 ConfigValueOption(500),
468 ConfigValueOption(750),
469 ConfigValueOption(1000),
470 ConfigValueOption(1500),
471 ConfigValueOption(1750),
472 ConfigValueOption(2000),
473 ConfigValueOption(2500),
474 ConfigValueOption(3000),
475 ],
476 default_value=default_buffer_depth(
477 self.device_info.manufacturer or "",
478 self.device_info.model or "",
479 get_decoded_property(self.airplay_discovery_info, "fv")
480 if self.airplay_discovery_info
481 else None,
482 ),
483 category="protocol_generic",
484 advanced=True,
485 requires_reload=True,
486 ),
487 ]
488
489 return base_entries
490
491 async def run_setup_flow(self, session: SetupSession) -> None:
492 """
493 Run the interactive setup flow for this AirPlay player (streaming pairing).
494
495 :param session: The setup flow session used to interact with the user.
496 """
497 collected: dict[str, ConfigValueType] = {}
498 await self._run_streaming_pairing(session, collected)
499 await session.finish(collected)
500
501 async def stop(self) -> None:
502 """Send STOP command to player."""
503 # an explicit stop (including power-off routed as stop) is user intent:
504 # drop any pending automatic re-join
505 self.cancel_group_rejoin()
506 async with self._lock:
507 if self.stream and self.stream.session:
508 # forward stop to the entire stream session
509 await self.stream.session.stop()
510 elif cast("AirPlayProvider", self.provider).bridge_manager.stop_streaming(
511 self.player_id
512 ):
513 # Sendspin bridge active: it tears the transport down straight
514 # away and takes the player out of the Sendspin session
515 pass
516 elif self.stream and self.stream.running:
517 # Fallback: stop protocol directly
518 await self.stream.stop(force=True)
519 self.stream = None
520 self._attr_current_media = None
521 self.update_state()
522
523 async def play(self) -> None:
524 """Handle PLAY (unpause) command on the player."""
525 session = self.stream.session if self.stream and self.stream.running else None
526 if self.group_members or self.synced_to or (session and session.parked):
527 # Grouped pause parks the whole session (standby); unpausing one
528 # member cannot restart the group in sync, and a parked member is
529 # held with nothing being fed until a re-anchor - which ACTION=PLAY
530 # does not carry, so it would report playback over silence. The park
531 # outlives the group, so a player left alone by an ungroup is keyed
532 # on the park itself, not on its membership. Resume via the queue
533 # instead: play_media flushes and re-anchors every parked member at
534 # one shared instant. The queue can belong to a linked native parent
535 # (for example Sonos), so resolve it instead of using the AirPlay ID.
536 active_queue = self.mass.players.get_active_queue(self)
537 if active_queue is None:
538 raise PlayerCommandFailed(
539 f"Cannot resume AirPlay player {self.display_name} without an active queue"
540 )
541 await self.mass.player_queues.resume(active_queue.queue_id, fade_in=False)
542 return
543 async with self._lock:
544 if self.stream and self.stream.running:
545 if await self.stream.send_cli_command("ACTION=PLAY"):
546 # Resuming re-anchors playout; the binary zeroes its own
547 # re-anchor total on resume, so drop the tracked shift to
548 # keep the server and binary baselines aligned.
549 self.stream.reset_reanchor_shift()
550
551 async def pause(self) -> None:
552 """Send PAUSE command to player."""
553 if self.group_members or self.synced_to:
554 # A broadcast pause cannot keep independent member processes
555 # sample-aligned on resume. Instead the session is parked: every
556 # member stalls but keeps its connection (and remote control), and
557 # the queue's resume flushes and re-anchors over the live
558 # connections â the same coordinated warm restart as seek/next.
559 if (
560 self.stream
561 and self.stream.running
562 and self.stream.session
563 and await self.stream.session.standby()
564 ):
565 return
566 # Some member no longer has a live connection: full stop and let
567 # the queue controller resume from the saved position.
568 self.logger.debug("Sync group cannot be parked, using STOP instead of PAUSE")
569 await self.stop()
570 return
571
572 async with self._lock:
573 if not self.stream or not self.stream.running:
574 return
575 await self.stream.send_cli_command("ACTION=PAUSE")
576
577 async def play_media(self, media: PlayerMedia) -> None:
578 """Handle PLAY MEDIA on given player."""
579 # the player is being (re)purposed on purpose: drop any pending
580 # automatic re-join left over from an unexpected stream loss
581 self.cancel_group_rejoin()
582 async with self._lock:
583 if self.synced_to:
584 # this should not happen, but guard anyways
585 raise RuntimeError("Player is synced")
586 self._attr_current_media = media
587
588 sync_clients = self._get_sync_clients()
589 session_pcm_format = await self._get_session_pcm_format(sync_clients, media)
590
591 # Warm path: a live, compatible session absorbs the new media via a
592 # flush-refill in place (seek/next never pays the reconnect cost).
593 if (
594 self.stream
595 and self.stream.running
596 and self.stream.session
597 and self.stream.session.can_replace(sync_clients, session_pcm_format)
598 ):
599 self._transitioning = True
600 audio_source = self.mass.streams.get_stream(
601 media, session_pcm_format, self.player_id
602 )
603 if await self.stream.session.replace(audio_source, media):
604 self._transitioning = False
605 # A seek changes no media identity, so the identity-driven
606 # metadata callback stays silent and receivers would show
607 # a stale Now Playing position; nudge every member once
608 # the queue position has settled.
609 for member in self.stream.session.sync_clients:
610 self.mass.call_later(
611 1,
612 member.on_player_media_updated,
613 task_id=f"player_media_updated_{member.player_id}",
614 )
615 return
616 # warm replacement failed; fall through to a cold restart
617
618 # Cold path: stop any existing stream and set up from scratch
619 if self.stream and self.stream.running and self.stream.session:
620 # Set transitioning flag to ignore stale DACP messages (like prevent-playback)
621 self._transitioning = True
622 stopped_stream = self.stream
623 await self.stream.session.stop()
624 # Only drop what this call stopped: tearing a group session down
625 # awaits every member, and a bridge can publish its own stream
626 # here. Erasing that would leave the start below with nothing to
627 # displace and a live process still on the speaker.
628 if self.stream is stopped_stream:
629 self.stream = None
630
631 # select audio source
632 audio_source = self.mass.streams.get_stream(media, session_pcm_format, self.player_id)
633
634 # setup StreamSession for player (and its sync childs if any)
635 provider = cast("AirPlayProvider", self.provider)
636 stream_session = AirPlayStreamSession(
637 provider,
638 sync_clients,
639 session_pcm_format,
640 media,
641 )
642 await stream_session.start(audio_source)
643 self._transitioning = False
644
645 async def play_announcement(
646 self, announcement: PlayerMedia, volume_level: int | None = None
647 ) -> None:
648 """
649 Play an announcement natively, mixed over the audio the player is rendering.
650
651 :param announcement: Details of the announcement that needs to be played.
652 :param volume_level: Optional volume level for the announcement.
653 """
654 # The lock windows live inside the orchestration: the dispatch decision and
655 # the arming hold self._lock like play_media does, while the multi-second
656 # clip waits run outside it (see announce.py).
657 await announce.play_announcement(self, announcement, volume_level)
658
659 async def volume_set(self, volume_level: int) -> None:
660 """Send VOLUME_SET command to given player."""
661 # Record before sending: the connect-time volume push reads this attribute,
662 # so a send that suspends first would let that push send the stale level.
663 self._attr_volume_level = volume_level
664 if self.stream and self.stream.running and self.volume_muted is not True:
665 await self.stream.send_cli_command(f"VOLUME={volume_level}")
666 self.update_state()
667 # store last state in playerconfig
668 self.mass.config.set_raw_player_config_value(
669 self.player_id, CONF_STORED_VOLUME, volume_level
670 )
671
672 async def volume_mute(self, muted: bool) -> None:
673 """Handle VOLUME_MUTE command on the player."""
674 self._attr_volume_muted = muted
675 if self.stream and self.stream.running:
676 volume = 0 if muted else (self.volume_level or 0)
677 await self.stream.send_cli_command(f"VOLUME={volume}")
678 self.update_state()
679
680 async def set_members(
681 self,
682 player_ids_to_add: list[str] | None = None,
683 player_ids_to_remove: list[str] | None = None,
684 ) -> None:
685 """Handle SET_MEMBERS command on the player."""
686 async with self._lock:
687 if self.synced_to:
688 # this should not happen, but guard anyways
689 raise RuntimeError("Player is synced, cannot set members")
690 if not player_ids_to_add and not player_ids_to_remove:
691 # nothing to do
692 return
693
694 stream_session = (
695 self.stream.session
696 if self.stream and self.stream.running and self.stream.session
697 else None
698 )
699 # handle removals first
700 if player_ids_to_remove:
701 if self.player_id in player_ids_to_remove:
702 # Callers only ask for this leader alone or for the whole group at once.
703 # A partial self+subset removal would need the other requested members
704 # released here as well, instead of returning right after the leader.
705 remaining_members = [
706 member_id
707 for member_id in self._attr_group_members
708 if member_id != self.player_id and member_id not in player_ids_to_remove
709 ]
710 if stream_session and remaining_members:
711 # Members stay behind: remove only this leader client,
712 # the session continues for the remaining players
713 await stream_session.remove_client(self, reason="leader removed from group")
714 elif stream_session:
715 # The whole group is being removed, tear the session down
716 await stream_session.stop()
717 self._attr_group_members = []
718 self.update_state()
719 return
720
721 for child_player in self._get_sync_clients():
722 if child_player.player_id in player_ids_to_remove:
723 # update group_members first to prevent race conditions
724 # where a concurrent play_media could re-include this player
725 if child_player.player_id in self._attr_group_members:
726 self._attr_group_members.remove(child_player.player_id)
727 if stream_session:
728 await stream_session.remove_client(
729 child_player, reason="child removed from group"
730 )
731 elif child_player.stream and child_player.stream.running:
732 # leader's stream is no longer running but child still has
733 # an active stream - stop it directly
734 await child_player.stream.stop(force=True)
735
736 # If group leader is left alone after removals, clear the group_members list
737 if (
738 self._attr_group_members
739 and len(self._attr_group_members) == 1
740 and self.player_id in self._attr_group_members
741 ):
742 self._attr_group_members = []
743
744 # handle additions
745 for player_id in player_ids_to_add or []:
746 if player_id == self.player_id or player_id in self.group_members:
747 # nothing to do: player is already part of the group
748 continue
749 child_player_to_add: AirPlayPlayer | None = cast(
750 "AirPlayPlayer | None", self.mass.players.get_player(player_id)
751 )
752 if not child_player_to_add:
753 # should not happen, but guard against it
754 continue
755
756 # ensure the child does not have an existing stream session active
757 if child_player_to_add := cast(
758 "AirPlayPlayer | None", self.mass.players.get_player(player_id)
759 ):
760 if (
761 child_player_to_add.playback_state == PlaybackState.PAUSED
762 and child_player_to_add.stream
763 ):
764 # Stop the paused stream to avoid a deadlock situation
765 await child_player_to_add.stream.stop()
766 if (
767 child_player_to_add.stream
768 and child_player_to_add.stream.running
769 and child_player_to_add.stream.session
770 and child_player_to_add.stream.session != stream_session
771 ):
772 await child_player_to_add.stream.session.remove_client(
773 child_player_to_add, reason="moving to different session"
774 )
775
776 # add new child to the existing stream (RAOP or AirPlay2) session (if any)
777 self._attr_group_members.append(player_id)
778 if stream_session and child_player_to_add is not None:
779 # Skip add_client if the player is already streaming in this session
780 # (e.g. after a dynamic leader switch where the stream continues)
781 if child_player_to_add not in stream_session.sync_clients:
782 await stream_session.add_client(child_player_to_add)
783 elif self.active_output_protocol not in (None, "native"):
784 # Members can only be attached to this player's own stream session, which
785 # does not exist while it renders through one of its output protocols.
786 self.logger.warning(
787 "%s joined the group of %s while that player renders through another "
788 "output protocol: there is no stream session to join, so it stays silent",
789 child_player_to_add.display_name if child_player_to_add else player_id,
790 self.display_name,
791 )
792
793 # Ensure group leader includes itself in group_members when it has members
794 # This is required for the synced_to property to work correctly
795 if self._attr_group_members and self.player_id not in self._attr_group_members:
796 self._attr_group_members.insert(0, self.player_id)
797
798 # always update the state after modifying group members
799 self.update_state()
800
801 @property
802 def ignore_volume_reports(self) -> bool:
803 """Return True if the device's own volume reports must not be acted on."""
804 if self._volume_reports_ignored_until > time.time():
805 # a level we sent ourselves is still echoing back
806 return True
807 return bool(
808 self.config.get_value(CONF_IGNORE_VOLUME)
809 or self.device_info.manufacturer.lower() == "apple"
810 )
811
812 def suppress_volume_reports(self, seconds: float = AIRPLAY_VOLUME_ECHO_GRACE_S) -> None:
813 """
814 Ignore the device's own volume reports for the given time.
815
816 :param seconds: How long from now the reports are ignored; a window that is
817 already open is only ever extended.
818 """
819 self._volume_reports_ignored_until = max(
820 self._volume_reports_ignored_until, time.time() + seconds
821 )
822
823 def update_volume_from_device(self, volume: int) -> None:
824 """Update volume from device feedback."""
825 if self.ignore_volume_reports:
826 return
827
828 cur_volume = self.volume_level or 0
829 if abs(cur_volume - volume) > 1 or (time.time() - self.last_command_sent) > 3:
830 self.mass.create_task(self._adopt_device_volume(volume))
831 else:
832 self._attr_volume_level = volume
833 self.mass.config.set_raw_player_config_value(self.player_id, CONF_STORED_VOLUME, volume)
834 self.update_state()
835
836 def set_discovery_info(self, discovery_info: AsyncServiceInfo, display_name: str) -> None:
837 """Set/update the discovery info for the player."""
838 self._attr_name = display_name
839 if discovery_info.type == AIRPLAY_DISCOVERY_TYPE:
840 self.airplay_discovery_info = discovery_info
841 elif discovery_info.type == RAOP_DISCOVERY_TYPE:
842 self.raop_discovery_info = discovery_info
843 else: # guard
844 return
845 cur_address = self.address
846 prefer_ipv6 = ":" in str(self.mass.streams.publish_ip)
847 new_address = get_primary_ip_address_from_zeroconf(discovery_info, prefer_ipv6=prefer_ipv6)
848 if new_address is None:
849 # should always be set, but guard against None
850 return
851 if cur_address != new_address:
852 # Ignore mDNS updates that replace a routable address with a Docker bridge one.
853 try:
854 if (
855 cur_address
856 and ipaddress.ip_address(new_address) in _DOCKER_SUBNET
857 and ipaddress.ip_address(cur_address) not in _DOCKER_SUBNET
858 ):
859 self.logger.warning(
860 "Ignoring mDNS update from %s to Docker address %s",
861 cur_address,
862 new_address,
863 )
864 self.update_state()
865 return
866 except ValueError:
867 pass
868 self.logger.debug("Address updated from %s to %s", cur_address, new_address)
869 self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, new_address)
870 self.address = new_address
871 self.update_state()
872
873 def set_state_from_stream(
874 self,
875 state: PlaybackState | None = None,
876 elapsed_time: float | None = None,
877 stream: AirPlayStream | None = None,
878 ) -> None:
879 """
880 Set the playback state from stream (RAOP or AirPlay2).
881
882 :param state: New playback state (or None to keep current).
883 :param elapsed_time: New elapsed time (or None to keep current).
884 :param stream: The stream instance sending this update (for validation).
885 """
886 # Ignore state updates from old/stale streams
887 if stream is not None and stream != self.stream:
888 return
889 # The stream reclaims the device: an external (Companion-observed)
890 # source snapshot can leak in during a brief stream-restart window and
891 # would otherwise stick, freezing the UI on a stale "external source"
892 # view while we stream. While MA streams, the stream is the sole
893 # authority on this player's state.
894 active_source = getattr(self, "_attr_active_source", None)
895 if active_source is not None and active_source in getattr(self, "_external_source_ids", ()):
896 media = getattr(self, "_attr_current_media", None)
897 if media is not None and media.source_id == active_source:
898 self._attr_current_media = None
899 self._attr_active_source = None
900 if state is not None:
901 self._attr_playback_state = state
902 if elapsed_time is not None:
903 self._attr_elapsed_time = elapsed_time
904 self._attr_elapsed_time_last_updated = time.time()
905 self.update_state()
906
907 def get_stream_pcm_format(self, session_pcm_format: AudioFormat) -> AudioFormat:
908 """
909 Return the PCM format to feed this player's cliairplay process.
910
911 :param session_pcm_format: The PCM format of the (shared) stream session.
912 """
913 if not self.hires_playback_enabled:
914 return AIRPLAY_PCM_FORMAT
915 # 24-bit: the binary expects raw s32le input on stdin (--bitdepth 24)
916 # and truncates to 24-bit ALAC internally.
917 supported_rates = {sample_rate for sample_rate, _ in self.supported_sample_rates}
918 sample_rate = (
919 session_pcm_format.sample_rate
920 if session_pcm_format.sample_rate in supported_rates
921 else AIRPLAY_PCM_FORMAT.sample_rate
922 )
923 return AudioFormat(
924 content_type=ContentType.PCM_S32LE,
925 sample_rate=sample_rate,
926 bit_depth=24,
927 )
928
929 @property
930 def owns_volume(self) -> bool:
931 """
932 Return True if this output is the resolved owner of its own volume.
933
934 AirPlay volume is the receiver's own volume: setting it writes through to the
935 device and persists there after the session ends. It may therefore only be set
936 when no other control owns the volume of this output.
937 """
938 if not (parent_id := self.protocol_parent_id):
939 # a standalone AirPlay player has no other interface to defer to
940 return True
941 if not (parent_player := self.mass.players.get_player(parent_id)):
942 return True
943 return self._control_routes_to_self(parent_player.volume_control_for_output(self.player_id))
944
945 def release_foreign_mute_latch(self) -> None:
946 """Clear our mute latch when another control owns the mute of this output."""
947 if not self._attr_volume_muted:
948 # nothing latched, so nothing that could silence this stream
949 return
950 if not (parent_id := self.protocol_parent_id):
951 return
952 if not (parent_player := self.mass.players.get_player(parent_id)):
953 return
954 if self._control_routes_to_self(parent_player.mute_control_for_output(self.player_id)):
955 # our own mute, applied through the parent
956 return
957 # The mute belongs to a control that does not own this output (a sibling interface,
958 # the receiver itself, or nothing at all). Our mute is a latch that only an explicit
959 # unmute clears, so leaving it set would report a mute we do not own and turn the
960 # next volume command into a silent one.
961 self._attr_volume_muted = False
962 self.update_state()
963
964 async def on_config_updated(self) -> None:
965 """Handle logic when the player config is updated."""
966 await super().on_config_updated()
967 prov = cast("AirPlayProvider", self.provider)
968 await prov.bridge_manager.evaluate_bridge(self)
969
970 async def on_unload(self) -> None:
971 """Handle logic when the player is unloaded from the Player controller."""
972 await super().on_unload()
973 self.cancel_group_rejoin()
974 if self.stream:
975 # remove this player from the stream session if it is running
976 if self.stream.running and self.stream.session:
977 await self.stream.session.remove_client(self, reason="player unloaded")
978 self.stream = None
979
980 def schedule_group_rejoin(self, candidate_ids: list[str]) -> None:
981 """
982 Schedule a bounded automatic re-join of this player to its still-active group.
983
984 Used when this player's stream process died unexpectedly while it was part
985 of a playing sync group (e.g. the device rode out a network blackout): the
986 player is re-added to the group's live session through the regular
987 late-join path after a short backoff. Any user action on the player (or it
988 joining a session by other means) cancels the re-join; when the group is
989 no longer playing, its membership was changed meanwhile or the device is
990 offline, the re-join is abandoned and the player simply stays idle.
991
992 :param candidate_ids: Player ids that led or shared the group at the
993 moment the stream was lost, used to resolve the re-join target (the
994 leadership may transfer while the backoff runs).
995 """
996 self.cancel_group_rejoin()
997 self.logger.info(
998 "Scheduling automatic re-join of %s to its group after unexpected stream loss",
999 self.display_name,
1000 )
1001 self._rejoin_task = self.mass.create_task(self._group_rejoin_attempts(candidate_ids))
1002
1003 def cancel_group_rejoin(self) -> None:
1004 """Cancel any pending automatic group re-join attempts for this player."""
1005 rejoin_task = self._rejoin_task
1006 # never self-cancel: the re-join attempt itself flows through the same
1007 # session (re)start paths that call this to clear stale schedules. The
1008 # handle also survives such a call, so a later user action can still
1009 # cancel the retry loop between attempts.
1010 if rejoin_task is None or rejoin_task is asyncio.current_task():
1011 return
1012 self._rejoin_task = None
1013 if not rejoin_task.done():
1014 rejoin_task.cancel()
1015
1016 def on_player_media_updated(self) -> None:
1017 """Handle callback when the current media of the player is updated."""
1018 if not self.stream or not self.stream.running:
1019 return
1020 metadata = self.state.current_media
1021 if not metadata:
1022 return
1023 progress = int(metadata.corrected_elapsed_time or 0)
1024 self.mass.create_task(self.stream.send_metadata(progress, metadata))
1025
1026 async def _adopt_device_volume(self, volume: int) -> None:
1027 """
1028 Take over a level the device set itself.
1029
1030 :param volume: The level the device reported.
1031 """
1032 ignored_until = self._volume_reports_ignored_until
1033 await self.volume_set(volume)
1034 # Writing the level back is a volume command like any other and opens the echo
1035 # window, but this one only hands the device its own level: leaving the window
1036 # open would swallow the rest of a volume the user is still turning up. A longer
1037 # window opened while this was in flight (an announcement) still stands.
1038 if self._volume_reports_ignored_until <= time.time() + AIRPLAY_VOLUME_ECHO_GRACE_S:
1039 self._volume_reports_ignored_until = ignored_until
1040
1041 def _control_routes_to_self(self, control: str) -> bool:
1042 """Return True if the given (resolved) control routes to this player."""
1043 if control == self.player_id:
1044 return True
1045 # bridge players riding on this player (e.g. Sendspin-over-AirPlay) forward to us
1046 if control_player := self.mass.players.get_player(control):
1047 return control_player.underlying_player_id == self.player_id
1048 return False
1049
1050 def _get_flags(self) -> int:
1051 # Flags are either present via "sf" or "flags". Taken from pyatv.protocols.airplay.utils.
1052 # We combine flags from both RAOP and AirPlay discovery services because
1053 # LEGACY_PAIRING_BIT (0x200) is typically only in the RAOP service sf field
1054 # (e.g. Apple TV HD), while PIN_REQUIRED (0x8) may only appear in the AirPlay
1055 # service sf/flags field. Using only one source misses the pairing requirement.
1056 flags = 0
1057 for discovery_info in filter(None, [self.raop_discovery_info, self.airplay_discovery_info]):
1058 raw = (
1059 discovery_info.properties.get(b"sf")
1060 or discovery_info.properties.get(b"flags")
1061 or b"0x0"
1062 )
1063 with contextlib.suppress(ValueError, TypeError):
1064 flags |= int(raw, 16)
1065 return flags
1066
1067 def _requires_pin_pairing(self) -> bool:
1068 """
1069 Check if this device requires pairing.
1070
1071 Adapted from pyatv.protocols.airplay.utils.get_pairing_requirement.
1072 """
1073 return bool(self._get_flags() & (LEGACY_PAIRING_BIT | PIN_REQUIRED))
1074
1075 def _get_credentials_key(self, protocol: StreamingProtocol) -> str:
1076 """Get the config key for credentials for given protocol."""
1077 if protocol == StreamingProtocol.RAOP:
1078 return CONF_RAOP_CREDENTIALS
1079 return CONF_AIRPLAY_CREDENTIALS
1080
1081 @property
1082 def _advertised_features(self) -> str | None:
1083 """Return the AirPlay features bitmask the device advertises via mDNS."""
1084 # Prefer the _airplay service's ``features``, falling back to the _raop
1085 # service's ``ft`` when the former is absent (some devices only populate one).
1086 features: str | None = None
1087 if self.airplay_discovery_info:
1088 features = self.airplay_discovery_info.decoded_properties.get(
1089 "features"
1090 ) or self.airplay_discovery_info.decoded_properties.get("ft")
1091 if not features and self.raop_discovery_info:
1092 features = self.raop_discovery_info.decoded_properties.get("ft")
1093 return features
1094
1095 @property
1096 def _is_airplay2_capable(self) -> bool:
1097 """
1098 Return whether this device can stream over AirPlay 2.
1099
1100 Mirrors the feature-bit test the cliairplay binary uses for its own route
1101 selection: a device is AirPlay 2 capable when it exposes the _airplay
1102 service and either advertises the AirPlay 2 feature bits or offers no RAOP
1103 fallback at all (i.e. it is a pure AirPlay 2 receiver).
1104 """
1105 if not self.airplay_discovery_info:
1106 return False
1107 return supports_airplay2(self._advertised_features) or not self.raop_discovery_info
1108
1109 async def _run_streaming_pairing(
1110 self, session: SetupSession, collected: dict[str, ConfigValueType]
1111 ) -> None:
1112 """
1113 Pair the streaming protocol (RAOP or AirPlay 2) and collect the device password.
1114
1115 The two are evaluated independently: a device that is already paired can
1116 still be missing its password (or have had it rejected), which is exactly
1117 the state a receiver ends up in when it gains password protection after
1118 it was set up.
1119
1120 :param session: The setup flow session used to interact with the user.
1121 :param collected: The values collected so far; updated in place.
1122 """
1123 password_collected = await self._run_protocol_pairing(session, collected)
1124 if not password_collected and self.needs_password_setup:
1125 await self._ask_device_password(session)
1126
1127 async def _run_protocol_pairing(
1128 self, session: SetupSession, collected: dict[str, ConfigValueType]
1129 ) -> bool:
1130 """
1131 Pair the streaming protocol (RAOP or AirPlay 2), unless already paired.
1132
1133 When the device requires pairing this runs it, re-offering it as a skippable
1134 step when credentials are already stored (so a re-launched flow can replace a
1135 stale pairing). When the device requires no pairing, any leftover credentials
1136 are cleared: they would keep forcing the pair-verify route, which some
1137 receivers (e.g. HomePods after their password was removed) accept while
1138 refusing to actually output audio. The obtained credentials are added to
1139 ``collected`` under the protocol-specific key.
1140
1141 :param session: The setup flow session used to interact with the user.
1142 :param collected: The values collected so far; updated in place.
1143 :return: Whether the device password was collected as part of the pairing.
1144 """
1145 pin_pairing = self._requires_pin_pairing()
1146 # a password only replaces PIN pairing on the native AirPlay 2 flow
1147 password_pairing = self.password_required and self.protocol == StreamingProtocol.AIRPLAY2
1148 if not (pin_pairing or password_pairing):
1149 for cred_key in (CONF_AIRPLAY_CREDENTIALS, CONF_RAOP_CREDENTIALS):
1150 if self.get_setup_value(cred_key) is not None:
1151 collected[cred_key] = None
1152 return False
1153 already_paired = bool(
1154 self.get_setup_value(CONF_AIRPLAY_CREDENTIALS)
1155 or self.get_setup_value(CONF_RAOP_CREDENTIALS)
1156 )
1157 if already_paired and not await self._offer_optional_pairing(
1158 session, "streaming_repair_offer"
1159 ):
1160 return False
1161
1162 protocol = self.protocol
1163 cred_key = self._get_credentials_key(protocol)
1164 if pin_pairing:
1165 step_id, field_key, field_type, field_format = (
1166 "pair_pin",
1167 CONF_PAIRING_PIN,
1168 ConfigEntryType.PAIRING_CODE,
1169 PAIRING_PIN_FORMAT,
1170 )
1171 else:
1172 step_id, field_key, field_type, field_format = (
1173 "pair_password",
1174 CONF_PAIRING_PASSWORD,
1175 ConfigEntryType.SECURE_STRING,
1176 None,
1177 )
1178
1179 errors: dict[str, str] | None = None
1180 while True:
1181 # Each attempt uses a fresh session: finish_pairing() closes the live
1182 # subprocess/session on completion, so a rejected PIN needs a new one
1183 # (and the device re-shows its PIN).
1184 pairing = await self._prepare_streaming_pairing(protocol, pin_pairing=pin_pairing)
1185 try:
1186 values = await session.form(
1187 [
1188 ConfigEntry(
1189 key=field_key,
1190 type=field_type,
1191 required=True,
1192 category="protocol_generic",
1193 format=field_format,
1194 )
1195 ],
1196 step_id=step_id,
1197 errors=errors,
1198 )
1199 entered_value = str(values[field_key])
1200 credentials = await pairing.finish_pairing(pin=entered_value)
1201 except PlayerCommandFailed as err:
1202 # leave a default-level trace: the flow swallows the error into
1203 # the re-served form, which support logs otherwise never show
1204 self.logger.warning("Pairing with %s failed: %s", self.display_name, err)
1205 errors = {"base": err.translation_key or str(err)}
1206 continue
1207 finally:
1208 # tears down the subprocess on retry, success and abort (cancellation)
1209 await pairing.close()
1210 collected[cred_key] = credentials
1211 if password_pairing:
1212 # The device password authenticates every later stream too (the
1213 # binary's transient leg), so keep it next to the credentials
1214 # instead of discarding it with the setup form.
1215 self._store_device_password(entered_value)
1216 return password_pairing
1217
1218 async def _ask_device_password(self, session: SetupSession) -> None:
1219 """
1220 Ask for the device password and store it, without attempting any pairing.
1221
1222 Covers the devices that have no pairing to do: a legacy RAOP receiver, and
1223 an already paired device whose password is missing or was rejected. There
1224 is no live session to validate the entry against, so a wrong password only
1225 surfaces on the next connect - which marks the player as needing setup again.
1226
1227 :param session: The setup flow session used to interact with the user.
1228 """
1229 values = await session.form(
1230 [
1231 ConfigEntry(
1232 key=CONF_PAIRING_PASSWORD,
1233 type=ConfigEntryType.SECURE_STRING,
1234 required=True,
1235 category="protocol_generic",
1236 )
1237 ],
1238 step_id="pair_password",
1239 )
1240 self._store_device_password(str(values[CONF_PAIRING_PASSWORD]))
1241
1242 async def _offer_optional_pairing(self, session: SetupSession, step_id: str) -> bool:
1243 """
1244 Ask whether to run the offered (optional) pairing now.
1245
1246 :param session: The setup flow session used to interact with the user.
1247 :param step_id: The (i18n) step id describing the offered pairing.
1248 """
1249 values = await session.form(
1250 [
1251 ConfigEntry(
1252 key=CONF_PAIR_NOW,
1253 type=ConfigEntryType.BOOLEAN,
1254 default_value=False,
1255 category="protocol_generic",
1256 )
1257 ],
1258 step_id=step_id,
1259 )
1260 return bool(values[CONF_PAIR_NOW])
1261
1262 async def _prepare_streaming_pairing(
1263 self, protocol: StreamingProtocol, *, pin_pairing: bool
1264 ) -> AirPlayPairing:
1265 """
1266 Build and start a streaming pairing session (the device shows its PIN).
1267
1268 A failure here cannot be recovered by re-prompting the user, so it aborts the
1269 flow; a partially started session is torn down first.
1270
1271 :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1272 :param pin_pairing: Whether the device shows a PIN the user must enter.
1273 """
1274 pairing: AirPlayPairing | None = None
1275 started = False
1276 try:
1277 pairing = self._build_streaming_pairing(protocol)
1278 await pairing.start_pairing_session()
1279 if pin_pairing:
1280 await pairing.start_pin_pairing()
1281 started = True
1282 except Exception as err:
1283 # a failure starting the session (device unreachable, binary/system
1284 # issue, ...) cannot be fixed by re-prompting, so abort with a clear
1285 # reason instead of letting it surface as a generic internal error
1286 self.logger.warning("Could not start AirPlay pairing session: %s", err)
1287 raise AbortFlow("pairing_failed") from err
1288 finally:
1289 if not started and pairing is not None:
1290 await pairing.close()
1291 assert pairing is not None # reached only when started, i.e. a live session
1292 return pairing
1293
1294 def _build_streaming_pairing(self, protocol: StreamingProtocol) -> AirPlayPairing:
1295 """
1296 Build an AirPlayPairing for the given streaming protocol.
1297
1298 :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1299 """
1300 from .pairing import AirPlayPairing # noqa: PLC0415
1301
1302 # For Apple devices pairing always happens on the AirPlay port (7000) even
1303 # when streaming will use RAOP; the RAOP port (5000) is only for streaming.
1304 port: int | None = None
1305 if self.airplay_discovery_info:
1306 port = self.airplay_discovery_info.port or AIRPLAY_DEFAULT_PORT
1307 elif self.raop_discovery_info:
1308 port = self.raop_discovery_info.port or RAOP_DEFAULT_PORT
1309 provider = cast("AirPlayProvider", self.provider)
1310 device_id = provider.dacp_id
1311 pairing_address = self.address
1312 if protocol == StreamingProtocol.AIRPLAY2 and not isinstance(
1313 ipaddress.ip_address(pairing_address), ipaddress.IPv4Address
1314 ):
1315 if self.airplay_discovery_info:
1316 discovered_address = get_primary_ip_address_from_zeroconf(
1317 self.airplay_discovery_info
1318 )
1319 if discovered_address and isinstance(
1320 ipaddress.ip_address(discovered_address), ipaddress.IPv4Address
1321 ):
1322 pairing_address = discovered_address
1323 if not isinstance(ipaddress.ip_address(pairing_address), ipaddress.IPv4Address):
1324 raise PlayerCommandFailed("AirPlay pairing requires an IPv4 device address")
1325 return AirPlayPairing(
1326 address=pairing_address,
1327 name=self.display_name,
1328 protocol=protocol,
1329 logger=self.logger,
1330 port=port,
1331 device_id=device_id,
1332 )
1333
1334 async def _get_session_pcm_format(
1335 self, sync_clients: list[AirPlayPlayer], media: PlayerMedia
1336 ) -> AudioFormat:
1337 """
1338 Select the shared PCM format for a new stream session.
1339
1340 :param sync_clients: All players that will take part in the session.
1341 :param media: The media that is about to be played.
1342 """
1343 queue = self.mass.player_queues.get(media.source_id) if media.source_id else None
1344 queue_item = (
1345 self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
1346 if media.source_id and media.queue_item_id
1347 else None
1348 )
1349 streamdetails = queue_item.streamdetails if queue_item else None
1350 crossfade_enabled = bool(
1351 queue
1352 and media.media_type == MediaType.TRACK
1353 and self.mass.streams.get_crossfade_mode(queue) != CrossfadeMode.DISABLED
1354 )
1355 return await self.mass.streams.audio.select_flow_pcm_format(
1356 self,
1357 start_streamdetails=streamdetails,
1358 crossfade_enabled=crossfade_enabled,
1359 overlay_active=bool(queue and overlay_active(queue)),
1360 fallback_sample_rate=AIRPLAY_PCM_FORMAT.sample_rate,
1361 output_players=sync_clients,
1362 )
1363
1364 def _get_sync_clients(self) -> list[AirPlayPlayer]:
1365 """Get all sync clients for a player."""
1366 sync_clients: list[AirPlayPlayer] = []
1367 # we need to return the player itself too
1368 group_child_ids = {self.player_id}
1369 group_child_ids.update(self.group_members)
1370 for child_id in group_child_ids:
1371 if client := cast("AirPlayPlayer | None", self.mass.players.get_player(child_id)):
1372 sync_clients.append(client)
1373 return sync_clients
1374
1375 async def _group_rejoin_attempts(self, candidate_ids: list[str]) -> None:
1376 """Re-join this player to its group's live session after a bounded backoff."""
1377 max_attempts = len(AIRPLAY_REJOIN_ATTEMPT_DELAYS)
1378 for attempt, delay in enumerate(AIRPLAY_REJOIN_ATTEMPT_DELAYS, start=1):
1379 await asyncio.sleep(delay)
1380 if (
1381 self.group_members
1382 or (self.stream and self.stream.running)
1383 or self.playback_state != PlaybackState.IDLE
1384 # synced into a group outside the original one = deliberate regroup.
1385 # Still pointing at an original candidate is fine: a static group
1386 # keeps the sync membership while only the session lost this player.
1387 or (self.synced_to and self.synced_to not in candidate_ids)
1388 ):
1389 # the player was grouped or repurposed by other means meanwhile
1390 self.logger.debug(
1391 "Automatic group re-join for %s cancelled: player is active again",
1392 self.display_name,
1393 )
1394 return
1395 if not self.available:
1396 # the device is offline: an attempt cannot succeed and the user
1397 # may well have switched it off on purpose
1398 self.logger.debug(
1399 "Automatic group re-join for %s cancelled: player is unavailable",
1400 self.display_name,
1401 )
1402 return
1403 target = self._resolve_rejoin_target(candidate_ids)
1404 if target is None:
1405 # the group may be between sessions (e.g. a track change); keep
1406 # trying until the attempts run out
1407 self.logger.debug(
1408 "Automatic group re-join attempt %d/%d for %s: no playing group found",
1409 attempt,
1410 max_attempts,
1411 self.display_name,
1412 )
1413 continue
1414 # When the sync membership survived the stream loss (a static group,
1415 # where membership is configuration), only the running session needs
1416 # healing; a group command would no-op on the existing membership.
1417 heal_session = (
1418 target.stream.session
1419 if self.player_id in target.group_members and target.stream is not None
1420 else None
1421 )
1422 try:
1423 if heal_session is not None:
1424 await heal_session.add_client(self)
1425 else:
1426 # Join through the target's own set_members: both ends are
1427 # players of this provider, so the join never needs the
1428 # visible-player translations of the controller's grouping
1429 # pipeline - and that pipeline's capability gate reflects
1430 # grouping state that is in flux right after a stream loss,
1431 # so it may silently refuse an internal re-join.
1432 await target.set_members(player_ids_to_add=[self.player_id])
1433 except Exception as err:
1434 self.logger.warning(
1435 "Automatic re-join of %s to group of %s failed (attempt %d/%d): %s",
1436 self.display_name,
1437 target.display_name,
1438 attempt,
1439 max_attempts,
1440 err,
1441 )
1442 continue
1443 # A late-join can also fail without raising (the player then holds
1444 # group membership without a live stream), so verify the session
1445 # actually carries this player before declaring success.
1446 if (
1447 self.stream
1448 and self.stream.running
1449 and self.stream.session
1450 and self in self.stream.session.sync_clients
1451 ):
1452 self.logger.info(
1453 "Automatically re-joined %s to the group of %s after stream loss",
1454 self.display_name,
1455 target.display_name,
1456 )
1457 return
1458 self.logger.warning(
1459 "Automatic re-join of %s did not produce a running stream (attempt %d/%d)",
1460 self.display_name,
1461 attempt,
1462 max_attempts,
1463 )
1464 if heal_session is None:
1465 # undo the group membership this attempt created so a retry (or
1466 # a manual regroup) starts from a clean join
1467 try:
1468 await target.set_members(player_ids_to_remove=[self.player_id])
1469 except Exception as err:
1470 # a failed undo leaves the membership for the next attempt,
1471 # which then heals the session instead of joining anew
1472 self.logger.debug(
1473 "Undo of failed re-join attempt for %s failed: %s",
1474 self.display_name,
1475 err,
1476 )
1477 self.logger.warning(
1478 "Giving up on automatic group re-join for %s after %d attempt(s); "
1479 "the player stays idle",
1480 self.display_name,
1481 max_attempts,
1482 )
1483
1484 def _resolve_rejoin_target(self, candidate_ids: list[str]) -> AirPlayPlayer | None:
1485 """Resolve which player now carries the group's actively playing session."""
1486 for candidate_id in candidate_ids:
1487 candidate = self.mass.players.get_player(candidate_id)
1488 if candidate is None or candidate is self:
1489 continue
1490 if not isinstance(candidate, AirPlayPlayer):
1491 continue
1492 if candidate.synced_to:
1493 # the candidate was absorbed into another group since the loss
1494 # (user intent): never follow the old group's players elsewhere.
1495 # A leadership transfer inside the original group is still found:
1496 # the promoted member is itself one of the candidates.
1497 continue
1498 if not candidate.available:
1499 continue
1500 # only a PLAYING session can absorb a late joiner: a parked (paused)
1501 # session has no live timeline to anchor against
1502 if candidate.playback_state != PlaybackState.PLAYING:
1503 continue
1504 if not (candidate.stream and candidate.stream.running and candidate.stream.session):
1505 continue
1506 return candidate
1507 return None
1508
1509 def _store_device_password(self, password: str) -> None:
1510 """
1511 Persist a device password so every later stream can authenticate with it.
1512
1513 :param password: The plaintext password entered by the user.
1514 """
1515 self.mass.config.set_raw_player_config_value(
1516 self.player_id, CONF_PASSWORD, self.mass.config.encrypt_string(password)
1517 )
1518 # a freshly entered password deserves a clean slate: the reject marker
1519 # would otherwise keep the player in "needs setup" until the next connect
1520 self.set_password_invalid(False)
1521
1522 @property
1523 def _hires_default_enabled(self) -> bool:
1524 """Return the per-device default for the 24-bit toggle."""
1525 return default_hires_enabled(
1526 self.device_info.manufacturer or "", self.device_info.model or ""
1527 )
1528
1529
1530class GenericAirPlayPlayer(AirPlayPlayer):
1531 """AirPlay protocol endpoint without independent device control."""
1532
1533 _attr_type = PlayerType.PROTOCOL
1534