music-assistant-server

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