music-assistant-server

1"""
2Announcements Mixin for the Player Controller.
3
4Handles playback of announcements (such as TTS messages) on a player: preparing the
5player, handing it the announcement, waiting for it to finish and restoring whatever
6the player was doing before.
7
8This module provides the AnnouncementsMixin class which is inherited by
9PlayerController to add announcement capabilities. The audio itself is rendered and
10served by the streams controller (see controllers/streams/announcements.py).
11"""
12
13from __future__ import annotations
14
15import asyncio
16import logging
17import time
18from math import ceil
19from typing import TYPE_CHECKING, cast
20
21from music_assistant_models.auth import Scope
22from music_assistant_models.constants import PLAYER_CONTROL_NONE
23from music_assistant_models.enums import (
24    MediaType,
25    PlaybackState,
26    PlayerFeature,
27    PlayerType,
28)
29from music_assistant_models.errors import PlayerCommandFailed
30from music_assistant_models.player import PlayerMedia
31
32from music_assistant.constants import (
33    ANNOUNCE_ALERT_FILE,
34    ATTR_ANNOUNCEMENT_IN_PROGRESS,
35    CONF_ENTRY_ANNOUNCE_VOLUME,
36    CONF_ENTRY_ANNOUNCE_VOLUME_MAX,
37    CONF_ENTRY_ANNOUNCE_VOLUME_MIN,
38    CONF_ENTRY_ANNOUNCE_VOLUME_STRATEGY,
39    CONF_ENTRY_TTS_PRE_ANNOUNCE,
40    CONF_PRE_ANNOUNCE_CHIME_URL,
41)
42from music_assistant.controllers.streams.announcements import MAX_CLIP_SECONDS
43from music_assistant.helpers.api import api_command
44from music_assistant.helpers.util import TaskManager, validate_announcement_chime_url
45from music_assistant.models.player import Player
46
47from .constants import PlayerLockPurpose
48from .helpers import AnnounceData, handle_player_command
49
50if TYPE_CHECKING:
51    from collections.abc import Iterator
52
53    from music_assistant import MusicAssistant
54
55
56class AnnouncementsMixin:
57    """
58    Mixin class providing announcement playback for PlayerController.
59
60    Handles:
61    - Resolving the pre-announce chime and announcement volume from configuration
62    - Forwarding a group announcement to its individual members
63    - Native announcement support (on the player itself or a linked protocol)
64    - The fallback implementation for players without native support
65
66    This mixin expects to be mixed with a class that provides:
67    - mass: MusicAssistant instance
68    - logger: logging.Logger instance
69    - get_player(): method to get a player by ID
70    - iter_group_members(): method to iterate the members of a group player
71    - _get_control_target(): method to resolve the player to send a command to
72    - the _handle_cmd_* / cmd_* playback and grouping commands used below
73    """
74
75    # Type hints for attributes provided by the class this mixin is used with
76    if TYPE_CHECKING:
77        mass: MusicAssistant
78        logger: logging.Logger
79        _players: dict[str, Player]
80
81        def get_player(  # noqa: D102
82            self, player_id: str, raise_unavailable: bool = False
83        ) -> Player | None: ...
84
85        def iter_group_members(  # noqa: D102
86            self,
87            group_player: Player,
88            only_powered: bool = False,
89            only_playing: bool = False,
90            active_only: bool = False,
91            exclude_self: bool = True,
92        ) -> Iterator[Player]: ...
93
94        def _get_control_target(
95            self,
96            player: Player,
97            required_feature: PlayerFeature,
98            require_active: bool = False,
99        ) -> Player | None: ...
100
101        async def _wait_for_playback_state(
102            self,
103            player: Player,
104            wanted_state: PlaybackState,
105            timeout: float,
106            minimal_time: float = 0,
107        ) -> None: ...
108
109        async def _handle_play_media(self, player_id: str, media: PlayerMedia) -> None: ...
110
111        async def _handle_cmd_stop(self, player_id: str) -> None: ...
112
113        async def _handle_cmd_volume_set(self, player_id: str, volume_level: int) -> None: ...
114
115        async def _handle_cmd_volume_mute(
116            self, player: Player, mute_control: str, muted: bool
117        ) -> None: ...
118
119        async def _handle_cmd_power(
120            self, player_id: str, powered: bool, skip_auto_play: bool = False
121        ) -> None: ...
122
123        async def _handle_cmd_resume(
124            self, player_id: str, source: str | None = None, media: PlayerMedia | None = None
125        ) -> None: ...
126
127        async def cmd_play(self, player_id: str) -> None: ...  # noqa: D102
128
129        async def cmd_ungroup(self, player_id: str) -> None: ...  # noqa: D102
130
131        async def cmd_set_members(  # noqa: D102
132            self,
133            target_player: str,
134            player_ids_to_add: list[str] | None = None,
135            player_ids_to_remove: list[str] | None = None,
136        ) -> None: ...
137
138    # handle_player_command is typed against PlayerController, which this mixin only
139    # becomes once mixed in; the attributes it needs are declared in the block above.
140    # mypy reports that on the outermost decorator, hence the ignore below.
141    @api_command("players/cmd/play_announcement", required_scope=Scope.PLAYERS_CONTROL)  # type: ignore[type-var]
142    @handle_player_command(lock=PlayerLockPurpose.PLAYBACK)
143    async def play_announcement(
144        self,
145        player_id: str,
146        url: str,
147        pre_announce: bool | None = None,
148        volume_level: int | None = None,
149        pre_announce_url: str | None = None,
150    ) -> None:
151        """
152        Handle playback of an announcement (url) on given player.
153
154        :param player_id: Player ID of the player to handle the command.
155        :param url: URL of the announcement to play.
156        :param pre_announce: Optional bool if pre-announce should be used.
157        :param volume_level: Optional volume level to set for the announcement.
158        :param pre_announce_url: Optional custom URL to use for the pre-announce chime.
159        """
160        player = self.get_player(player_id, True)
161        assert player is not None  # for type checking
162        if not url.startswith("http"):
163            raise PlayerCommandFailed("Only URLs are supported for announcements")
164        if (
165            pre_announce
166            and pre_announce_url
167            and not validate_announcement_chime_url(pre_announce_url)
168        ):
169            raise PlayerCommandFailed("Invalid pre-announce chime URL specified.")
170        # determine pre-announce from (group)player config
171        if pre_announce is None and "tts" in url:
172            conf_pre_announce = self.mass.config.get_raw_player_config_value(
173                player_id,
174                CONF_ENTRY_TTS_PRE_ANNOUNCE.key,
175                CONF_ENTRY_TTS_PRE_ANNOUNCE.default_value,
176            )
177            pre_announce = cast("bool", conf_pre_announce)
178        if pre_announce_url is None:
179            if conf_pre_announce_url := self.mass.config.get_raw_player_config_value(
180                player_id,
181                CONF_PRE_ANNOUNCE_CHIME_URL,
182            ):
183                # player default custom chime url
184                pre_announce_url = cast("str", conf_pre_announce_url)
185            else:
186                # use global default chime url
187                pre_announce_url = ANNOUNCE_ALERT_FILE
188        announce_data = AnnounceData(
189            announcement_url=url,
190            pre_announce=bool(pre_announce),
191            pre_announce_url=pre_announce_url,
192            # filled in below, once we know which player fetches the stream
193            announce_player_id=None,
194        )
195        # Register right away, so the audio is (nearly always fully) rendered by the time
196        # the player is ready for it. The render is shared by everything that consumes
197        # this announcement, including all members of a group.
198        render = self.mass.streams.announcement_renderer.register(player_id, announce_data)
199        try:
200            # mark announcement_in_progress on player
201            player.extra_data[ATTR_ANNOUNCEMENT_IN_PROGRESS] = True
202            # if player type is group with all members supporting announcements,
203            # we forward the request to each individual player
204            if player.state.type == PlayerType.GROUP and (
205                all(
206                    PlayerFeature.PLAY_ANNOUNCEMENT in x.state.supported_features
207                    for x in self.iter_group_members(player)
208                )
209            ):
210                # forward the request to each individual player
211                async with TaskManager(self.mass) as tg:
212                    for group_member in player.state.group_members:
213                        tg.create_task(
214                            self.play_announcement(
215                                group_member,
216                                url=url,
217                                pre_announce=pre_announce,
218                                volume_level=volume_level,
219                                pre_announce_url=pre_announce_url,
220                            )
221                        )
222                return
223            self.logger.info(
224                "Playback announcement to player %s (with pre-announce: %s): %s",
225                player.state.name,
226                pre_announce,
227                url,
228            )
229            # determine if the player has native announcements support
230            # or if any linked protocol has announcement support
231            native_announce_support = False
232            if announce_player := self._get_control_target(
233                player,
234                required_feature=PlayerFeature.PLAY_ANNOUNCEMENT,
235                require_active=False,
236            ):
237                native_announce_support = True
238            else:
239                announce_player = player
240            # create a PlayerMedia object for the announcement so
241            # we can send a regular play-media call downstream
242            announce_data["announce_player_id"] = (
243                announce_player.player_id if native_announce_support else None
244            )
245            announcement = PlayerMedia(
246                uri=self.mass.streams.get_announcement_url(player_id),
247                media_type=MediaType.ANNOUNCEMENT,
248                title="Announcement",
249                custom_data=dict(announce_data),
250            )
251            # handle native announce support (player or linked protocol)
252            if native_announce_support:
253                # hand the url to the player as soon as there is audio to serve from;
254                # its exact length is resolved further downstream, while it plays
255                if not await render.wait_ready():
256                    self.logger.warning(
257                        "Announcement to player %s - no audio available for %s",
258                        player.state.name,
259                        url,
260                    )
261                announcement_volume = self.get_announcement_volume(player_id, volume_level)
262                await announce_player.play_announcement(announcement, announcement_volume)
263                return
264            # use fallback/default implementation
265            await self._play_announcement(player, announcement, volume_level)
266        finally:
267            player.extra_data[ATTR_ANNOUNCEMENT_IN_PROGRESS] = False
268            await self.mass.streams.announcement_renderer.unregister(player_id, render)
269
270    def get_announcement_volume(self, player_id: str, volume_override: int | None) -> int | None:
271        """
272        Get the (player specific) volume for a announcement.
273
274        :param player_id: The player the announcement is played on.
275        :param volume_override: Volume level that overrides the configured strategy.
276        """
277        volume_strategy = self.mass.config.get_raw_player_config_value(
278            player_id,
279            CONF_ENTRY_ANNOUNCE_VOLUME_STRATEGY.key,
280            CONF_ENTRY_ANNOUNCE_VOLUME_STRATEGY.default_value,
281        )
282        volume_strategy_volume = self.mass.config.get_raw_player_config_value(
283            player_id,
284            CONF_ENTRY_ANNOUNCE_VOLUME.key,
285            CONF_ENTRY_ANNOUNCE_VOLUME.default_value,
286        )
287        if volume_strategy == "none":
288            return None
289        volume_level = volume_override
290        if volume_level is None and volume_strategy == "absolute":
291            volume_level = int(cast("float", volume_strategy_volume))
292        elif volume_level is None and volume_strategy == "relative":
293            if (player := self.get_player(player_id)) and player.state.volume_level is not None:
294                volume_level = int(
295                    player.state.volume_level + cast("float", volume_strategy_volume)
296                )
297        elif volume_level is None and volume_strategy == "percentual":
298            if (player := self.get_player(player_id)) and player.state.volume_level is not None:
299                percentual = (player.state.volume_level / 100) * cast(
300                    "float", volume_strategy_volume
301                )
302                volume_level = int(player.state.volume_level + percentual)
303        if volume_level is not None:
304            announce_volume_min = cast(
305                "float",
306                self.mass.config.get_raw_player_config_value(
307                    player_id,
308                    CONF_ENTRY_ANNOUNCE_VOLUME_MIN.key,
309                    CONF_ENTRY_ANNOUNCE_VOLUME_MIN.default_value,
310                ),
311            )
312            volume_level = max(int(announce_volume_min), volume_level)
313            announce_volume_max = cast(
314                "float",
315                self.mass.config.get_raw_player_config_value(
316                    player_id,
317                    CONF_ENTRY_ANNOUNCE_VOLUME_MAX.key,
318                    CONF_ENTRY_ANNOUNCE_VOLUME_MAX.default_value,
319                ),
320            )
321            volume_level = min(int(announce_volume_max), volume_level)
322        return None if volume_level is None else int(volume_level)
323
324    async def _play_announcement(
325        self,
326        player: Player,
327        announcement: PlayerMedia,
328        volume_level: int | None = None,
329    ) -> None:
330        """
331        Handle (default/fallback) implementation of the play announcement feature.
332
333        This default implementation will;
334        - stop playback of the current media (if needed)
335        - power on the player (if needed)
336        - raise the volume a bit
337        - play the announcement (from given url)
338        - wait for the player to finish playing
339        - restore the previous power and volume
340        - restore playback (if needed and if possible)
341
342        This default implementation will only be used if the player
343        (provider) has no native support for the PLAY_ANNOUNCEMENT feature.
344        """
345        prev_state = player.state.playback_state
346        # A player without power control has no power state to restore, so it counts as
347        # powered here - otherwise the restore below would be skipped altogether for it,
348        # leaving the player ungrouped from its (sync)group.
349        prev_power = (
350            player.state.power_control == PLAYER_CONTROL_NONE
351            or bool(player.state.powered)
352            or prev_state != PlaybackState.IDLE
353        )
354        prev_synced_to = player.state.synced_to
355        prev_group = (
356            self.get_player(player.state.active_group) if player.state.active_group else None
357        )
358        prev_source = player.state.active_source
359        prev_media = player.state.current_media
360        prev_media_name = prev_media.title or prev_media.uri if prev_media else None
361        # An announcement is transient: a player that is still busy with an earlier
362        # announcement holds no user content, so there is nothing to restore for it.
363        # The raw media attribute is read here (instead of state.current_media, which
364        # reports the active queue item) since it tells what the device is playing.
365        restore_playback = prev_state == PlaybackState.PLAYING and not (
366            player.current_media is not None
367            and player.current_media.media_type == MediaType.ANNOUNCEMENT
368        )
369        # filled while the players are unmuted and the temporary volume is applied below
370        prev_volumes: dict[str, int] = {}
371        prev_muted: set[str] = set()
372        # everything from here on alters the player state, so the restore in the finally
373        # block must run even when the announcement itself fails halfway through
374        try:
375            await self._prepare_for_announcement(
376                player,
377                volume_level=volume_level,
378                prev_state=prev_state,
379                prev_synced_to=prev_synced_to,
380                prev_group=prev_group,
381                prev_media_name=prev_media_name,
382                prev_volumes=prev_volumes,
383                prev_muted=prev_muted,
384            )
385            # play the announcement
386            self.logger.debug(
387                "Announcement to player %s - playing the announcement on the player...",
388                player.state.name,
389            )
390            render = (
391                self.mass.streams.announcement_renderer.get(
392                    cast("AnnounceData", announcement.custom_data)
393                )
394                if announcement.custom_data
395                else None
396            )
397            if render is not None and not await render.wait_ready():
398                # the render has been filling while the player was prepared above; play on
399                # regardless when it came up empty, so the restore still runs
400                self.logger.warning(
401                    "Announcement to player %s - no audio available for %s",
402                    player.state.name,
403                    announcement.uri,
404                )
405            await self._handle_play_media(player.player_id, announcement)
406            # wait for the player(s) to play
407            await self._wait_for_playback_state(player, PlaybackState.PLAYING, 10, minimal_time=0.1)
408            playback_started = time.time()
409            # wait for the player to stop playing
410            duration = float(announcement.duration) if announcement.duration else None
411            if duration is None and render is not None:
412                # the render knows the exact length of the audio it produced
413                duration = await render.wait_finished()
414                if duration:
415                    announcement.duration = ceil(duration)
416            if duration is None:
417                # length unknown (e.g. the source stalled): wait for the player to report it
418                # finished, bounded by the longest clip an announcement can produce
419                await self._wait_for_playback_state(
420                    player, PlaybackState.IDLE, timeout=MAX_CLIP_SECONDS + 10
421                )
422            else:
423                # waiting for the length above already consumed part of the announcement
424                elapsed = time.time() - playback_started
425                await self._wait_for_playback_state(
426                    player,
427                    PlaybackState.IDLE,
428                    timeout=max(duration + 10 - elapsed, 1),
429                    minimal_time=max(duration + 2 - elapsed, 0),
430                )
431        finally:
432            await self._restore_after_announcement(
433                player,
434                prev_power=prev_power,
435                prev_volumes=prev_volumes,
436                prev_muted=prev_muted,
437                prev_synced_to=prev_synced_to,
438                prev_group=prev_group,
439                prev_source=prev_source,
440                prev_media=prev_media,
441                restore_playback=restore_playback,
442            )
443
444    async def _prepare_for_announcement(
445        self,
446        player: Player,
447        *,
448        volume_level: int | None,
449        prev_state: PlaybackState,
450        prev_synced_to: str | None,
451        prev_group: Player | None,
452        prev_media_name: str | None,
453        prev_volumes: dict[str, int],
454        prev_muted: set[str],
455    ) -> None:
456        """
457        Free up the player for an announcement and apply the temporary announcement volume.
458
459        :param player: The player the announcement will be played on.
460        :param volume_level: Optional volume level override for the announcement.
461        :param prev_state: The playback state the player had before the announcement.
462        :param prev_synced_to: Player ID of the sync leader the player is synced to (if any).
463        :param prev_group: The group player the player is a member of (if any).
464        :param prev_media_name: Name of the media the player was playing (for logging).
465        :param prev_volumes: Mapping that is filled in-place with the previous volume level
466            per player id, so the caller can restore the volumes even if this call fails.
467        :param prev_muted: Set that is filled in-place with the ids of the players that were
468            muted, so the caller can restore the mute state even if this call fails.
469        """
470        if prev_synced_to:
471            # ungroup player if its currently synced
472            self.logger.debug(
473                "Announcement to player %s - ungrouping player from %s...",
474                player.state.name,
475                prev_synced_to,
476            )
477            await self.cmd_ungroup(player.player_id)
478        elif prev_group:
479            # if the player is part of a group player, we need to ungroup it
480            if PlayerFeature.SET_MEMBERS in prev_group.supported_features:
481                self.logger.debug(
482                    "Announcement to player %s - ungrouping from group player %s...",
483                    player.state.name,
484                    prev_group.display_name,
485                )
486                await prev_group.set_members(player_ids_to_remove=[player.player_id])
487            else:
488                # if the player is part of a group player that does not support ungrouping,
489                # we need to power off the groupplayer instead
490                self.logger.debug(
491                    "Announcement to player %s - turning off group player %s...",
492                    player.state.name,
493                    prev_group.display_name,
494                )
495                await self._handle_cmd_power(prev_group.player_id, False)
496        elif prev_state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
497            # normal/standalone player: stop player if its currently playing
498            self.logger.debug(
499                "Announcement to player %s - stop existing content (%s)...",
500                player.state.name,
501                prev_media_name,
502            )
503            await self._handle_cmd_stop(player.player_id)
504            # wait for the player to stop
505            await self._wait_for_playback_state(player, PlaybackState.IDLE, 10, 0.4)
506        # unmute and adjust volume if needed
507        # in case of a (sync) group, we need to do this for all child players
508        async with TaskManager(self.mass) as tg:
509            for volume_player_id in player.state.group_members or (player.player_id,):
510                if not (volume_player := self.get_player(volume_player_id)):
511                    continue
512                # catch any players that have a different source active
513                if (
514                    volume_player.state.active_source
515                    not in (
516                        player.state.active_source,
517                        volume_player.player_id,
518                        None,
519                    )
520                    and volume_player.state.playback_state == PlaybackState.PLAYING
521                ):
522                    self.logger.warning(
523                        "Detected announcement to playergroup %s while group member %s is playing "
524                        "other content, this may lead to unexpected behavior.",
525                        player.state.name,
526                        volume_player.state.name,
527                    )
528                    tg.create_task(self._handle_cmd_stop(volume_player.player_id))
529                tg.create_task(
530                    self._unmute_and_set_announcement_volume(
531                        volume_player, volume_level, prev_volumes, prev_muted
532                    )
533                )
534
535    async def _restore_after_announcement(
536        self,
537        player: Player,
538        *,
539        prev_power: bool,
540        prev_volumes: dict[str, int],
541        prev_muted: set[str],
542        prev_synced_to: str | None,
543        prev_group: Player | None,
544        prev_source: str | None,
545        prev_media: PlayerMedia | None,
546        restore_playback: bool,
547    ) -> None:
548        """
549        Restore the player state that was captured before an announcement was played.
550
551        This also runs when the announcement failed halfway through, so a failing restore
552        step is logged instead of raised: it may never mask the error that caused it.
553
554        :param player: The player the announcement was played on.
555        :param prev_power: Whether the player was powered before the announcement.
556        :param prev_volumes: The previous volume level per player id.
557        :param prev_muted: The ids of the players that were muted before the announcement.
558        :param prev_synced_to: Player ID of the sync leader the player was synced to (if any).
559        :param prev_group: The group player the player was a member of (if any).
560        :param prev_source: The source that was active before the announcement.
561        :param prev_media: The media that was loaded before the announcement.
562        :param restore_playback: Whether playback needs to be resumed.
563        """
564        self.logger.debug(
565            "Announcement to player %s - restore previous state...", player.state.name
566        )
567        # restore volume
568        async with TaskManager(self.mass) as tg:
569            for volume_player_id, prev_volume in prev_volumes.items():
570                tg.create_task(self._handle_cmd_volume_set(volume_player_id, prev_volume))
571        # restore mute after the volume, because setting a volume unmutes the player again
572        async with TaskManager(self.mass) as tg:
573            for muted_player_id in prev_muted:
574                if not (muted_player := self.get_player(muted_player_id)):
575                    continue
576                tg.create_task(self._set_announcement_mute(muted_player, True))
577        await asyncio.sleep(0.2)
578        try:
579            # either power off the player or resume playing
580            if not prev_power:
581                # prev_power is always True for a player without power control,
582                # so there is an actual power control to switch off here
583                self.logger.debug(
584                    "Announcement to player %s - turning player off again...", player.state.name
585                )
586                await self._handle_cmd_power(player.player_id, False)
587                return
588            if prev_synced_to:
589                self.logger.debug(
590                    "Announcement to player %s - syncing back to %s...",
591                    player.state.name,
592                    prev_synced_to,
593                )
594                await self.cmd_set_members(prev_synced_to, player_ids_to_add=[player.player_id])
595            elif prev_group:
596                if PlayerFeature.SET_MEMBERS in prev_group.supported_features:
597                    self.logger.debug(
598                        "Announcement to player %s - grouping back to group player %s...",
599                        player.state.name,
600                        prev_group.display_name,
601                    )
602                    await prev_group.set_members(player_ids_to_add=[player.player_id])
603                elif restore_playback:
604                    # if the player is part of a group player that does not support set_members,
605                    # we need to restart the groupplayer
606                    self.logger.debug(
607                        "Announcement to player %s - restarting playback on group player %s...",
608                        player.state.name,
609                        prev_group.display_name,
610                    )
611                    await self.cmd_play(prev_group.player_id)
612            elif restore_playback:
613                # player was playing something before the announcement - try to resume that here
614                await self._handle_cmd_resume(player.player_id, prev_source, prev_media)
615        except Exception as err:
616            # deliberately broad: set_members is a raw provider call that is not wrapped
617            # into a MusicAssistantError, so it can surface anything its client library
618            # raises. CancelledError is a BaseException and still propagates.
619            self.logger.warning(
620                "Announcement to player %s - restoring the previous state failed: %s",
621                player.state.name,
622                err,
623            )
624
625    async def _unmute_and_set_announcement_volume(
626        self,
627        volume_player: Player,
628        volume_level: int | None,
629        prev_volumes: dict[str, int],
630        prev_muted: set[str],
631    ) -> None:
632        """
633        Make a single player ready to be heard: unmute it and set the announcement volume.
634
635        :param volume_player: The player to prepare.
636        :param volume_level: Optional volume level override for the announcement.
637        :param prev_volumes: Mapping that is filled in-place with the previous volume level
638            per player id.
639        :param prev_muted: Set that is filled in-place with the ids of the players that
640            were muted.
641        """
642        if volume_player.state.volume_muted:
643            # an announcement is always meant to be heard, so a deliberate mute is lifted
644            # for its duration. this happens before the volume is read below, since a
645            # fake mute control parks the player at volume 0 to mute it.
646            prev_muted.add(volume_player.player_id)
647            await self._set_announcement_mute(volume_player, False)
648        if volume_player.state.volume_control == PLAYER_CONTROL_NONE:
649            return
650        if (prev_volume := volume_player.state.volume_level) is None:
651            return
652        announcement_volume = self.get_announcement_volume(volume_player.player_id, volume_level)
653        # get_announcement_volume already returns None when the volume must be left
654        # alone, so any number it does return is the volume to announce at - including
655        # 0, which must not be mistaken for 'no volume configured'
656        if announcement_volume is None or announcement_volume == prev_volume:
657            return
658        prev_volumes[volume_player.player_id] = prev_volume
659        self.logger.debug(
660            "Announcement to player %s - setting temporary volume (%s)...",
661            volume_player.state.name,
662            announcement_volume,
663        )
664        await self._handle_cmd_volume_set(volume_player.player_id, announcement_volume)
665
666    async def _set_announcement_mute(self, player: Player, muted: bool) -> None:
667        """
668        Mute or unmute a player for the duration of an announcement.
669
670        :param player: The player to mute or unmute.
671        :param muted: bool if the player should be muted.
672        """
673        # the internal handler is used instead of cmd_volume_mute so the player keeps the
674        # mute lock it earned as a group member: the public command clears that lock on
675        # unmute and the player may well be ungrouped by the time it is muted back.
676        mute_control = player.mute_control
677        if mute_control == PLAYER_CONTROL_NONE:
678            return
679        await self._handle_cmd_volume_mute(player, mute_control, muted)
680