music-assistant-server

1"""Discovery core controller."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import inspect
8import logging
9import os
10import re
11from collections import defaultdict
12from ipaddress import IPv4Address
13from typing import TYPE_CHECKING, Any
14
15from aiohttp import ClientTimeout
16from music_assistant_models.config_entries import ConfigEntry
17from music_assistant_models.enums import ConfigEntryType
18from zeroconf import (
19    NonUniqueNameException,
20    ServiceStateChange,
21    Zeroconf,
22)
23from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf
24
25from music_assistant.constants import (
26    CONF_ENTRY_ZEROCONF_INTERFACES,
27    CONF_ZEROCONF_INTERFACES,
28    INGRESS_SERVER_PORT,
29    VERBOSE_LOG_LEVEL,
30)
31from music_assistant.helpers.util import get_ip_pton, get_zeroconf_args
32from music_assistant.models.core_controller import CoreController
33
34if TYPE_CHECKING:
35    from async_upnp_client.utils import CaseInsensitiveDict
36    from music_assistant_models.config_entries import CoreConfig
37
38    from music_assistant.models import ProviderInstanceType
39
40# RAOP cache keys prefix the device name with the device MAC, e.g. "aabbccddeeff@Kelder".
41# Cache keys are lowercased, so the hex is matched in lowercase.
42RAOP_MAC_PREFIX = re.compile(r"^[0-9a-f]{12}@")
43
44CONF_UPNP_NETWORK_SCAN = "upnp_network_scan"
45UPNP_DISCOVERY_INTERVAL = 300
46UPNP_DISCOVERY_BROADCAST_TARGET = (str(IPv4Address("255.255.255.255")), 1900)
47UPNP_DISCOVERY_TASK_ID = "discovery_upnp_cycle"
48UPNP_DISCOVERY_TIMER_ID = "discovery_upnp_timer"
49
50# Re-announce daily so the HA integration token rotates before expiry, also on long uptimes
51HA_ANNOUNCE_INTERVAL = 86400
52HA_ANNOUNCE_TIMER_ID = "discovery_ha_announce_timer"
53
54
55async def async_upnp_search(*args: Any, **kwargs: Any) -> None:
56    """Run async_upnp_client SSDP search with lazy import."""
57    from async_upnp_client.search import async_search  # noqa: PLC0415
58
59    await async_search(*args, **kwargs)
60
61
62class DiscoveryController(CoreController):
63    """Core controller that manages mDNS/Zeroconf and SSDP/UPnP discovery."""
64
65    domain = "discovery"
66
67    def __init__(self, *args: Any, **kwargs: Any) -> None:
68        """Initialize discovery controller."""
69        super().__init__(*args, **kwargs)
70        self.manifest.name = "Discovery"
71        self.manifest.description = (
72            "Handles mDNS/Zeroconf, SSDP/UPnP discovery and Music Assistant network broadcast."
73        )
74        self.manifest.icon = "radar"
75        self._aiozc: AsyncZeroconf | None = None
76        self._mdns_browser: AsyncServiceBrowser | None = None
77        self._mass_service_info: AsyncServiceInfo | None = None
78        self._mdns_locks: dict[str, asyncio.Lock] = {}
79        self._upnp_locks: dict[str, asyncio.Lock] = {}
80        self._upnp_run_lock = asyncio.Lock()
81        self._mdns_waiters: list[asyncio.Event] = []
82
83    @property
84    def aiozc(self) -> AsyncZeroconf:
85        """Return the shared AsyncZeroconf instance for discovery consumers."""
86        assert self._aiozc is not None, "DiscoveryController is not initialized"
87        return self._aiozc
88
89    async def setup(self, config: CoreConfig) -> None:
90        """Initialize discovery controller."""
91        self.config = config
92        if self._aiozc is None:
93            self._aiozc = self._create_aiozc(config)
94        self._configure_library_loggers()
95        await self._setup_mdns_browser()
96        await self._register_mass_service()
97        if self.mass.running_as_hass_addon:
98            # (re)announce to HA supervisor to make sure that HA picks it up
99            await self._announce_to_homeassistant()
100            self._schedule_periodic_ha_announce()
101        self._schedule_periodic_upnp_discovery()
102
103    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
104        """Return config entries for the discovery controller."""
105        return (
106            CONF_ENTRY_ZEROCONF_INTERFACES,
107            ConfigEntry(
108                key=CONF_UPNP_NETWORK_SCAN,
109                type=ConfigEntryType.BOOLEAN,
110                default_value=False,
111                requires_reload=False,
112            ),
113        )
114
115    async def close(self) -> None:
116        """Handle logic on server stop."""
117        self.mass.cancel_timer(UPNP_DISCOVERY_TIMER_ID)
118        self.mass.cancel_task(UPNP_DISCOVERY_TASK_ID)
119        self.mass.cancel_timer(HA_ANNOUNCE_TIMER_ID)
120
121        await self._cancel_mdns_browser()
122
123        aiozc = self._aiozc
124        if self._mass_service_info:
125            with contextlib.suppress(Exception):
126                assert aiozc is not None
127                await aiozc.async_unregister_service(self._mass_service_info)
128            self._mass_service_info = None
129
130        self._mdns_locks.clear()
131        self._upnp_locks.clear()
132        if aiozc is not None:
133            with contextlib.suppress(Exception):
134                await aiozc.async_close()
135            self._aiozc = None
136
137    async def run_provider_discovery(self, provider: ProviderInstanceType) -> None:
138        """Run discovery for a specific provider."""
139        if provider.manifest.mdns_discovery:
140            await self._replay_mdns_discovery(provider)
141        if provider.manifest.upnp_discovery:
142            await self._run_upnp_discovery_cycle(set(provider.manifest.upnp_discovery))
143        self._schedule_periodic_upnp_discovery()
144
145    def on_provider_unload(self, instance_id: str) -> None:
146        """Clean up provider-specific discovery state."""
147        self._mdns_locks.pop(instance_id, None)
148        self._upnp_locks.pop(instance_id, None)
149        self._schedule_periodic_upnp_discovery()
150
151    async def async_find_mdns_service(
152        self, service_type: str, name_filter: str, timeout: float = 3.0
153    ) -> AsyncServiceInfo | None:
154        """
155        Find an mDNS service by exact device name match, checking cache first then waiting.
156
157        :param service_type: The mDNS service type (e.g., "_raop._tcp.local.").
158        :param name_filter: Device name that must exactly match the service name portion.
159        :param timeout: Maximum time to wait in seconds.
160        """
161        deadline = asyncio.get_event_loop().time() + timeout
162        # Cache keys are lowercased DNS names, so we must match case-insensitively
163        name_filter_lower = name_filter.lower()
164        service_type_lower = service_type.lower()
165        event = asyncio.Event()
166        self._mdns_waiters.append(event)
167        try:
168            while True:
169                # Clear before scanning so events arriving during the scan are not lost
170                event.clear()
171                # Check cache for a matching entry
172                for mdns_name in set(self.aiozc.zeroconf.cache.cache):
173                    if service_type_lower not in mdns_name or mdns_name == service_type_lower:
174                        continue
175                    # Use exact matching on the device name portion to prevent a device named
176                    # "Foo" from cross-matching another device named "ATV Foo".
177                    # mDNS names are either "[email protected]." or "DeviceName.service.local."
178                    # Strip the MAC prefix only when present, so device names that legitimately
179                    # contain "@" are not truncated.
180                    device_part = mdns_name.split(".")[0]
181                    device_name = RAOP_MAC_PREFIX.sub("", device_part, count=1)
182                    if device_name != name_filter_lower:
183                        continue
184                    info = AsyncServiceInfo(service_type, mdns_name)
185                    if await info.async_request(self.aiozc.zeroconf, 3000):
186                        return info
187                remaining = deadline - asyncio.get_event_loop().time()
188                if remaining <= 0:
189                    return None
190                # Wait for the next mDNS state change event, then re-check the cache
191                try:
192                    await asyncio.wait_for(event.wait(), timeout=remaining)
193                except TimeoutError:
194                    return None
195        finally:
196            self._mdns_waiters.remove(event)
197
198    def _configure_library_loggers(self) -> None:
199        """Align third-party discovery logging with the discovery controller log level."""
200        library_log_level = (
201            logging.DEBUG if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL) else self.logger.level + 10
202        )
203        for logger_name in ("async_upnp_client", "zeroconf"):
204            logging.getLogger(logger_name).setLevel(library_log_level)
205
206    def _create_aiozc(self, config: CoreConfig) -> AsyncZeroconf:
207        """Create the shared AsyncZeroconf instance for the discovery controller."""
208        zeroconf_interfaces = str(config.get_value(CONF_ZEROCONF_INTERFACES, "default"))
209        use_all_interfaces = zeroconf_interfaces == "all"
210        zc_args = get_zeroconf_args(use_all_interfaces)
211        self.logger.debug("Zeroconf configuration: %s", zc_args)
212        return AsyncZeroconf(
213            ip_version=zc_args["ip_version"],
214            interfaces=zc_args["interfaces"],
215        )
216
217    async def _setup_mdns_browser(self) -> None:
218        """Create the global mDNS browser for all subscribed provider types."""
219        await self._cancel_mdns_browser()
220
221        all_types: set[str] = set()
222        for manifest in self.mass.get_provider_manifests():
223            if manifest.mdns_discovery:
224                all_types.update(manifest.mdns_discovery)
225
226        if not all_types:
227            return
228
229        self._mdns_browser = AsyncServiceBrowser(
230            self.aiozc.zeroconf,
231            list(all_types),
232            handlers=[self._on_mdns_service_state_change],
233        )
234
235    async def _cancel_mdns_browser(self) -> None:
236        """Cancel the active mDNS browser if one is running."""
237        if not self._mdns_browser:
238            return
239
240        async_cancel = getattr(self._mdns_browser, "async_cancel", None)
241        cancel = getattr(self._mdns_browser, "cancel", None)
242        if callable(async_cancel):
243            result = async_cancel()
244            if inspect.isawaitable(result):
245                await result
246            elif callable(cancel):
247                cancel()
248        elif callable(cancel):
249            cancel()
250        self._mdns_browser = None
251
252    async def _register_mass_service(self) -> None:
253        """Register the Music Assistant server on the network via Zeroconf."""
254        zeroconf_type = "_mass._tcp.local."
255        server_id = self.mass.server_id
256        self.logger.debug("Starting Zeroconf broadcast...")
257        info = AsyncServiceInfo(
258            zeroconf_type,
259            name=f"{server_id}.{zeroconf_type}",
260            addresses=[
261                await get_ip_pton(address) for address in self.mass.webserver.publish_addresses
262            ],
263            port=self.mass.webserver.publish_port,
264            properties=self.mass.get_server_info().to_dict(),
265            server="mass.local.",
266        )
267        try:
268            if self._mass_service_info:
269                await self.aiozc.async_update_service(info)
270            else:
271                await self.aiozc.async_register_service(info)
272            self._mass_service_info = info
273        except NonUniqueNameException:
274            self.logger.error(
275                "Music Assistant instance with identical name present in the local network!"
276            )
277
278    def _on_mdns_service_state_change(
279        self,
280        zeroconf: Zeroconf,
281        service_type: str,
282        name: str,
283        state_change: ServiceStateChange,
284    ) -> None:
285        """Handle mDNS service state callbacks."""
286
287        async def process_mdns_state_change(provider: ProviderInstanceType) -> None:
288            lock = self._mdns_locks.setdefault(provider.instance_id, asyncio.Lock())
289            if state_change == ServiceStateChange.Removed:
290                info = None
291            else:
292                info = AsyncServiceInfo(service_type, name)
293                await info.async_request(zeroconf, 3000)
294            async with lock:
295                await provider.on_mdns_service_state_change(name, state_change, info)
296
297        self.logger.log(
298            VERBOSE_LOG_LEVEL,
299            "Service %s of type %s state changed: %s",
300            name,
301            service_type,
302            state_change,
303        )
304        # Notify any waiters that a new mDNS event arrived
305        for waiter in self._mdns_waiters:
306            waiter.set()
307        for provider in list(self.mass.providers):
308            if not provider.available or not provider.manifest.mdns_discovery:
309                continue
310            if service_type in provider.manifest.mdns_discovery:
311                self.mass.create_task(process_mdns_state_change(provider))
312
313    async def _replay_mdns_discovery(self, provider: ProviderInstanceType) -> None:
314        """Replay cached mDNS results for a provider after it loads."""
315        lock = self._mdns_locks.setdefault(provider.instance_id, asyncio.Lock())
316        async with lock:
317            for mdns_type in provider.manifest.mdns_discovery or []:
318                for mdns_name in set(self.aiozc.zeroconf.cache.cache):
319                    if mdns_type not in mdns_name or mdns_type == mdns_name:
320                        continue
321                    info = AsyncServiceInfo(mdns_type, mdns_name)
322                    if await info.async_request(self.aiozc.zeroconf, 3000):
323                        await provider.on_mdns_service_state_change(
324                            mdns_name, ServiceStateChange.Added, info
325                        )
326
327    def _schedule_periodic_upnp_discovery(self) -> None:
328        """Ensure the periodic SSDP discovery cycle matches active subscriptions."""
329        subscriptions, _ = self._get_upnp_subscriptions()
330        if not subscriptions:
331            self.mass.cancel_timer(UPNP_DISCOVERY_TIMER_ID)
332            self.mass.cancel_task(UPNP_DISCOVERY_TASK_ID)
333            return
334
335        def run_discovery() -> None:
336            self.mass.create_task(
337                self._run_upnp_discovery_cycle(),
338                task_id=UPNP_DISCOVERY_TASK_ID,
339            )
340
341        self.mass.call_later(
342            UPNP_DISCOVERY_INTERVAL,
343            run_discovery,
344            task_id=UPNP_DISCOVERY_TIMER_ID,
345        )
346
347    def _get_upnp_subscriptions(
348        self,
349    ) -> tuple[dict[str, list[ProviderInstanceType]], set[str]]:
350        """Return active SSDP subscriptions keyed by search target."""
351        subscriptions: dict[str, list[ProviderInstanceType]] = defaultdict(list)
352        broadcast_targets: set[str] = set()
353        allow_network_scan = bool(self.config.get_value(CONF_UPNP_NETWORK_SCAN))
354        for provider in list(self.mass.providers):
355            if not provider.available or not provider.manifest.upnp_discovery:
356                continue
357            for search_target in provider.manifest.upnp_discovery:
358                subscriptions[search_target].append(provider)
359                if allow_network_scan:
360                    broadcast_targets.add(search_target)
361        return subscriptions, broadcast_targets
362
363    async def _run_upnp_discovery_cycle(self, search_targets: set[str] | None = None) -> None:
364        """Run one SSDP discovery cycle for all active subscriptions."""
365        try:
366            async with self._upnp_run_lock:
367                subscriptions, broadcast_targets = self._get_upnp_subscriptions()
368                if search_targets is not None:
369                    subscriptions = {
370                        search_target: providers
371                        for search_target, providers in subscriptions.items()
372                        if search_target in search_targets
373                    }
374                    broadcast_targets.intersection_update(search_targets)
375                if not subscriptions:
376                    return
377
378                for search_target, providers in subscriptions.items():
379                    seen_results: set[tuple[str | None, str | None, str | None]] = set()
380                    await self._run_upnp_search(search_target, providers, seen_results)
381                    if search_target in broadcast_targets:
382                        await self._run_upnp_search(
383                            search_target,
384                            providers,
385                            seen_results,
386                            target=UPNP_DISCOVERY_BROADCAST_TARGET,
387                        )
388        finally:
389            self._schedule_periodic_upnp_discovery()
390
391    async def _run_upnp_search(
392        self,
393        search_target: str,
394        providers: list[ProviderInstanceType],
395        seen_results: set[tuple[str | None, str | None, str | None]],
396        target: tuple[str, int] | None = None,
397    ) -> None:
398        """Run one SSDP search and dispatch matching responses to subscribed providers."""
399
400        async def on_response(discovery_info: CaseInsensitiveDict) -> None:
401            dedupe_key = (
402                discovery_info.get("usn"),
403                discovery_info.get("location"),
404                discovery_info.get("_host"),
405            )
406            if dedupe_key in seen_results:
407                return
408            seen_results.add(dedupe_key)
409            for provider in providers:
410                if not provider.available:
411                    continue
412                await self._dispatch_upnp_discovery(provider, search_target, discovery_info)
413
414        try:
415            if target is None:
416                await async_upnp_search(on_response, search_target=search_target)
417            else:
418                await async_upnp_search(on_response, search_target=search_target, target=target)
419        except OSError as err:
420            target_label = f" via {target[0]}" if target else ""
421            self.logger.warning(
422                "UPnP discovery for %s failed%s: %s",
423                search_target,
424                target_label,
425                err,
426            )
427
428    async def _dispatch_upnp_discovery(
429        self,
430        provider: ProviderInstanceType,
431        search_target: str,
432        discovery_info: CaseInsensitiveDict,
433    ) -> None:
434        """Dispatch one SSDP discovery result to a provider callback."""
435        lock = self._upnp_locks.setdefault(provider.instance_id, asyncio.Lock())
436        async with lock:
437            try:
438                await provider.on_upnp_service_discovered(search_target, discovery_info)
439            except Exception as err:
440                self.logger.warning(
441                    "Error handling UPnP discovery for %s: %s",
442                    provider.name,
443                    err,
444                    exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
445                )
446
447    async def _announce_to_homeassistant(self) -> None:
448        """Announce Music Assistant Ingress server to Home Assistant via Supervisor API."""
449        supervisor_token = os.environ["SUPERVISOR_TOKEN"]
450        addon_hostname = os.environ["HOSTNAME"]
451        ha_integration_token = await self.mass.webserver.auth.get_homeassistant_system_user_token()
452        discovery_payload = {
453            "service": "music_assistant",
454            "config": {
455                "host": addon_hostname,
456                "port": INGRESS_SERVER_PORT,
457                "auth_token": ha_integration_token,
458            },
459        }
460        try:
461            async with self.mass.http_session_no_ssl.post(
462                "http://supervisor/discovery",
463                headers={"Authorization": f"Bearer {supervisor_token}"},
464                json=discovery_payload,
465                timeout=ClientTimeout(total=10),
466            ) as response:
467                response.raise_for_status()
468                result = await response.json()
469                self.logger.debug(
470                    "Successfully announced to Home Assistant. Discovery UUID: %s",
471                    result.get("uuid"),
472                )
473        except Exception as err:
474            self.logger.warning("Failed to announce to Home Assistant: %s", err)
475
476    def _schedule_periodic_ha_announce(self) -> None:
477        """Schedule the periodic (re)announce to Home Assistant."""
478
479        def run_announce() -> None:
480            self.mass.create_task(self._announce_to_homeassistant())
481            self._schedule_periodic_ha_announce()
482
483        self.mass.call_later(
484            HA_ANNOUNCE_INTERVAL,
485            run_announce,
486            task_id=HA_ANNOUNCE_TIMER_ID,
487        )
488