/
/
/
1"""WiiM/LinkPlay Player Provider implementation."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, cast
7
8from async_upnp_client.aiohttp import AiohttpSessionRequester
9from async_upnp_client.client_factory import UpnpFactory
10from async_upnp_client.exceptions import UpnpError
11from music_assistant_models.enums import IdentifierType
12from pywiim import WiiMClient, WiiMError
13from wiim import WiimController
14from wiim.discovery import async_create_wiim_device
15from wiim.exceptions import WiimDeviceException, WiimRequestException
16from zeroconf import ServiceStateChange
17
18from music_assistant.constants import CONF_ENTRY_MANUAL_DISCOVERY_IPS, VERBOSE_LOG_LEVEL
19from music_assistant.helpers.util import (
20 get_port_from_zeroconf,
21 get_primary_ip_address_from_zeroconf,
22)
23from music_assistant.models.player_provider import PlayerProvider
24
25from .constants import PLAYER_ID_PREFIX
26from .grouping import NativeGroupCoordinator
27from .helpers import is_official_manufacturer
28from .linkplay_player import LinkPlayPlayer
29from .player import WiimPlayer
30
31if TYPE_CHECKING:
32 from async_upnp_client.client import UpnpDevice
33 from music_assistant_models.config_entries import ConfigEntry
34 from zeroconf.asyncio import AsyncServiceInfo
35
36 from music_assistant.models.player import Player
37
38# UPnP description.xml ports used by LinkPlay devices, tried in addition to the
39# mDNS-advertised port: 49152 serves the description, 59152 is the advertised UPnP port.
40LINKPLAY_UPNP_PORTS = (49152, 59152)
41
42
43class WiimProvider(PlayerProvider):
44 """
45 WiiM/LinkPlay player provider.
46
47 Official WiiM and Audio Pro speakers are driven by the official WiiM SDK,
48 while other compatible LinkPlay speakers (e.g. Edifier) are driven natively
49 through the public pywiim API within this same provider instance.
50 """
51
52 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
53 """Return Config entries to setup this provider."""
54 return (CONF_ENTRY_MANUAL_DISCOVERY_IPS,)
55
56 async def handle_async_init(self) -> None:
57 """Handle async initialization of the provider."""
58 # the sdk logs routine keep-alive chatter at INFO
59 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
60 logging.getLogger("wiim").setLevel(logging.DEBUG)
61 else:
62 logging.getLogger("wiim").setLevel(max(self.logger.level + 10, logging.WARNING))
63
64 self.wiim_controller = WiimController(self.mass.http_session_no_ssl)
65 # The single native multiroom topology authority shared by both backends.
66 self.native_groups = NativeGroupCoordinator(self)
67 # UPnP identity probe used to classify a discovered device (official WiiM/Audio Pro
68 # vs a generic LinkPlay device such as Edifier). The description.xml is served over
69 # plain HTTP; no UPnP eventing is used by the generic backend.
70 requester = AiohttpSessionRequester(self.mass.http_session_no_ssl, with_sleep=True)
71 self.upnp_factory = UpnpFactory(requester, non_strict=True)
72
73 async def loaded_in_mass(self) -> None:
74 """Call after the provider has been loaded."""
75 manual_ip_config: list[str] = cast(
76 "list[str]", self.config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key)
77 )
78 for ip_address in manual_ip_config:
79 stripped_ip_address = ip_address.strip()
80 await self._discover_device(
81 stripped_ip_address, "Unknown", self._candidate_locations(stripped_ip_address)
82 )
83
84 async def on_mdns_service_state_change(
85 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
86 ) -> None:
87 """Handle MDNS service state callback."""
88 if not info:
89 return
90 if state_change == ServiceStateChange.Removed:
91 return # ignore, rely on availability polling
92
93 cur_address = get_primary_ip_address_from_zeroconf(info)
94 if cur_address is None:
95 return
96
97 locations = self._candidate_locations(cur_address, get_port_from_zeroconf(info))
98
99 # Try to get player_id from mDNS properties first (avoids a network call)
100 udn = info.decoded_properties.get("uuid") if info.decoded_properties else None
101 player_id = f"{PLAYER_ID_PREFIX}{udn}" if udn else None
102 if player_id and (mass_player := self.mass.players.get_player(player_id)):
103 await self._handle_known_player_address(mass_player, cur_address, locations)
104 self.mass.players.trigger_player_update(player_id)
105 return
106
107 mac_address = info.decoded_properties.get("MAC") if info.decoded_properties else None
108 # debounce: mDNS can fire several times in quick succession for one device
109 self.mass.call_later(
110 5,
111 self._discover_device,
112 cur_address,
113 name,
114 locations,
115 mac_address,
116 task_id=f"setup_wiim_{cur_address}",
117 )
118
119 async def try_add_player(
120 self,
121 player_id: str,
122 ip_address: str,
123 name: str,
124 upnp_location: str,
125 mac_address: str | None = None,
126 ) -> None:
127 """Add an official WiiM/Audio Pro device via the official SDK."""
128 try:
129 wiim_dev = await async_create_wiim_device(
130 upnp_location,
131 self.mass.http_session_no_ssl,
132 host=ip_address,
133 local_host=await self.mass.streams.get_source_ip(ip_address),
134 polling_interval=60,
135 )
136 except (WiimRequestException, WiimDeviceException) as err:
137 self.logger.warning("Failed to initialize WiiM device at %s: %s", ip_address, err)
138 return
139 except Exception:
140 self.logger.exception("Unexpected error initializing WiiM device at %s", ip_address)
141 return
142
143 await self.wiim_controller.add_device(wiim_dev)
144 try:
145 player = WiimPlayer(
146 provider=self,
147 player_id=player_id,
148 device=wiim_dev,
149 mac_address=mac_address,
150 )
151 await player.setup()
152 await self.mass.players.register_or_update(player)
153 # read the live topology now the player is registered (setup runs before
154 # registration, so the coordinator would discard a read taken there), then
155 # reconcile so any leader that already lists this device, or that it leads,
156 # self-heals regardless of discovery order.
157 await self.native_groups.refresh_leader(player, force=True)
158 self.native_groups.schedule_reconcile()
159 self.logger.info("WiiM player registered: %s (%s)", wiim_dev.name, player_id)
160 except Exception:
161 self.logger.exception("Failed to register WiiM player %s", wiim_dev.name)
162 await self.wiim_controller.remove_device(wiim_dev.udn)
163 await wiim_dev.disconnect()
164
165 async def try_add_linkplay_player(
166 self,
167 player_id: str,
168 ip_address: str,
169 upnp_device: UpnpDevice,
170 description_url: str,
171 mac_address: str | None = None,
172 ) -> None:
173 """Add a generic LinkPlay device as a grouping/identity shell."""
174 client = WiiMClient(ip_address, session=self.mass.http_session)
175 try:
176 # A successful call confirms the device speaks the LinkPlay API and yields the
177 # device info primed on the shell (used for native group join-mode selection),
178 # so this stays the single authoritative probe done at discovery.
179 device_info = await client.get_device_info_model()
180 except WiiMError as err:
181 self.logger.warning(
182 "Device at %s is not a controllable LinkPlay device: %s", ip_address, err
183 )
184 return
185
186 player = LinkPlayPlayer(
187 provider=self,
188 player_id=player_id,
189 client=client,
190 upnp_device=upnp_device,
191 description_url=description_url,
192 mac_address=mac_address,
193 device_info=device_info,
194 )
195 await player.setup()
196 await self.mass.players.register_or_update(player)
197 # read the live topology now the player is registered (setup runs before
198 # registration, so the coordinator would discard a read taken there), then
199 # reconcile so any leader that already lists this device, or that it leads,
200 # self-heals regardless of discovery order.
201 await self.native_groups.refresh_leader(player, force=True)
202 self.native_groups.schedule_reconcile()
203 self.logger.info("LinkPlay player registered: %s (%s)", player.name, player_id)
204
205 def _candidate_locations(
206 self, ip_address: str, advertised_port: int | None = None
207 ) -> tuple[str, ...]:
208 """Build the ordered, de-duplicated list of description.xml URLs to probe."""
209 ports: list[int] = []
210 if advertised_port:
211 ports.append(advertised_port)
212 ports.extend(LINKPLAY_UPNP_PORTS)
213 locations: list[str] = []
214 for port in ports:
215 location = f"http://{ip_address}:{port}/description.xml"
216 if location not in locations:
217 locations.append(location)
218 root = f"http://{ip_address}/description.xml"
219 if root not in locations:
220 locations.append(root)
221 return tuple(locations)
222
223 async def _probe_locations(
224 self, locations: tuple[str, ...]
225 ) -> tuple[UpnpDevice, str] | tuple[None, None]:
226 """Probe candidate description URLs once and return the first reachable device."""
227 for location in locations:
228 try:
229 upnp_device = await self.upnp_factory.async_create_device(location)
230 except UpnpError:
231 # transient/unreachable or wrong port; try the next candidate
232 continue
233 return upnp_device, location
234 return None, None
235
236 async def _discover_device(
237 self,
238 ip_address: str,
239 name: str,
240 locations: tuple[str, ...],
241 mac_address: str | None = None,
242 ) -> None:
243 """Probe a device's UPnP identity once and route it to the right backend."""
244 upnp_device, matched_location = await self._probe_locations(locations)
245 if upnp_device is None or matched_location is None:
246 # No reachable UPnP description; leave the backend undecided and retry later.
247 return
248
249 player_id = f"{PLAYER_ID_PREFIX}{upnp_device.udn}"
250 if (existing := self.mass.players.get_player(player_id)) is not None:
251 # Already registered; the fast path may have missed, so still reconcile a
252 # moved device here using the description we just probed.
253 await self._reconcile_player_address(
254 existing, ip_address, upnp_device, matched_location
255 )
256 return
257
258 if is_official_manufacturer(upnp_device.manufacturer):
259 await self.try_add_player(player_id, ip_address, name, matched_location, mac_address)
260 else:
261 await self.try_add_linkplay_player(
262 player_id, ip_address, upnp_device, matched_location, mac_address
263 )
264
265 async def _handle_known_player_address(
266 self, mass_player: Player, cur_address: str, locations: tuple[str, ...]
267 ) -> None:
268 """Reconcile an already-registered player with its current mDNS address."""
269 if cur_address == mass_player.device_info.ip_address:
270 return
271 # Probe the new address so the shared reconciler can verify the device identity
272 # before touching the player (works for both backends).
273 upnp_device, matched_location = await self._probe_locations(locations)
274 if upnp_device is not None and matched_location is not None:
275 await self._reconcile_player_address(
276 mass_player, cur_address, upnp_device, matched_location
277 )
278
279 async def _reconcile_player_address(
280 self, mass_player: Player, cur_address: str, upnp_device: UpnpDevice, matched_location: str
281 ) -> None:
282 """Apply an address change to an already-registered player."""
283 if cur_address == mass_player.device_info.ip_address:
284 return
285 # Guard against a stale mDNS/DHCP update where the address now hosts a
286 # different speaker: never bind this player to another device's UPnP identity.
287 if f"{PLAYER_ID_PREFIX}{upnp_device.udn}" != mass_player.player_id:
288 self.logger.warning(
289 "Ignoring address update for %s: %s now hosts a different device (udn=%s)",
290 mass_player.player_id,
291 cur_address,
292 upnp_device.udn,
293 )
294 return
295 if isinstance(mass_player, LinkPlayPlayer):
296 # The generic backend binds its address at construction, so a moved device
297 # needs its HTTP + UPnP resources rebuilt against the new location.
298 await mass_player.async_handle_address_change(
299 cur_address, upnp_device, matched_location
300 )
301 else:
302 # Official players self-heal their connection; only refresh their identifier.
303 mass_player.device_info.add_identifier(IdentifierType.IP_ADDRESS, cur_address)
304