/
/
/
1"""State synchronization mapper."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7from music_assistant_models.enums import PlaybackState
8from music_assistant_models.player import PlayerSource
9
10from music_assistant.providers.samsung_wam.features.playback.models import (
11 WamPlaybackState,
12 WamSource,
13)
14
15from .models import WamSpeakerAttributes
16
17if TYPE_CHECKING:
18 from pywam.speaker import Speaker
19
20 from music_assistant.providers.samsung_wam.player import WamPlayer
21
22PLAYBACK_STATE_MAP: dict[WamPlaybackState, PlaybackState] = {
23 WamPlaybackState.PLAY: PlaybackState.PLAYING,
24 WamPlaybackState.PAUSE: PlaybackState.PAUSED,
25 WamPlaybackState.STOP: PlaybackState.IDLE,
26}
27
28PLAYER_SOURCE_MAP = {
29 WamSource.WIFI: PlayerSource(id=WamSource.WIFI, name="Wi-Fi", passive=True),
30 WamSource.BLUETOOTH: PlayerSource(id=WamSource.BLUETOOTH, name="Bluetooth", passive=False),
31 WamSource.AUX: PlayerSource(id=WamSource.AUX, name="AUX", passive=True),
32 WamSource.OPTICAL: PlayerSource(id=WamSource.OPTICAL, name="Optical", passive=True),
33 WamSource.TV_SOUNDCONNECT: PlayerSource(
34 id=WamSource.TV_SOUNDCONNECT, name="TV SoundConnect", passive=False
35 ),
36}
37
38
39class StateSyncMapper:
40 """Maps pywam speaker attributes to player state."""
41
42 @staticmethod
43 def create_speaker_attributes(wam_speaker: Speaker) -> WamSpeakerAttributes:
44 """
45 Create a type-safe attribute snapshot from the raw pywam object.
46
47 :param wam_speaker: The underlying pywam speaker instance.
48 :return: A WamSpeakerAttributes instance.
49 """
50 raw = wam_speaker.attribute
51 return WamSpeakerAttributes(
52 mac=str(raw.mac or ""),
53 name=str(raw.name or "Unknown"),
54 model=raw.model if isinstance(raw.model, str) else None,
55 software_version=raw.software_version
56 if isinstance(raw.software_version, str)
57 else None,
58 source=raw.source if isinstance(raw.source, str) else None,
59 source_list=list(raw.source_list) if isinstance(raw.source_list, list) else [],
60 group_leader_mac=raw.master_mac if isinstance(raw.master_mac, str) else None,
61 is_group_child=bool(raw.is_slave),
62 volume=int(raw.volume) if isinstance(raw.volume, (int, float)) else None,
63 muted=bool(raw.muted) if isinstance(raw.muted, bool) else None,
64 play_status=raw._playstatus if isinstance(raw._playstatus, str) else None,
65 )
66
67 @staticmethod
68 def apply_attributes_to_player(
69 player: WamPlayer,
70 speaker_attrs: WamSpeakerAttributes,
71 group_children: set[str],
72 stream_active: bool,
73 ) -> None:
74 """
75 Apply a full attribute snapshot to the player state.
76
77 :param player: The WamPlayer to update.
78 :param speaker_attrs: The current speaker attributes.
79 :param group_children: Known group children for this player.
80 :param stream_active: Whether the audio stream is currently active.
81 """
82 player._attr_available = True
83 player._attr_name = speaker_attrs.name
84
85 if speaker_attrs.volume is not None:
86 player._attr_volume_level = speaker_attrs.volume
87 if speaker_attrs.muted is not None:
88 player._attr_volume_muted = speaker_attrs.muted
89
90 play_status_str = speaker_attrs.play_status or "stop"
91 try:
92 play_status = WamPlaybackState(play_status_str)
93 player._attr_playback_state = PLAYBACK_STATE_MAP.get(play_status, PlaybackState.IDLE)
94 except ValueError:
95 player._attr_playback_state = PlaybackState.IDLE
96
97 if stream_active:
98 player._attr_active_source = player.player_id
99 else:
100 player._attr_active_source = speaker_attrs.source
101
102 new_sources: list[PlayerSource] = []
103 for source_name in speaker_attrs.source_list:
104 try:
105 wam_source = WamSource(source_name)
106 if source := PLAYER_SOURCE_MAP.get(wam_source):
107 new_sources.append(source)
108 except ValueError:
109 continue
110 player._attr_source_list = new_sources
111
112 if speaker_attrs.is_group_child and speaker_attrs.group_leader_mac:
113 player.synced_to_internal = speaker_attrs.group_leader_mac
114 player._attr_group_members = []
115 else:
116 player.synced_to_internal = None
117 if group_children:
118 player._attr_group_members = [player.player_id, *sorted(group_children)]
119 else:
120 player._attr_group_members = []
121