/
/
/
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 await self.mass.player_queues.set_repeat(queue.queue_id, repeat_mode)
618 # publish the requested mode: on a delegated queue the queue snapshot only
619 # updates once the session's options echo lands
620 self.client.extra_data["playlist repeat"] = REPEATMODE_MAP[repeat_mode]
621 self.client.signal_update()
622 elif event.data == "button shuffle":
623 shuffle_enabled = not queue.shuffle_enabled
624 await self.mass.player_queues.set_shuffle(queue.queue_id, shuffle_enabled)
625 self.client.extra_data["playlist shuffle"] = int(shuffle_enabled)
626 self.client.signal_update()
627 elif event_data in ("button jump_fwd", "button fwd"):
628 await self.mass.player_queues.next(queue.queue_id)
629 elif event_data in ("button jump_rew", "button rew"):
630 await self.mass.player_queues.previous(queue.queue_id)
631 elif event_data.startswith("time "):
632 # seek request
633 _, param = event_data.split(" ", 1)
634 if param.isnumeric():
635 await self.mass.player_queues.seek(queue.queue_id, int(param))
636 self.logger.log(VERBOSE_LOG_LEVEL, "CLI Event: %s", event_data)
637
638 def _handle_sync(self) -> None:
639 """Synchronize audio of a sync slimplayer."""
640 sync_master_id = self.synced_to
641 if not sync_master_id:
642 # we only correct sync members, not the sync master itself
643 return
644 if not self._provider.slimproto or not (
645 sync_master := self._provider.slimproto.get_player(sync_master_id)
646 ):
647 return # just here as a guard as bad things can happen
648
649 if sync_master.state != SlimPlayerState.PLAYING:
650 return
651 if self.client.state != SlimPlayerState.PLAYING:
652 return
653
654 # we collect a few playpoints of the player to determine
655 # average lag/drift so we can adjust accordingly
656 sync_playpoints = self._sync_playpoints
657
658 now = time.time()
659 if now < self._do_not_resync_before:
660 return
661
662 last_playpoint = sync_playpoints[-1] if sync_playpoints else None
663 if last_playpoint and (now - last_playpoint.timestamp) > 10:
664 # last playpoint is too old, invalidate
665 sync_playpoints.clear()
666 if last_playpoint and last_playpoint.sync_master != sync_master.player_id:
667 # this should not happen, but just in case
668 sync_playpoints.clear()
669
670 diff = int(
671 self._provider.get_corrected_elapsed_milliseconds(sync_master)
672 - self._provider.get_corrected_elapsed_milliseconds(self.client)
673 )
674
675 sync_playpoints.append(SyncPlayPoint(now, sync_master.player_id, diff))
676
677 # ignore unexpected spikes
678 if (
679 sync_playpoints
680 and abs(statistics.fmean(abs(x.diff) for x in sync_playpoints) - abs(diff))
681 > DEVIATION_JUMP_IGNORE
682 ):
683 return
684
685 min_req_playpoints = 2 if sync_master.elapsed_seconds < 2 else MIN_REQ_PLAYPOINTS
686 if len(sync_playpoints) < min_req_playpoints:
687 return
688
689 # get the average diff
690 avg_diff = statistics.fmean(x.diff for x in sync_playpoints)
691 delta = int(abs(avg_diff))
692
693 if delta < MIN_DEVIATION_ADJUST:
694 return
695
696 # resync the player by skipping ahead or pause for x amount of (milli)seconds
697 sync_playpoints.clear()
698 self._do_not_resync_before = now + 5
699 if avg_diff > MAX_SKIP_AHEAD_MS:
700 # player lagging behind more than MAX_SKIP_AHEAD_MS,
701 # we need to correct the sync_master
702 self.logger.debug("%s resync: pauseFor %sms", sync_master.name, delta)
703 self.mass.create_task(pause_and_unpause(sync_master, delta))
704 elif avg_diff > 0:
705 # handle player lagging behind, fix with skip_ahead
706 self.logger.debug("%s resync: skipAhead %sms", self.display_name, delta)
707 self.mass.create_task(self.client.skip_over(delta))
708 else:
709 # handle player is drifting too far ahead, use pause_for to adjust
710 self.logger.debug("%s resync: pauseFor %sms", self.display_name, delta)
711 self.mass.create_task(pause_and_unpause(self.client, delta))
712
713 async def _set_preset_items(self) -> None:
714 """Set the presets for a player."""
715 preset_items: list[SlimPreset] = []
716 for preset_index in range(1, 11):
717 if preset_conf := self.mass.config.get_raw_player_config_value(
718 self.player_id, f"preset_{preset_index}"
719 ):
720 try:
721 media_item = await self.mass.music.get_item_by_uri(cast("str", preset_conf))
722 preset_items.append(
723 SlimPreset(
724 uri=media_item.uri,
725 text=media_item.name,
726 icon=(
727 self.mass.metadata.get_image_url(media_item.image)
728 if media_item.image
729 else ""
730 ),
731 )
732 )
733 except MusicAssistantError:
734 # non-existing media item or some other edge case
735 preset_items.append(
736 SlimPreset(
737 uri=f"preset_{preset_index}",
738 text=f"ERROR <preset {preset_index}>",
739 icon="",
740 )
741 )
742 else:
743 break
744 self.client.presets = preset_items
745
746 async def _set_display(self) -> None:
747 """Set the display config for a player."""
748 display_enabled = self.mass.config.get_raw_player_config_value(
749 self.player_id,
750 CONF_ENTRY_DISPLAY.key,
751 CONF_ENTRY_DISPLAY.default_value,
752 )
753 visualization = self.mass.config.get_raw_player_config_value(
754 self.player_id,
755 CONF_ENTRY_VISUALIZATION.key,
756 CONF_ENTRY_VISUALIZATION.default_value,
757 )
758 await self.client.configure_display(
759 visualisation=SlimVisualisationType(visualization), disabled=not display_enabled
760 )
761
762 def _get_sync_clients(self) -> Iterator[SlimClient]:
763 """Get all sync clients for a player."""
764 yield self.client
765 for member_id in self.group_members:
766 if member_id == self.player_id: # â Skip if it's the leader itself
767 continue
768 if self._provider.slimproto and (
769 slimplayer := self._provider.slimproto.get_player(member_id)
770 ):
771 yield slimplayer
772
773 def _get_member_output_codec(self, member_player_id: str, media: PlayerMedia) -> str:
774 """Return the stream format to request for a sync group member."""
775 if media.media_type == MediaType.AUDIO_SOURCE:
776 member_player = self.mass.players.get_player(member_player_id)
777 if member_player and member_player.config.get_value(
778 CONF_PREFER_WAV_FOR_LIVE_SOURCES, default=False
779 ):
780 return "wav"
781 return self.mass.config.get_raw_player_config_value(
782 member_player_id, CONF_OUTPUT_CODEC, "flac"
783 )
784
785
786async def pause_and_unpause(slim_client: SlimClient, pause_duration_ms: int) -> None:
787 """
788 Pause player and schedule unpause after specified duration.
789
790 This is used instead of pause_for because WiiM devices
791 don't properly auto-unpause after pause_for interval.
792 """
793 await slim_client.pause()
794 unpause_timestamp = slim_client.jiffies + pause_duration_ms
795 await slim_client.unpause_at(unpause_timestamp)
796
797
798async def _patched_send_strm( # noqa: PLR0913
799 self: SlimClient,
800 player: SqueezelitePlayer,
801 command: bytes = b"q",
802 autostart: bytes = b"0",
803 codec_details: bytes = b"p1321",
804 threshold: int = 0,
805 spdif: bytes = b"0",
806 trans_duration: int = 0,
807 trans_type: bytes = b"0",
808 flags: int = 0x20,
809 output_threshold: int = 0,
810 replay_gain: int = 0,
811 server_port: int = 0,
812 server_ip: int = 0,
813 httpreq: bytes = b"",
814) -> None:
815 """Create stream request message based on given arguments."""
816 if player._low_latency_stream:
817 threshold = 64 # KB of input buffer data before autostart or notify
818 output_threshold = (
819 1 # amount of output buffer data before playback starts, in tenths of second
820 )
821 data = struct.pack(
822 "!cc5sBcBcBBBLHL",
823 command,
824 autostart,
825 codec_details,
826 threshold,
827 spdif,
828 trans_duration,
829 trans_type,
830 flags,
831 output_threshold,
832 0,
833 replay_gain,
834 server_port,
835 server_ip,
836 )
837 await self.send_frame(b"strm", data + httpreq)
838