/
/
/
1"""AmpliPi zone player for Music Assistant."""
2
3from __future__ import annotations
4
5import time
6from contextlib import suppress
7from typing import TYPE_CHECKING, cast
8
9from music_assistant_models.enums import PlaybackState, PlayerFeature, PlayerType
10from music_assistant_models.errors import PlayerCommandFailed
11from music_assistant_models.player import DeviceInfo, PlayerMedia, PlayerSource
12from pyamplipi.models import MultiZoneUpdate, SourceUpdate, ZoneUpdate
13
14from music_assistant.models.player import Player
15
16from .constants import (
17 AMPLIPI_API_ERRORS,
18 FREE_SOURCE_INPUTS,
19 INPUT_STREAM_TYPES,
20 SOURCE_DISCONNECTED,
21 SOURCE_ID_STREAM_PREFIX,
22 STREAM_TYPE_LABELS,
23 VOLUME_DB_FLOOR,
24 ZONE_OFF,
25)
26
27if TYPE_CHECKING:
28 from pyamplipi.models import Source, Status, Zone
29 from pyamplipi.models import Stream as AmpliPiStream
30
31 from .provider import AmpliPiPlayerProvider
32
33
34# NOTE: AmpliPi has no native pause for a stream (the backend keeps reading the source URL),
35# so PAUSE is emulated by stopping the stream. Music Assistant captures the queue position
36# before pausing and, on resume, re-issues play_media; in flow mode that re-resolves the
37# stream URL at the resume offset, so playback continues from where it was paused. This
38# relies on the player self-clocking its position (AmpliPi reports none) - see play_media.
39PLAYER_FEATURES = {
40 PlayerFeature.PLAY_MEDIA,
41 PlayerFeature.PAUSE,
42 PlayerFeature.VOLUME_SET,
43 PlayerFeature.VOLUME_MUTE,
44 PlayerFeature.POWER,
45 PlayerFeature.SET_MEMBERS,
46 PlayerFeature.SELECT_SOURCE,
47}
48
49
50class AmpliPiZonePlayer(Player):
51 """Representation of a single AmpliPi zone as a Music Assistant player."""
52
53 def __init__(self, provider: AmpliPiPlayerProvider, zone_id: int) -> None:
54 """Initialize the AmpliPi zone player."""
55 super().__init__(provider, f"{provider.instance_id}_zone_{zone_id}")
56 self._zone_id = zone_id
57 self._source_id: int | None = None
58 # default the player name to the zone's AmpliPi name (e.g. "Living Room"); this is
59 # only a default - a name set on the player in Music Assistant always takes priority
60 zone = next((z for z in provider.status.zones if z.id == zone_id), None)
61 self._attr_name = self._zone_display_name(zone, zone_id)
62 self._attr_type = PlayerType.PLAYER
63 self._attr_supported_features = PLAYER_FEATURES
64 # all zones on the same AmpliPi controller can be grouped with each other
65 self._attr_can_group_with = {provider.instance_id}
66 self._attr_device_info = DeviceInfo(manufacturer="MicroNova", model="AmpliPi Zone")
67
68 @property
69 def zone_id(self) -> int:
70 """Return the AmpliPi zone id of this player."""
71 return self._zone_id
72
73 @property
74 def requires_flow_mode(self) -> bool:
75 """Return if the player requires flow mode (AmpliPi plays a single stream URL)."""
76 return True
77
78 async def volume_set(self, volume_level: int) -> None:
79 """Handle VOLUME_SET command on the player."""
80 await self._prov.api.set_zone(
81 self._zone_id, ZoneUpdate(vol=self._volume_to_db(volume_level))
82 )
83 self._attr_volume_level = volume_level
84 self.update_state()
85
86 async def volume_mute(self, muted: bool) -> None:
87 """Handle VOLUME MUTE command on the player."""
88 await self._prov.api.set_zone(self._zone_id, ZoneUpdate(mute=muted))
89 self._attr_volume_muted = muted
90 self.update_state()
91
92 async def power(self, powered: bool) -> None:
93 """Handle POWER command on the player."""
94 if powered:
95 # AmpliPi mutes a zone while it is disconnected (source -1); unmute it so
96 # that MA's mute state stays consistent and playback is audible.
97 await self._prov.api.set_zone(
98 self._zone_id, ZoneUpdate(source_id=SOURCE_DISCONNECTED, mute=False)
99 )
100 self._attr_powered = True
101 self._attr_volume_muted = False
102 else:
103 # turn off this zone and any zones grouped to it
104 await self._prov.api.set_zones(
105 MultiZoneUpdate(
106 zones=self._member_zone_ids(), update=ZoneUpdate(source_id=ZONE_OFF)
107 )
108 )
109 self._attr_powered = False
110 self._attr_playback_state = PlaybackState.IDLE
111 self._attr_active_source = None
112 self._source_id = None
113 self._dissolve_group()
114 self.update_state()
115
116 async def play(self) -> None:
117 """Handle PLAY command on the player."""
118 # AmpliPi has no native unpause: pause() stopped the stream, so a play/unpause of a
119 # Music Assistant queue must re-resolve playback from the saved resume position
120 # rather than replay the (stale) stream, which would restart from the stream's start.
121 queue = self.mass.player_queues.get(self.player_id)
122 if self._attr_playback_state == PlaybackState.PAUSED and queue is not None and queue.active:
123 await self.mass.player_queues.resume(self.player_id)
124 return
125 if (stream_id := await self._active_stream_id()) is not None:
126 await self._prov.api.play_stream(stream_id)
127 self._attr_playback_state = PlaybackState.PLAYING
128 self.update_state()
129
130 async def stop(self) -> None:
131 """Handle STOP command on the player."""
132 self.mark_stop_called()
133 if (stream_id := await self._active_stream_id()) is not None:
134 await self._prov.api.stop_stream(stream_id)
135 self._attr_playback_state = PlaybackState.IDLE
136 self.update_state()
137
138 async def pause(self) -> None:
139 """Handle PAUSE command on the player."""
140 # AmpliPi cannot truly pause a stream, so we stop it and report PAUSED; Music
141 # Assistant captures the resume position before this runs and, on play, resumes
142 # from it (see the PLAYER_FEATURES note and the play/resume delegation above).
143 if (stream_id := await self._active_stream_id()) is not None:
144 await self._prov.api.stop_stream(stream_id)
145 self._attr_playback_state = PlaybackState.PAUSED
146 self.update_state()
147
148 async def play_media(self, media: PlayerMedia) -> None:
149 """Handle PLAY MEDIA command on the player."""
150 url = await self.mass.streams.resolve_stream_url(self.player_id, media)
151 source = await self._acquire_source()
152 if source is None or source.id is None:
153 raise PlayerCommandFailed("All AmpliPi sources are currently in use.")
154 self._source_id = source.id
155 # Use AmpliPi's internetradio stream type (not the announcement fileplayer):
156 # it is built for continuous HTTP streams and supports reliable play/stop,
157 # where the fileplayer does not. (Pause is emulated via stop - see the
158 # PLAYER_FEATURES note above.)
159 stream_id = await self._prov.ensure_stream(source.id, url)
160 zone_ids = self._member_zone_ids()
161 self.logger.debug(
162 "play_media: stream %s on source %s -> zones %s (%s)",
163 stream_id,
164 source.id,
165 zone_ids,
166 url,
167 )
168 # point the source at our stream, then make its zones EXACTLY this group
169 # (attach our members unmuted; detach any stray zones left on the source),
170 # so playback follows the Music Assistant group and nothing else
171 await self._prov.api.set_source(
172 source.id, SourceUpdate(input=f"{SOURCE_ID_STREAM_PREFIX}{stream_id}")
173 )
174 await self._attach_only_zones(source.id, zone_ids)
175 await self._prov.api.play_stream(stream_id)
176 self._attr_active_source = self.player_id
177 self._attr_current_media = media
178 self._attr_powered = True
179 self._attr_volume_muted = False
180 self._attr_playback_state = PlaybackState.PLAYING
181 # AmpliPi reports no playback position, so we self-clock: start at 0 for this (flow)
182 # stream and let corrected_elapsed_time advance with wall time while PLAYING. Music
183 # Assistant maps this cumulative stream-time back to the queue position - without it
184 # the queue's elapsed time stays pinned at the stream start, so pause/seek-resume
185 # would always jump back to the start of the stream.
186 self._attr_elapsed_time = 0
187 self._attr_elapsed_time_last_updated = time.time()
188 self.update_state()
189
190 async def select_source(self, source: str) -> None:
191 """
192 Handle SELECT_SOURCE command on the player.
193
194 Routes an AmpliPi-side source (a native stream such as its own Spotify Connect /
195 AirPlay, or a physical RCA/line input) to this zone and any grouped members. The
196 source id encodes the AmpliPi stream id as "stream=<id>", which doubles as the value
197 written to the AmpliPi source input.
198
199 :param source: The source id to select, as defined in source_list.
200 """
201 # Setting source.input="stream=<id>" both selects and starts the source: native
202 # streams (e.g. internetradio) auto-play on connect, and RCA/line inputs are
203 # passthrough, so no explicit play_stream is needed.
204 if not source.startswith(SOURCE_ID_STREAM_PREFIX):
205 raise PlayerCommandFailed(f"Unknown source '{source}' for {self.display_name}")
206 amplipi_source = await self._acquire_source()
207 if amplipi_source is None or amplipi_source.id is None:
208 raise PlayerCommandFailed("All AmpliPi sources are currently in use.")
209 self._source_id = amplipi_source.id
210 zone_ids = self._member_zone_ids()
211 # the source id ("stream=<id>") is exactly the AmpliPi source.input value
212 await self._prov.api.set_source(amplipi_source.id, SourceUpdate(input=source))
213 await self._attach_only_zones(amplipi_source.id, zone_ids)
214 self._attr_active_source = source
215 self._attr_current_media = None
216 self._attr_powered = True
217 self._attr_volume_muted = False
218 self._attr_playback_state = PlaybackState.PLAYING
219 self.update_state()
220
221 async def set_members(
222 self,
223 player_ids_to_add: list[str] | None = None,
224 player_ids_to_remove: list[str] | None = None,
225 ) -> None:
226 """Handle SET_MEMBERS command on the player."""
227 affected: set[str] = set()
228 if player_ids_to_add:
229 # the group leader needs a source to share with its members
230 if self._source_id is None:
231 source = await self._acquire_source()
232 if source is None or source.id is None:
233 raise PlayerCommandFailed("All AmpliPi sources are currently in use.")
234 self._source_id = source.id
235 await self._prov.api.set_zone(
236 self._zone_id, ZoneUpdate(source_id=self._source_id, mute=False)
237 )
238 # connect and UNMUTE the added zones: AmpliPi keeps disconnected zones muted,
239 # so without mute=False a newly grouped zone would join silently
240 add_zone_ids = self._zone_ids_for(player_ids_to_add)
241 if add_zone_ids:
242 await self._prov.api.set_zones(
243 MultiZoneUpdate(
244 zones=add_zone_ids,
245 update=ZoneUpdate(source_id=self._source_id, mute=False),
246 )
247 )
248 members = self._attr_group_members or [self.player_id]
249 for player_id in player_ids_to_add:
250 if player_id not in members:
251 members.append(player_id)
252 affected.add(player_id)
253 self._attr_group_members = members
254 if player_ids_to_remove:
255 remove_zone_ids = self._zone_ids_for(player_ids_to_remove)
256 if remove_zone_ids:
257 await self._prov.api.set_zones(
258 MultiZoneUpdate(
259 zones=remove_zone_ids, update=ZoneUpdate(source_id=SOURCE_DISCONNECTED)
260 )
261 )
262 members = [m for m in self._attr_group_members if m not in player_ids_to_remove]
263 self._attr_group_members = [] if members == [self.player_id] else members
264 affected.update(player_ids_to_remove)
265 self.update_state()
266 # refresh the state of the affected member players so their sync state updates
267 for player_id in affected:
268 self.mass.players.trigger_player_update(player_id)
269
270 def set_unavailable(self) -> None:
271 """Mark the player as (temporarily) unavailable."""
272 if not self._attr_available:
273 return
274 self._attr_available = False
275 self.update_state()
276
277 def update_from_status(self, status: Status) -> None:
278 """Update the player state from a polled AmpliPi status object."""
279 zone = next((z for z in status.zones if z.id == self._zone_id), None)
280 if zone is None:
281 self.set_unavailable()
282 return
283 self._attr_available = True
284 # keep the default name in sync with the zone's (possibly renamed) AmpliPi name
285 self._attr_name = self._zone_display_name(zone, self._zone_id)
286 self._build_source_list()
287 self._attr_volume_level = self._db_to_volume(zone.vol)
288 self._attr_volume_muted = zone.mute
289 self._attr_powered = zone.source_id != ZONE_OFF
290 self._source_id = zone.source_id if zone.source_id >= 0 else None
291 if self._source_id is None:
292 # zone is disconnected or powered off: nothing is playing on it
293 self._attr_playback_state = PlaybackState.IDLE
294 self._attr_active_source = None
295 else:
296 # connected to a source: default to MA playback (active_source == player_id),
297 # then, if a user-selectable AmpliPi source (native stream / RCA input) is the
298 # connected input, reflect that instead. Resetting first avoids leaving a stale
299 # external source selected after the input changes back to our own MA stream.
300 self._attr_active_source = self.player_id
301 self._reflect_external_active_source(status)
302 # NOTE: while the zone is connected to a source we deliberately keep the
303 # playback_state set by our play/stop commands and do NOT derive it from the
304 # AmpliPi source state. The internetradio stream play_media uses reports an
305 # unreliable info.state (e.g. "stopped" while audio is actually playing), which
306 # would otherwise continuously desync the player state in the UI.
307 if self._attr_group_members:
308 self._prune_group_members(status)
309 self.update_state()
310
311 # private helpers
312
313 @property
314 def _prov(self) -> AmpliPiPlayerProvider:
315 """Return the (typed) AmpliPi provider for this player."""
316 return cast("AmpliPiPlayerProvider", self.provider)
317
318 def _zone_ids_for(self, player_ids: list[str]) -> list[int]:
319 """Return the AmpliPi zone ids for the given Music Assistant player_ids."""
320 zone_ids: list[int] = []
321 for player_id in player_ids:
322 if (zone_id := self._prov.zone_id_for(player_id)) is not None:
323 zone_ids.append(zone_id)
324 return zone_ids
325
326 def _member_zone_ids(self) -> list[int]:
327 """Return this zone's id plus the zone ids of any grouped members."""
328 zone_ids = [self._zone_id]
329 for player_id in self._attr_group_members:
330 if player_id == self.player_id:
331 continue
332 if (zone_id := self._prov.zone_id_for(player_id)) is not None:
333 zone_ids.append(zone_id)
334 return zone_ids
335
336 async def _attach_only_zones(self, source_id: int, zone_ids: list[int]) -> None:
337 """
338 Attach exactly the given zones to the source, detaching any others.
339
340 Connects the requested zones to the source (unmuted) and disconnects any stray
341 zones still bound to it (e.g. left over from a previous group), so the source's
342 zones match the Music Assistant group exactly.
343
344 :param source_id: The AmpliPi source id to attach the zones to.
345 :param zone_ids: The zone ids that should be playing on the source.
346 """
347 wanted = set(zone_ids)
348 strays = [
349 z.id
350 for z in self._prov.status.zones
351 if z.id is not None
352 and not z.disabled
353 and z.source_id == source_id
354 and z.id not in wanted
355 ]
356 if strays:
357 await self._prov.api.set_zones(
358 MultiZoneUpdate(zones=strays, update=ZoneUpdate(source_id=SOURCE_DISCONNECTED))
359 )
360 await self._prov.api.set_zones(
361 MultiZoneUpdate(zones=zone_ids, update=ZoneUpdate(source_id=source_id, mute=False))
362 )
363
364 def _dissolve_group(self) -> None:
365 """Clear this player's group and refresh any former members."""
366 former_members = [m for m in self._attr_group_members if m != self.player_id]
367 self._attr_group_members = []
368 for player_id in former_members:
369 self.mass.players.trigger_player_update(player_id)
370
371 def _prune_group_members(self, status: Status) -> None:
372 """Drop group members that are no longer connected to this zone's source."""
373 if self._source_id is None:
374 self._dissolve_group()
375 return
376 valid = [self.player_id]
377 for player_id in self._attr_group_members:
378 if player_id == self.player_id:
379 continue
380 zone_id = self._prov.zone_id_for(player_id)
381 zone = next((z for z in status.zones if z.id == zone_id), None)
382 if zone is not None and zone.source_id == self._source_id:
383 valid.append(player_id)
384 self._attr_group_members = [] if valid == [self.player_id] else valid
385
386 async def _acquire_source(self) -> Source | None:
387 """
388 Acquire an AmpliPi source for this zone to play on.
389
390 Reuses the currently connected source if any, otherwise claims a free source.
391 Returns None if all sources are in use (AmpliPi has 4 sources for up to 6+ zones).
392 """
393 status = self._prov.status
394 if self._source_id is not None:
395 if source := next((s for s in status.sources if s.id == self._source_id), None):
396 return source
397 used_source_ids = {z.source_id for z in status.zones if not z.disabled and z.source_id >= 0}
398 for source in status.sources:
399 if source.id is None or source.id in used_source_ids:
400 continue
401 if source.input in FREE_SOURCE_INPUTS:
402 return source
403 # fall back to any source not currently bound to a zone
404 for source in status.sources:
405 if source.id is not None and source.id not in used_source_ids:
406 return source
407 return None
408
409 async def _active_stream_id(self) -> int | None:
410 """Return the id of the stream currently connected to this zone's source, if any."""
411 if self._source_id is None:
412 return None
413 with suppress(*AMPLIPI_API_ERRORS):
414 source = await self._prov.api.get_source(self._source_id)
415 if source.input and source.input.startswith(SOURCE_ID_STREAM_PREFIX):
416 with suppress(ValueError):
417 return int(source.input.removeprefix(SOURCE_ID_STREAM_PREFIX))
418 return None
419
420 def _build_source_list(self) -> None:
421 """
422 Rebuild the selectable source list from the AmpliPi streams.
423
424 Exposes the AmpliPi-side sources the user can route to this zone: physical line
425 inputs (RCA/aux) and native streams (its own Spotify Connect, AirPlay, internet
426 radio, ...). These are routing-only in Music Assistant - the wired input or the
427 owning app drives the audio - so no transport capabilities are advertised.
428 """
429 self._attr_source_list = [
430 PlayerSource(
431 id=f"{SOURCE_ID_STREAM_PREFIX}{stream.id}",
432 name=self._source_label(stream),
433 passive=False,
434 )
435 for stream in self._prov.selectable_streams()
436 ]
437
438 def _reflect_external_active_source(self, status: Status) -> None:
439 """Set active_source to a user-selectable AmpliPi source if one is connected."""
440 source = next((s for s in status.sources if s.id == self._source_id), None)
441 if source is None or not (source.input or "").startswith(SOURCE_ID_STREAM_PREFIX):
442 return
443 if any(source.input == src.id for src in self._attr_source_list):
444 self._attr_active_source = source.input
445
446 @staticmethod
447 def _zone_display_name(zone: Zone | None, zone_id: int) -> str:
448 """
449 Return a sensible default player name for a zone.
450
451 Prefers the zone's configured AmpliPi name; when that is unset (a common case,
452 as users often leave zones unnamed) it falls back to a generic, 1-based label.
453 This is only the default - a name set on the player in Music Assistant wins.
454
455 :param zone: The AmpliPi zone, or None if it is not present in the status.
456 :param zone_id: The AmpliPi zone id, used for the fallback label.
457 """
458 name = (getattr(zone, "name", None) or "").strip()
459 return name or f"AmpliPi Zone {zone_id + 1}"
460
461 @staticmethod
462 def _source_label(stream: AmpliPiStream) -> str:
463 """
464 Return a friendly, disambiguated label for a selectable AmpliPi stream.
465
466 Physical inputs keep their configured name ("Input 1", "Aux"); native streams get a
467 type suffix so same-named endpoints (e.g. Spotify vs AirPlay "AmpliPro 1") are distinct.
468 """
469 if stream.type in INPUT_STREAM_TYPES:
470 return str(stream.name)
471 return f"{stream.name} ({STREAM_TYPE_LABELS.get(stream.type, stream.type)})"
472
473 @staticmethod
474 def _volume_to_db(volume_level: int) -> int:
475 """Map a Music Assistant volume (0-100) to an AmpliPi volume in dB."""
476 return round(VOLUME_DB_FLOOR * (1 - volume_level / 100))
477
478 @staticmethod
479 def _db_to_volume(vol_db: int) -> int:
480 """Map an AmpliPi volume in dB back to a Music Assistant volume (0-100)."""
481 volume = round(100 * (1 - vol_db / VOLUME_DB_FLOOR))
482 return max(0, min(100, volume))
483