/
/
/
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