/
/
/
1"""Discovery handler."""
2
3from __future__ import annotations
4
5import asyncio
6import xml.etree.ElementTree as ET
7from http import HTTPStatus
8from typing import TYPE_CHECKING, cast
9
10import aiohttp
11from aiohttp import ClientTimeout
12from defusedxml import ElementTree as DefusedET
13from pywam.speaker import Speaker
14
15from music_assistant.constants import CONF_ENTRY_MANUAL_DISCOVERY_IPS
16from music_assistant.providers.samsung_wam.features.base import WamProviderFeatureBase
17from music_assistant.providers.samsung_wam.features.playback.models import WamSource
18from music_assistant.providers.samsung_wam.features.state_sync.mapper import StateSyncMapper
19from music_assistant.providers.samsung_wam.player import WamPlayer
20
21from .consts import (
22 PROBE_INTERVAL,
23 PROBE_TASK_ID,
24 PROBE_TIMEOUT,
25 UPNP_DEVICE_DESCRIPTION_PATH,
26 UPNP_PORT,
27)
28
29if TYPE_CHECKING:
30 from music_assistant.providers.samsung_wam.provider import SamsungWamProvider
31
32
33class DiscoveryHandler(WamProviderFeatureBase):
34 """Coordinates finding and initializing speakers on the network."""
35
36 def __init__(self, provider: SamsungWamProvider) -> None:
37 """
38 Initialize the discovery handler.
39
40 :param provider: The SamsungWamProvider instance.
41 """
42 super().__init__(provider)
43 self._discovery_locks: dict[str, asyncio.Lock] = {}
44
45 async def start(self) -> None:
46 """Start speaker discovery."""
47 manual_ips: list[str] = (
48 cast("list[str]", self.provider.config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key))
49 or []
50 )
51 await self._probe_ips(manual_ips)
52 self.mass.create_task(self._periodic_probe_task(), task_id=PROBE_TASK_ID)
53
54 async def stop(self) -> None:
55 """Stop speaker discovery."""
56 self.mass.cancel_task(PROBE_TASK_ID)
57
58 async def on_upnp_discovered(self, udn: str, ip_address: str) -> None:
59 """
60 Handle a UPnP/SSDP presence notification.
61
62 :param udn: The Universal Device Name of the device.
63 :param ip_address: The IP address of the discovered device.
64 """
65 # Skip if the player is already registered and available
66 existing = self._get_player_by_udn(udn)
67 if existing and existing.available:
68 return
69
70 # Probe the device to confirm model compatibility and obtain its canonical UDN,
71 # which is sourced directly from the device description XML rather than SSDP headers
72 if canonical_udn := await self.probe_ip(ip_address):
73 await self._handle_presence(canonical_udn, ip_address)
74
75 async def probe_ip(self, ip_address: str) -> str | None:
76 """
77 Probe a device to verify it is reachable and a supported WAM model.
78
79 :param ip_address: The IP address to probe.
80 :return: The device UDN, or None if the device is not reachable or unsupported.
81 """
82 location = f"http://{ip_address}:{UPNP_PORT}{UPNP_DEVICE_DESCRIPTION_PATH}"
83 try:
84 async with self.mass.http_session.get(
85 location, timeout=ClientTimeout(total=PROBE_TIMEOUT)
86 ) as resp:
87 if resp.status != HTTPStatus.OK:
88 return None
89 xml_text = await resp.text()
90
91 root = DefusedET.fromstring(xml_text)
92
93 model_el = root.find(".//{*}modelName")
94 model_name = model_el.text if model_el is not None else None
95 if model_name not in self.provider.supported_models:
96 return None
97
98 udn_el = root.find(".//{*}UDN")
99 if udn_el is None or not udn_el.text:
100 return None
101
102 return str(udn_el.text.removeprefix("uuid:"))
103 except TimeoutError, aiohttp.ClientError, ET.ParseError:
104 return None
105
106 async def _handle_presence(self, udn: str, ip_address: str) -> None:
107 """
108 Process a validated presence event for a device.
109
110 :param udn: The canonical Universal Device Name of the device.
111 :param ip_address: The IP address of the device.
112 """
113 lock = self._discovery_locks.setdefault(udn, asyncio.Lock())
114 async with lock:
115 existing = self._get_player_by_udn(udn)
116 if existing:
117 if not existing.available:
118 await existing.poll()
119 else:
120 await self._setup_player(udn, ip_address)
121
122 async def _probe_ips(self, ips_to_probe: list[str]) -> None:
123 """
124 Actively probe a list of IP addresses.
125
126 :param ips_to_probe: List of IP addresses to probe.
127 """
128 for ip in (ip for ip in ips_to_probe if ip):
129 self.logger.debug("Probing %s", ip)
130 if udn := await self.probe_ip(ip):
131 await self._handle_presence(udn, ip)
132
133 async def _periodic_probe_task(self) -> None:
134 """Periodically probe manually configured IP addresses."""
135 while not self.mass.closing:
136 try:
137 await asyncio.sleep(PROBE_INTERVAL)
138 manual_ips: list[str] = (
139 cast(
140 "list[str]",
141 self.provider.config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key),
142 )
143 or []
144 )
145 if manual_ips:
146 await self._probe_ips(manual_ips)
147 except asyncio.CancelledError:
148 break
149 except Exception as err:
150 self.logger.warning("Periodic probe failed: %s", err, exc_info=err)
151
152 async def _setup_player(self, udn: str, ip_address: str) -> None:
153 """
154 Initialize and register a newly discovered player.
155
156 :param udn: The Universal Device Name of the device.
157 :param ip_address: The IP address of the device.
158 """
159 self.logger.debug("Connecting to new player at %s", ip_address)
160
161 temp_speaker = Speaker(ip_address)
162 try:
163 await temp_speaker.connect()
164 except Exception as err:
165 self.logger.warning("Failed to connect to player at %s: %s", ip_address, err)
166 return
167
168 try:
169 await temp_speaker.update()
170 attrs = StateSyncMapper.create_speaker_attributes(temp_speaker)
171 if not attrs.mac:
172 raise ConnectionError("Could not retrieve MAC address from speaker")
173 except Exception as err:
174 self.logger.warning("Failed to set up player at %s: %s", ip_address, err)
175 await temp_speaker.disconnect()
176 return
177
178 if not self.mass.config.get_raw_player_config_value(attrs.mac, "enabled", True):
179 self.logger.debug("Player at %s is disabled in configuration", ip_address)
180 await temp_speaker.disconnect()
181 return
182
183 try:
184 player = WamPlayer(self.provider, ip_address, udn, attrs.mac, temp_speaker)
185 player.state_sync.apply_initial_state(attrs)
186
187 await self.mass.players.register_or_update(player)
188 player.state_sync.subscribe_speaker_events()
189 self.provider.groups.register_player(player)
190
191 if attrs.source and attrs.source not in (WamSource.WIFI, "Unknown"):
192 # Set immediately if the speaker is already on an external input,
193 # so the UI reflects the correct source on first register
194 player.set_active_mass_source(attrs.source)
195 except Exception as err:
196 self.logger.warning("Failed to set up player at %s: %s", ip_address, err)
197 await temp_speaker.disconnect()
198
199 def _get_player_by_udn(self, udn: str) -> WamPlayer | None:
200 """
201 Retrieve an existing player by its UDN.
202
203 :param udn: The Universal Device Name to search for.
204 :return: The matching WamPlayer instance or None.
205 """
206 return next((p for p in self.players if p.udn == udn), None)
207