music-assistant-server

1"""Squeezelite Player implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import statistics
7import struct
8import time
9from collections import deque
10from collections.abc import Iterator
11from typing import TYPE_CHECKING, cast
12
13from aioslimproto.models import EventType as SlimEventType
14from aioslimproto.models import PlayerState as SlimPlayerState
15from aioslimproto.models import Preset as SlimPreset
16from aioslimproto.models import SlimEvent
17from aioslimproto.models import VisualisationType as SlimVisualisationType
18from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
19from music_assistant_models.enums import (
20    ConfigEntryType,
21    IdentifierType,
22    MediaType,
23    PlaybackState,
24    PlayerFeature,
25    PlayerType,
26    RepeatMode,
27)
28from music_assistant_models.errors import InvalidCommand, MusicAssistantError
29
30from music_assistant.constants import (
31    CONF_ENTRY_HTTP_PROFILE_FORCED_2,
32    CONF_ENTRY_PREFER_WAV_FOR_LIVE_SOURCES_DEFAULT_ENABLED,
33    CONF_ENTRY_SYNC_ADJUST,
34    CONF_OUTPUT_CODEC,
35    CONF_PREFER_WAV_FOR_LIVE_SOURCES,
36    VERBOSE_LOG_LEVEL,
37)
38from music_assistant.controllers.streams.audio_processing import get_media_session_id
39from music_assistant.helpers.audio import get_mime_type
40from music_assistant.helpers.util import TaskManager
41from music_assistant.models.player import (
42    POSITION_JUMP_THRESHOLD,
43    DeviceInfo,
44    Player,
45    PlayerMedia,
46)
47
48from .constants import (
49    CONF_ENTRY_DISPLAY,
50    CONF_ENTRY_VISUALIZATION,
51    DEFAULT_PLAYER_VOLUME,
52    DEVIATION_JUMP_IGNORE,
53    MAX_SKIP_AHEAD_MS,
54    MIN_DEVIATION_ADJUST,
55    MIN_REQ_PLAYPOINTS,
56    REPEATMODE_MAP,
57    STATE_MAP,
58    SyncPlayPoint,
59)
60from .multi_client_stream import MultiClientStream
61
62if TYPE_CHECKING:
63    from aioslimproto.client import SlimClient
64
65    from .provider import SqueezelitePlayerProvider
66
67
68CACHE_CATEGORY_PREV_STATE = (
69    1  # category for caching previous player state (bumped to invalidate old format)
70)
71
72PROTOCOL_ONLY_MODELS = (
73    # Device models where slimproto is only a secondary protocol on a device with
74    # its own (native) identity: WiiM/LinkPlay devices (ModelName=WiiM Player) and
75    # the LMS bridge tools that expose AirPlay/Chromecast/UPnP devices as
76    # squeezelite players. These register as PlayerType.PROTOCOL and get linked
77    # to the device's visible player.
78    "wiim",
79    "raopbridge",
80    "castbridge",
81    "upnpbridge",
82)
83
84
85def is_protocol_only_device(device_model: str) -> bool:
86    """Return True if the device uses squeezelite as a secondary protocol only."""
87    device_model_lower = device_model.lower()
88    return any(model.lower() in device_model_lower for model in PROTOCOL_ONLY_MODELS)
89
90
91class SqueezelitePlayer(Player):
92    """Squeezelite Player implementation."""
93
94    def __init__(
95        self,
96        provider: SqueezelitePlayerProvider,
97        player_id: str,
98        client: SlimClient,
99    ) -> None:
100        """Initialize the Squeezelite Player."""
101        super().__init__(provider, player_id)
102        self.client = client
103        self._provider: SqueezelitePlayerProvider = provider
104        # Set static player attributes
105        self._attr_supported_features = {
106            PlayerFeature.PLAY_MEDIA,
107            PlayerFeature.SET_MEMBERS,
108            PlayerFeature.MULTI_DEVICE_DSP,
109            PlayerFeature.VOLUME_SET,
110            PlayerFeature.PAUSE,
111            PlayerFeature.ENQUEUE,
112            PlayerFeature.GAPLESS_PLAYBACK,
113        }
114        # Protocol players are powered on/off with the stream and expose no power
115        # control; full players get native power support (slimproto power can e.g.
116        # drive a GPIO/script wired to an amplifier).
117        if is_protocol_only_device(client.device_model):
118            self._attr_type = PlayerType.PROTOCOL
119        else:
120            self._attr_type = PlayerType.PLAYER
121            self._attr_supported_features.add(PlayerFeature.POWER)
122        self._attr_can_group_with = {provider.instance_id}
123        max_sr = int(self.client.max_sample_rate)
124        self._attr_supported_sample_rates = [
125            (sr, bd)
126            for sr in (44100, 48000, 88200, 96000, 176400, 192000)
127            if sr <= max_sr
128            for bd in (16, 24)
129        ]
130        self.multi_client_stream: MultiClientStream | None = None
131        self._sync_playpoints: deque[SyncPlayPoint] = deque(maxlen=MIN_REQ_PLAYPOINTS)
132        self._do_not_resync_before: float = 0.0
133        self._audio_source_active: bool = False
134        self._low_latency_stream: bool = False
135        # TEMP: patch slimclient send_strm to adjust buffer thresholds
136        # this can be removed when we did a new release of aioslimproto with this change
137        # after this has been tested in beta for a while
138        client._send_strm = lambda *args, **kwargs: _patched_send_strm(
139            client, self, *args, **kwargs
140        )
141
142    async def on_config_updated(self) -> None:
143        """Handle logic when the PlayerConfig is first loaded or updated."""
144        # set presets and display
145        await self._set_preset_items()
146        await self._set_display()
147
148    async def setup(self) -> None:
149        """Set up the player."""
150        player_id = self.client.player_id
151        self.logger.info("Player %s connected", self.client.name or player_id)
152        # update all dynamic attributes
153        self.update_attributes()
154        # restore volume state
155        if last_state := await self.mass.cache.get(
156            key=player_id, provider=self.provider.instance_id, category=CACHE_CATEGORY_PREV_STATE
157        ):
158            init_muted = last_state[0]
159            init_volume = last_state[1]
160        else:
161            init_muted = False
162            init_volume = DEFAULT_PLAYER_VOLUME
163        await self.client.stop()
164        await self.client.mute(init_muted)
165        await self.client.volume_set(init_volume)
166        await self.mass.players.register_or_update(self)
167
168    async def get_config_entries(self) -> list[ConfigEntry]:
169        """Return all (provider/player specific) Config Entries for the player."""
170        base_entries = await super().get_config_entries()
171        # create preset entries (for players that support it)
172        presets = []
173        async for playlist in self.mass.music.playlists.iter_library_items(True):
174            presets.append(ConfigValueOption(playlist.uri, title=playlist.name))
175        async for radio in self.mass.music.radio.iter_library_items(True):
176            presets.append(ConfigValueOption(radio.uri, title=radio.name))
177        preset_count = 10
178        preset_entries = [
179            ConfigEntry(
180                key=f"preset_{index}",
181                type=ConfigEntryType.STRING,
182                options=presets,
183                translation_key="preset",
184                translation_params=[str(index)],
185                category="presets",
186                required=False,
187            )
188            for index in range(1, preset_count + 1)
189        ]
190        return [
191            *base_entries,
192            *preset_entries,
193            CONF_ENTRY_SYNC_ADJUST,
194            CONF_ENTRY_DISPLAY,
195            CONF_ENTRY_VISUALIZATION,
196            CONF_ENTRY_HTTP_PROFILE_FORCED_2,
197            CONF_ENTRY_PREFER_WAV_FOR_LIVE_SOURCES_DEFAULT_ENABLED,
198        ]
199
200    async def volume_set(self, volume_level: int) -> None:
201        """Handle VOLUME_SET command on the player."""
202        await self.client.volume_set(volume_level)
203        # store last state in cache
204        await self.mass.cache.set(
205            key=self.player_id,
206            data=[self.client.muted, volume_level],
207            provider=self.provider.instance_id,
208            category=CACHE_CATEGORY_PREV_STATE,
209        )
210
211    async def volume_mute(self, muted: bool) -> None:
212        """Handle VOLUME MUTE command on the player."""
213        await self.client.mute(muted)
214        # store last state in cache
215        await self.mass.cache.set(
216            key=self.player_id,
217            data=[muted, self.client.volume_level],
218            provider=self.provider.instance_id,
219            category=CACHE_CATEGORY_PREV_STATE,
220        )
221
222    async def power(self, powered: bool) -> None:
223        """Handle POWER command on the player."""
224        async with TaskManager(self.mass) as tg:
225            for client in self._get_sync_clients():
226                tg.create_task(client.power(powered))
227
228    async def stop(self) -> None:
229        """Handle STOP command on the player."""
230        self._audio_source_active = False
231        # Clean up any existing multi-client stream
232        if self.multi_client_stream is not None:
233            await self.multi_client_stream.stop()
234            self.multi_client_stream = None
235        async with TaskManager(self.mass) as tg:
236            for client in self._get_sync_clients():
237                if self.type == PlayerType.PROTOCOL:
238                    tg.create_task(client.power(False))
239                else:
240                    tg.create_task(client.stop())
241        self.update_state()
242
243    async def play(self) -> None:
244        """Handle PLAY command on the player."""
245        async with TaskManager(self.mass) as tg:
246            for client in self._get_sync_clients():
247                tg.create_task(client.play())
248
249    async def pause(self) -> None:
250        """Handle PAUSE command on the player."""
251        async with TaskManager(self.mass) as tg:
252            for client in self._get_sync_clients():
253                tg.create_task(client.pause())
254
255    async def play_media(self, media: PlayerMedia) -> None:
256        """Handle PLAY MEDIA on the player."""
257        if self.synced_to:
258            msg = "A synced player cannot receive play commands directly"
259            raise InvalidCommand(msg)
260
261        # Clean up any existing multi-client stream before starting a new one
262        if self.multi_client_stream is not None:
263            await self.multi_client_stream.stop()
264            self.multi_client_stream = None
265
266        # Clear next media item during announcements to prevent playing the
267        # next enqueued track after it finishes.
268        if media.media_type == MediaType.ANNOUNCEMENT:
269            self.client._next_media = None
270
271        if not self.group_members:
272            # Simple, single-player playback
273            stream_url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
274            await self._handle_play_url_for_slimplayer(
275                self.client,
276                url=stream_url,
277                media=media,
278                send_flush=True,
279                auto_play=False,
280            )
281            return
282
283        # this is a syncgroup; we need to handle this with a multi client stream.
284        # pick the master flow format honoring the leader's flow-mode-sample-rate
285        # setting and supported rates. anchor on the first track when its streamdetails
286        # are already resolved so smart/bit-perfect modes start at the right rate.
287        start_queue_item = (
288            self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
289            if media.source_id and media.queue_item_id
290            else None
291        )
292        # crossfade is a queue-scoped setting; read it from the queue being played (source_id),
293        # which can differ from this player when playing on behalf of a group/linked queue
294        queue = self.mass.player_queues.get(media.source_id) if media.source_id else None
295        crossfade_enabled = bool(
296            queue and queue.crossfade_enabled and media.media_type == MediaType.TRACK
297        )
298        master_audio_format = await self.mass.streams.audio.select_flow_pcm_format(
299            self,
300            start_streamdetails=start_queue_item.streamdetails if start_queue_item else None,
301            crossfade_enabled=crossfade_enabled,
302            overlay_active=bool(queue and queue.overlay_enabled and queue.overlay_source),
303        )
304
305        # select audio source, we force flow mode
306        # because multi-client streaming does not support enqueueing
307        audio_source = self.mass.streams.get_stream(
308            media, master_audio_format, player_id=self.player_id, force_flow_mode=True
309        )
310
311        # start the stream task
312        self.multi_client_stream = stream = MultiClientStream(
313            audio_source=audio_source,
314            audio_format=master_audio_format,
315            queue_id=media.source_id,
316            session_id=get_media_session_id(media),
317        )
318        base_url = f"{self.mass.streams.base_url}/slimproto/multi?player_id={self.player_id}"
319
320        # Count how many clients will connect
321        expected_clients = len(list(self._get_sync_clients()))
322        stream.expected_clients = expected_clients
323
324        # forward to downstream play_media commands
325        # Per-member output_codec: classic Squeezeboxes silently fail on fixed flac in sync (#5506).
326        async with TaskManager(self.mass) as tg:
327            for slimplayer in self._get_sync_clients():
328                member_codec = self._get_member_output_codec(slimplayer.player_id, media)
329                url = f"{base_url}&fmt={member_codec}&child_player_id={slimplayer.player_id}"
330                tg.create_task(
331                    self._handle_play_url_for_slimplayer(
332                        slimplayer,
333                        url=url,
334                        media=media,
335                        send_flush=True,
336                        auto_play=False,
337                        is_group_playback=True,
338                    )
339                )
340
341    async def enqueue_next_media(self, media: PlayerMedia) -> None:
342        """Handle enqueuing next media item."""
343        stream_url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
344        await self._handle_play_url_for_slimplayer(
345            self.client,
346            url=stream_url,
347            media=media,
348            enqueue=True,
349            send_flush=False,
350            auto_play=True,
351        )
352
353    async def set_members(
354        self,
355        player_ids_to_add: list[str] | None = None,
356        player_ids_to_remove: list[str] | None = None,
357    ) -> None:
358        """Handle SET_MEMBERS command on the player."""
359        if self.synced_to:
360            # this should not happen, but guard anyways
361            raise InvalidCommand("Player is synced, cannot set members")
362        if not player_ids_to_add and not player_ids_to_remove:
363            # nothing to do
364            return
365
366        # handle removals first
367        if player_ids_to_remove:
368            for sync_client in self._get_sync_clients():
369                if sync_client.player_id in player_ids_to_remove:
370                    if sync_client.player_id in self._attr_group_members:
371                        # remove child from the group
372                        self._attr_group_members.remove(sync_client.player_id)
373                        if sync_client.state != SlimPlayerState.STOPPED:
374                            # stop the player if it is playing
375                            await sync_client.stop()
376            # if no children remain, remove ourselves from the list too
377            remaining = [x for x in self._attr_group_members if x != self.player_id]
378            if not remaining:
379                self._attr_group_members.clear()
380
381        # handle additions
382        players_added = False
383        for player_id in player_ids_to_add or []:
384            if player_id == self.player_id or player_id in self.group_members:
385                # nothing to do: player is already part of the group
386                continue
387            child_player = cast("SqueezelitePlayer | None", self.mass.players.get_player(player_id))
388            if not child_player:
389                # should not happen, but guard against it
390                continue
391            if child_player.state != SlimPlayerState.STOPPED:
392                # stop the player if it is already playing something else
393                await child_player.stop()
394            self._attr_group_members.append(player_id)
395            players_added = True
396        # ensure the sync leader (self) is the first item in group_members
397        if self._attr_group_members and self.player_id not in self._attr_group_members:
398            self._attr_group_members.insert(0, self.player_id)
399
400        # always update the state after modifying group members
401        self.update_state()
402
403        if (
404            (players_added or player_ids_to_remove)
405            and self.state.current_media
406            and self._attr_playback_state == PlaybackState.PLAYING
407        ):
408            # restart stream session if it was already playing
409            # for now, we dont support late joining into an existing stream
410            self.mass.create_task(self.mass.players.cmd_resume(self.player_id))
411
412    def handle_slim_event(self, event: SlimEvent) -> None:
413        """Handle player event from slimproto server."""
414        if event.type == SlimEventType.PLAYER_BUFFER_READY:
415            self.mass.create_task(self._handle_buffer_ready())
416            return
417
418        if event.type == SlimEventType.PLAYER_HEARTBEAT:
419            self._handle_player_heartbeat()
420            return
421
422        if event.type in (SlimEventType.PLAYER_BTN_EVENT, SlimEventType.PLAYER_CLI_EVENT):
423            self.mass.create_task(self._handle_player_cli_event(event))
424            return
425
426        # all other: update attributes and update state
427        self.update_attributes()
428        self.update_state()
429
430    def update_attributes(self) -> None:
431        """Update player attributes from slim player."""
432        # Update player state from slim player
433        self._attr_available = self.client.connected
434        self._attr_name = self.client.name
435        self._attr_powered = self.client.powered
436        old_state = self._attr_playback_state
437        self._attr_playback_state = STATE_MAP[self.client.state]
438        self._attr_volume_level = self.client.volume_level
439        self._attr_volume_muted = self.client.muted
440        self._attr_device_info = DeviceInfo(
441            model=self.client.device_model,
442            manufacturer=self.client.device_type,
443        )
444        self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, self.client.device_address)
445        # player_id is the MAC address in slimproto
446        self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, self.client.player_id)
447        if (
448            old_state != PlaybackState.PLAYING
449            and self._attr_playback_state == PlaybackState.PLAYING
450        ):
451            # Invalidate elapsed time interpolation to avoid jumps when resuming from pause/stop
452            # We need this because some players (e.g. WiiM) keep sending increasing elapsed time
453            self._attr_elapsed_time_last_updated = time.time()
454        # Update current media if available
455        if self.client.current_media and (metadata := self.client.current_media.metadata):
456            self._attr_current_media = PlayerMedia(
457                uri=metadata.get("item_id"),
458                title=metadata.get("title"),
459                album=metadata.get("album"),
460                artist=metadata.get("artist"),
461                image_url=metadata.get("image_url"),
462                duration=metadata.get("duration"),
463                source_id=metadata.get("source_id"),
464                queue_item_id=metadata.get("queue_item_id"),
465            )
466        else:
467            self._attr_current_media = None
468
469    async def _handle_play_url_for_slimplayer(
470        self,
471        slimplayer: SlimClient,
472        url: str,
473        media: PlayerMedia,
474        enqueue: bool = False,
475        send_flush: bool = True,
476        auto_play: bool = False,
477        is_group_playback: bool = False,
478    ) -> None:
479        """Handle playback of an url on slimproto player(s)."""
480        metadata = {
481            "item_id": media.uri,
482            "title": media.title,
483            "album": media.album,
484            "artist": media.artist,
485            "image_url": media.image_url,
486            "duration": media.stream_duration or media.duration,
487            "source_id": media.source_id,
488            "queue_item_id": media.queue_item_id,
489        }
490        queue = None
491        if media.source_id and (queue := self.mass.player_queues.get(media.source_id)):
492            self.extra_data["playlist repeat"] = REPEATMODE_MAP[queue.repeat_mode]
493            self.extra_data["playlist shuffle"] = int(queue.shuffle_enabled)
494        audio_source_active = media.media_type == MediaType.AUDIO_SOURCE
495        low_latency_stream = audio_source_active or media.media_type == MediaType.RADIO
496        # set the flags on the player that owns the slimclient (may differ from self
497        # during group playback where self is the leader but slimplayer is a member)
498        target_player = self.mass.players.get_player(slimplayer.player_id)
499        if isinstance(target_player, SqueezelitePlayer):
500            target_player._audio_source_active = audio_source_active
501            target_player._low_latency_stream = low_latency_stream
502        await slimplayer.play_url(
503            url=url,
504            mime_type=get_mime_type(url.rsplit(".", maxsplit=1)[-1].split("?", maxsplit=1)[0]),
505            metadata=metadata,
506            enqueue=enqueue,
507            send_flush=send_flush,
508            # if autoplay=False playback will not start automatically
509            # instead 'buffer ready' will be called when the buffer is full
510            # to coordinate a start of multiple synced players
511            autostart=auto_play,
512        )
513        # TODO: When we implement server clock sync, we can remove the pause here
514        # and rely on unpause_at + HEADROOM in the buffer_ready handler. LMS
515        # also does NOT use an explicit pause. For now, we pause here to avoid
516        # WiiM devices starting playback too early, causing huge initial drift.
517        if is_group_playback:
518            await slimplayer.pause()
519        # if queue is set to single track repeat,
520        # immediately set this track as the next
521        # this prevents race conditions with super short audio clips (on single repeat)
522        # https://github.com/music-assistant/hass-music-assistant/issues/2059
523        if queue and queue.repeat_mode == RepeatMode.ONE:
524            self.mass.call_later(
525                0.2,
526                slimplayer.play_url(
527                    url=url,
528                    mime_type=get_mime_type(
529                        url.rsplit(".", maxsplit=1)[-1].split("?", maxsplit=1)[0]
530                    ),
531                    metadata=metadata,
532                    enqueue=True,
533                    send_flush=False,
534                    autostart=True,
535                ),
536            )
537
538    def _handle_player_heartbeat(self) -> None:
539        """Process SlimClient elapsed_time update."""
540        if self._attr_playback_state != PlaybackState.PLAYING:
541            # ignore server heartbeats when not playing
542            # Some players keep sending heartbeat with increasing elapsed time
543            # even when paused (e.g. WiiM)
544            return
545        self._attr_elapsed_time = self.client.elapsed_seconds
546        self._attr_elapsed_time_last_updated = time.time()
547        # only involve the state machine when the reported position diverged (e.g. buffering/seek)
548        published_position = self.state.corrected_elapsed_time
549        if (
550            published_position is None
551            or abs(published_position - self.client.elapsed_seconds) > POSITION_JUMP_THRESHOLD
552        ):
553            self.update_state()
554
555        # handle sync
556        if self.synced_to:
557            self._handle_sync()
558
559    async def _handle_buffer_ready(self) -> None:
560        """
561        Handle buffer ready event, player has buffered a (new) track.
562
563        Only used when autoplay=0 for coordinated start of synced players.
564        """
565        if self.synced_to:
566            # unpause of sync child is handled by sync master
567            return
568        if not self.group_members:
569            # not a sync group, continue
570            await self.client.unpause_at(self.client.jiffies)
571            return
572        count = 0
573        while count < 40:
574            childs_total = 0
575            childs_ready = 0
576            await asyncio.sleep(0.2)
577            for sync_child in self._get_sync_clients():
578                childs_total += 1
579                if sync_child.state == SlimPlayerState.BUFFER_READY:
580                    childs_ready += 1
581            if childs_total == childs_ready:
582                break
583            count += 1
584
585        # all child's ready (or timeout) - start play
586        async with TaskManager(self.mass) as tg:
587            for sync_client in self._get_sync_clients():
588                # NOTE: Officially you should do an unpause_at based on the player timestamp
589                # but I did not have any good results with that.
590                # Instead just start playback on all players and let the sync logic work out
591                # the delays etc.
592                tg.create_task(pause_and_unpause(sync_client, 200))
593
594    async def _handle_player_cli_event(self, event: SlimEvent) -> None:
595        """Process CLI Event."""
596        if not event.data:
597            return
598        # event data is str, not dict
599        # TODO: fix this in the aioslimproto lib
600        event_data = cast("str", event.data)
601        queue = self.mass.player_queues.get_active_queue(self.player_id)
602        if not queue:
603            return
604        if event_data.startswith("button preset_") and event_data.endswith(".single"):
605            preset_id = event_data.split("preset_")[1].split(".")[0]
606            preset_index = int(preset_id) - 1
607            if len(self.client.presets) >= preset_index + 1:
608                preset = self.client.presets[preset_index]
609                await self.mass.player_queues.play_media(queue.queue_id, preset.uri)
610        elif event_data == "button repeat":
611            if queue.repeat_mode == RepeatMode.OFF:
612                repeat_mode = RepeatMode.ONE
613            elif queue.repeat_mode == RepeatMode.ONE:
614                repeat_mode = RepeatMode.ALL
615            else:
616                repeat_mode = RepeatMode.OFF
617            self.mass.player_queues.set_repeat(queue.queue_id, repeat_mode)
618            self.client.extra_data["playlist repeat"] = REPEATMODE_MAP[queue.repeat_mode]
619            self.client.signal_update()
620        elif event.data == "button shuffle":
621            await self.mass.player_queues.set_shuffle(queue.queue_id, not queue.shuffle_enabled)
622            self.client.extra_data["playlist shuffle"] = int(queue.shuffle_enabled)
623            self.client.signal_update()
624        elif event_data in ("button jump_fwd", "button fwd"):
625            await self.mass.player_queues.next(queue.queue_id)
626        elif event_data in ("button jump_rew", "button rew"):
627            await self.mass.player_queues.previous(queue.queue_id)
628        elif event_data.startswith("time "):
629            # seek request
630            _, param = event_data.split(" ", 1)
631            if param.isnumeric():
632                await self.mass.player_queues.seek(queue.queue_id, int(param))
633        self.logger.log(VERBOSE_LOG_LEVEL, "CLI Event: %s", event_data)
634
635    def _handle_sync(self) -> None:
636        """Synchronize audio of a sync slimplayer."""
637        sync_master_id = self.synced_to
638        if not sync_master_id:
639            # we only correct sync members, not the sync master itself
640            return
641        if not self._provider.slimproto or not (
642            sync_master := self._provider.slimproto.get_player(sync_master_id)
643        ):
644            return  # just here as a guard as bad things can happen
645
646        if sync_master.state != SlimPlayerState.PLAYING:
647            return
648        if self.client.state != SlimPlayerState.PLAYING:
649            return
650
651        # we collect a few playpoints of the player to determine
652        # average lag/drift so we can adjust accordingly
653        sync_playpoints = self._sync_playpoints
654
655        now = time.time()
656        if now < self._do_not_resync_before:
657            return
658
659        last_playpoint = sync_playpoints[-1] if sync_playpoints else None
660        if last_playpoint and (now - last_playpoint.timestamp) > 10:
661            # last playpoint is too old, invalidate
662            sync_playpoints.clear()
663        if last_playpoint and last_playpoint.sync_master != sync_master.player_id:
664            # this should not happen, but just in case
665            sync_playpoints.clear()
666
667        diff = int(
668            self._provider.get_corrected_elapsed_milliseconds(sync_master)
669            - self._provider.get_corrected_elapsed_milliseconds(self.client)
670        )
671
672        sync_playpoints.append(SyncPlayPoint(now, sync_master.player_id, diff))
673
674        # ignore unexpected spikes
675        if (
676            sync_playpoints
677            and abs(statistics.fmean(abs(x.diff) for x in sync_playpoints) - abs(diff))
678            > DEVIATION_JUMP_IGNORE
679        ):
680            return
681
682        min_req_playpoints = 2 if sync_master.elapsed_seconds < 2 else MIN_REQ_PLAYPOINTS
683        if len(sync_playpoints) < min_req_playpoints:
684            return
685
686        # get the average diff
687        avg_diff = statistics.fmean(x.diff for x in sync_playpoints)
688        delta = int(abs(avg_diff))
689
690        if delta < MIN_DEVIATION_ADJUST:
691            return
692
693        # resync the player by skipping ahead or pause for x amount of (milli)seconds
694        sync_playpoints.clear()
695        self._do_not_resync_before = now + 5
696        if avg_diff > MAX_SKIP_AHEAD_MS:
697            # player lagging behind more than MAX_SKIP_AHEAD_MS,
698            # we need to correct the sync_master
699            self.logger.debug("%s resync: pauseFor %sms", sync_master.name, delta)
700            self.mass.create_task(pause_and_unpause(sync_master, delta))
701        elif avg_diff > 0:
702            # handle player lagging behind, fix with skip_ahead
703            self.logger.debug("%s resync: skipAhead %sms", self.display_name, delta)
704            self.mass.create_task(self.client.skip_over(delta))
705        else:
706            # handle player is drifting too far ahead, use pause_for to adjust
707            self.logger.debug("%s resync: pauseFor %sms", self.display_name, delta)
708            self.mass.create_task(pause_and_unpause(self.client, delta))
709
710    async def _set_preset_items(self) -> None:
711        """Set the presets for a player."""
712        preset_items: list[SlimPreset] = []
713        for preset_index in range(1, 11):
714            if preset_conf := self.mass.config.get_raw_player_config_value(
715                self.player_id, f"preset_{preset_index}"
716            ):
717                try:
718                    media_item = await self.mass.music.get_item_by_uri(cast("str", preset_conf))
719                    preset_items.append(
720                        SlimPreset(
721                            uri=media_item.uri,
722                            text=media_item.name,
723                            icon=(
724                                self.mass.metadata.get_image_url(media_item.image)
725                                if media_item.image
726                                else ""
727                            ),
728                        )
729                    )
730                except MusicAssistantError:
731                    # non-existing media item or some other edge case
732                    preset_items.append(
733                        SlimPreset(
734                            uri=f"preset_{preset_index}",
735                            text=f"ERROR <preset {preset_index}>",
736                            icon="",
737                        )
738                    )
739            else:
740                break
741        self.client.presets = preset_items
742
743    async def _set_display(self) -> None:
744        """Set the display config for a player."""
745        display_enabled = self.mass.config.get_raw_player_config_value(
746            self.player_id,
747            CONF_ENTRY_DISPLAY.key,
748            CONF_ENTRY_DISPLAY.default_value,
749        )
750        visualization = self.mass.config.get_raw_player_config_value(
751            self.player_id,
752            CONF_ENTRY_VISUALIZATION.key,
753            CONF_ENTRY_VISUALIZATION.default_value,
754        )
755        await self.client.configure_display(
756            visualisation=SlimVisualisationType(visualization), disabled=not display_enabled
757        )
758
759    def _get_sync_clients(self) -> Iterator[SlimClient]:
760        """Get all sync clients for a player."""
761        yield self.client
762        for member_id in self.group_members:
763            if member_id == self.player_id:  # ← Skip if it's the leader itself
764                continue
765            if self._provider.slimproto and (
766                slimplayer := self._provider.slimproto.get_player(member_id)
767            ):
768                yield slimplayer
769
770    def _get_member_output_codec(self, member_player_id: str, media: PlayerMedia) -> str:
771        """Return the stream format to request for a sync group member."""
772        if media.media_type == MediaType.AUDIO_SOURCE:
773            member_player = self.mass.players.get_player(member_player_id)
774            if member_player and member_player.config.get_value(
775                CONF_PREFER_WAV_FOR_LIVE_SOURCES, default=False
776            ):
777                return "wav"
778        return self.mass.config.get_raw_player_config_value(
779            member_player_id, CONF_OUTPUT_CODEC, "flac"
780        )
781
782
783async def pause_and_unpause(slim_client: SlimClient, pause_duration_ms: int) -> None:
784    """
785    Pause player and schedule unpause after specified duration.
786
787    This is used instead of pause_for because WiiM devices
788    don't properly auto-unpause after pause_for interval.
789    """
790    await slim_client.pause()
791    unpause_timestamp = slim_client.jiffies + pause_duration_ms
792    await slim_client.unpause_at(unpause_timestamp)
793
794
795async def _patched_send_strm(  # noqa: PLR0913
796    self: SlimClient,
797    player: SqueezelitePlayer,
798    command: bytes = b"q",
799    autostart: bytes = b"0",
800    codec_details: bytes = b"p1321",
801    threshold: int = 0,
802    spdif: bytes = b"0",
803    trans_duration: int = 0,
804    trans_type: bytes = b"0",
805    flags: int = 0x20,
806    output_threshold: int = 0,
807    replay_gain: int = 0,
808    server_port: int = 0,
809    server_ip: int = 0,
810    httpreq: bytes = b"",
811) -> None:
812    """Create stream request message based on given arguments."""
813    if player._low_latency_stream:
814        threshold = 64  # KB of input buffer data before autostart or notify
815        output_threshold = (
816            1  # amount of output buffer data before playback starts, in tenths of second
817        )
818    data = struct.pack(
819        "!cc5sBcBcBBBLHL",
820        command,
821        autostart,
822        codec_details,
823        threshold,
824        spdif,
825        trans_duration,
826        trans_type,
827        flags,
828        output_threshold,
829        0,
830        replay_gain,
831        server_port,
832        server_ip,
833    )
834    await self.send_frame(b"strm", data + httpreq)
835