music-assistant-server

1"""DLNA Player Provider."""
2
3import asyncio
4import logging
5from ipaddress import IPv4Address
6
7from async_upnp_client.aiohttp import AiohttpSessionRequester
8from async_upnp_client.client import UpnpRequester
9from async_upnp_client.client_factory import UpnpFactory
10from async_upnp_client.search import async_search
11from async_upnp_client.utils import CaseInsensitiveDict
12from music_assistant_models.player import DeviceInfo
13
14from music_assistant.constants import CONF_PLAYERS, VERBOSE_LOG_LEVEL
15from music_assistant.models.player_provider import PlayerProvider
16
17from .constants import CONF_NETWORK_SCAN
18from .helpers import DLNANotifyServer
19from .player import DLNAPlayer
20
21
22class DLNAPlayerProvider(PlayerProvider):
23    """DLNA Player provider."""
24
25    _discovery_running: bool = False
26    _ignored_udns: set[str]
27
28    lock: asyncio.Lock
29    requester: UpnpRequester
30    upnp_factory: UpnpFactory
31    notify_server: DLNANotifyServer
32
33    async def handle_async_init(self) -> None:
34        """Handle async initialization of the provider."""
35        self.lock = asyncio.Lock()
36        self._ignored_udns = set()
37        # silence the async_upnp_client logger
38        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
39            logging.getLogger("async_upnp_client").setLevel(logging.DEBUG)
40        else:
41            logging.getLogger("async_upnp_client").setLevel(self.logger.level + 10)
42        self.requester = AiohttpSessionRequester(self.mass.http_session, with_sleep=True)
43        self.upnp_factory = UpnpFactory(self.requester, non_strict=True)
44        self.notify_server = DLNANotifyServer(self.requester, self.mass)
45
46    async def unload(self, is_removed: bool = False) -> None:
47        """
48        Handle unload/close of the provider.
49
50        Called when provider is deregistered (e.g. MA exiting or config reloading).
51        """
52        self.mass.streams.unregister_dynamic_route("/notify", "NOTIFY")
53        self._ignored_udns = set()
54
55    async def discover_players(self, use_multicast: bool = False) -> None:
56        """Discover DLNA players on the network."""
57        if self._discovery_running:
58            return
59        try:
60            self._discovery_running = True
61            self.logger.debug("DLNA discovery started...")
62            allow_network_scan = self.config.get_value(CONF_NETWORK_SCAN)
63            discovered_devices: set[str] = set()
64
65            async def on_response(discovery_info: CaseInsensitiveDict) -> None:
66                """Process discovered device from ssdp search."""
67                ssdp_st: str = discovery_info.get("st", discovery_info.get("nt"))
68                if not ssdp_st:
69                    return
70
71                if "MediaRenderer" not in ssdp_st:
72                    # we're only interested in MediaRenderer devices
73                    return
74
75                ssdp_usn: str = discovery_info["usn"]
76                ssdp_udn: str | None = discovery_info.get("_udn")
77                if not ssdp_udn and ssdp_usn.startswith("uuid:"):
78                    ssdp_udn = ssdp_usn.split("::")[0]
79
80                if ssdp_udn in discovered_devices:
81                    # already processed this device
82                    return
83
84                assert ssdp_udn is not None  # for type checking
85
86                discovered_devices.add(ssdp_udn)
87
88                await self._device_discovered(ssdp_udn, discovery_info["location"])
89
90            # we iterate between using a regular and multicast search (if enabled)
91            if allow_network_scan and use_multicast:
92                await async_search(on_response, target=(str(IPv4Address("255.255.255.255")), 1900))
93            else:
94                await async_search(on_response)
95
96        finally:
97            self._discovery_running = False
98
99            def reschedule() -> None:
100                self.mass.create_task(self.discover_players(use_multicast=not use_multicast))
101
102            # reschedule self once finished
103            self.mass.loop.call_later(300, reschedule)
104
105    async def _device_discovered(self, udn: str, description_url: str) -> None:
106        """Handle discovered DLNA player."""
107        async with self.lock:
108            # skip devices that we've already determined should be ignored
109            if udn in self._ignored_udns:
110                return
111
112            if dlna_player := self.mass.players.get_player(udn):
113                # existing player
114                assert isinstance(dlna_player, DLNAPlayer)
115                if dlna_player.description_url == description_url and dlna_player.available:
116                    # nothing to do, device is already connected
117                    return
118                # update description url to newly discovered one
119                dlna_player.description_url = description_url
120            else:
121                # new player detected, setup our DLNAPlayer wrapper
122                conf_key = f"{CONF_PLAYERS}/{udn}/enabled"
123                enabled = self.mass.config.get(conf_key, True)
124                # ignore disabled players
125                if not enabled:
126                    self.logger.debug("Ignoring disabled player: %s", udn)
127                    return
128
129                dlna_player = DLNAPlayer(
130                    provider=self,
131                    player_id=udn,
132                    description_url=description_url,
133                )
134                # will be updated later when device connects
135                dlna_player._attr_device_info = DeviceInfo(
136                    model="unknown",
137                    manufacturer="unknown",
138                )
139
140            # Setup will return False if the device should be ignored (e.g., passive speaker)
141            if not await dlna_player.setup():
142                self._ignored_udns.add(udn)
143