/
/
/
1"""Samsung WAM player provider."""
2
3from __future__ import annotations
4
5from collections.abc import Mapping
6from typing import TYPE_CHECKING, Any, cast
7from urllib.parse import urlparse
8
9from music_assistant_models.enums import ProviderFeature
10from pywam.device import SPEAKER_MODELS
11
12from music_assistant.constants import CONF_ENTRY_MANUAL_DISCOVERY_IPS
13from music_assistant.models.player_provider import PlayerProvider
14
15from .features.discovery.handler import DiscoveryHandler
16from .features.grouping.coordinator import GroupingCoordinator
17
18if TYPE_CHECKING:
19 from music_assistant_models.config_entries import ConfigEntry
20
21 from .player import WamPlayer
22
23
24class SamsungWamProvider(PlayerProvider):
25 """Samsung WAM player provider."""
26
27 supported_models: dict[str, dict[str, Any]]
28 groups: GroupingCoordinator
29 discovery: DiscoveryHandler
30
31 @property
32 def supported_features(self) -> set[ProviderFeature]:
33 """Return the features supported by this Provider."""
34 return {ProviderFeature.SYNC_PLAYERS}
35
36 def get_players(self) -> list[WamPlayer]:
37 """Return all registered WAM players."""
38 return cast("list[WamPlayer]", self.players)
39
40 def get_player(self, player_id: str) -> WamPlayer | None:
41 """
42 Return a WAM player by ID.
43
44 :param player_id: The player ID to look up.
45 :return: The matching WamPlayer, or None if not found.
46 """
47 return cast("WamPlayer | None", self.mass.players.get_player(player_id))
48
49 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
50 """Return all config entries for this provider."""
51 return (CONF_ENTRY_MANUAL_DISCOVERY_IPS,)
52
53 async def handle_async_init(self) -> None:
54 """Initialise the provider."""
55 self.supported_models = SPEAKER_MODELS
56 self.groups = GroupingCoordinator(self)
57 self.discovery = DiscoveryHandler(self)
58
59 async def loaded_in_mass(self) -> None:
60 """Start discovery and initial group sync."""
61 await self.discovery.start()
62 self.groups.start_sync_task()
63
64 async def unload(self, is_removed: bool = False) -> None:
65 """
66 Handle close/cleanup of the provider.
67
68 :param is_removed: True if the provider is being permanently removed.
69 """
70 await self.discovery.stop()
71 self.groups.stop_sync_task()
72 for player in list(self.get_players()):
73 self.logger.debug("Unloading player %s", player.log_name)
74 await self.mass.players.unregister(player.player_id, is_removed)
75 self.groups.states.clear()
76
77 async def on_upnp_service_discovered(
78 self, search_target: str, discovery_info: Mapping[str, Any]
79 ) -> None:
80 """
81 Handle a UPnP/SSDP presence notification.
82
83 :param search_target: The SSDP service type that was matched.
84 :param discovery_info: The raw SSDP response headers.
85 """
86 location = discovery_info.get("location", "")
87 ip_address = urlparse(location).hostname
88 if not ip_address:
89 return
90
91 usn = discovery_info.get("usn", "")
92 udn = usn.split("::")[0].removeprefix("uuid:")
93 if not udn:
94 return
95
96 await self.discovery.on_upnp_discovered(udn, ip_address)
97