/
/
/
1"""DLNA Player Provider."""
2
3import asyncio
4import time
5from typing import TYPE_CHECKING
6
7from async_upnp_client.aiohttp import AiohttpSessionRequester
8from async_upnp_client.client_factory import UpnpFactory
9from music_assistant_models.player import DeviceInfo
10
11from music_assistant.constants import CONF_PLAYERS
12from music_assistant.helpers.json import SerializableType
13from music_assistant.models.player_provider import PlayerProvider
14
15from .helpers import DLNANotifyServer
16from .player import DLNAPlayer
17
18if TYPE_CHECKING:
19 from async_upnp_client.client import UpnpRequester
20 from async_upnp_client.utils import CaseInsensitiveDict
21 from music_assistant_models.config_entries import ConfigEntry
22
23
24class DLNAPlayerProvider(PlayerProvider):
25 """DLNA Player provider."""
26
27 _ignored_udns: set[str]
28
29 lock: asyncio.Lock
30 requester: UpnpRequester
31 upnp_factory: UpnpFactory
32 notify_server: DLNANotifyServer
33
34 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
35 """Return Config entries to configure this provider."""
36 return ()
37
38 async def handle_async_init(self) -> None:
39 """Handle async initialization of the provider."""
40 self.lock = asyncio.Lock()
41 self._ignored_udns = set()
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 get_diagnostics(self) -> dict[str, SerializableType]:
56 """Return diagnostics info for this provider to include in diagnostics reports."""
57 now = time.time()
58 devices: list[SerializableType] = [
59 {
60 "model": player.device_info.model,
61 "manufacturer": player.device_info.manufacturer,
62 "available": player.available,
63 "eventing": not player.force_poll,
64 "last_seen_age_sec": round(now - player.last_seen),
65 }
66 for player in self.players
67 if isinstance(player, DLNAPlayer)
68 ]
69 return {
70 "ignored_devices": len(self._ignored_udns),
71 "devices": devices,
72 }
73
74 async def on_upnp_service_discovered(
75 self, search_target: str, discovery_info: CaseInsensitiveDict
76 ) -> None:
77 """Handle SSDP discovery callbacks."""
78 del search_target
79 ssdp_st: str | None = discovery_info.get("st", discovery_info.get("nt"))
80 if not ssdp_st or "MediaRenderer" not in ssdp_st:
81 return
82
83 ssdp_usn: str | None = discovery_info.get("usn")
84 if not ssdp_usn:
85 return
86
87 ssdp_udn: str | None = discovery_info.get("_udn")
88 if not ssdp_udn and ssdp_usn.startswith("uuid:"):
89 ssdp_udn = ssdp_usn.split("::")[0]
90 if not ssdp_udn:
91 return
92
93 description_url: str | None = discovery_info.get("location")
94 if not description_url:
95 return
96
97 await self._device_discovered(ssdp_udn, description_url)
98
99 async def _device_discovered(self, udn: str, description_url: str) -> None:
100 """Handle discovered DLNA player."""
101 async with self.lock:
102 # skip devices that we've already determined should be ignored
103 if udn in self._ignored_udns:
104 return
105
106 if dlna_player := self.mass.players.get_player(udn):
107 # existing player
108 assert isinstance(dlna_player, DLNAPlayer)
109 if dlna_player.description_url == description_url and dlna_player.available:
110 # nothing to do, device is already connected
111 return
112 # update description url to newly discovered one
113 dlna_player.description_url = description_url
114 else:
115 # new player detected, setup our DLNAPlayer wrapper
116 conf_key = f"{CONF_PLAYERS}/{udn}/enabled"
117 enabled = self.mass.config.get(conf_key, True)
118 # ignore disabled players
119 if not enabled:
120 self.logger.debug("Ignoring disabled player: %s", udn)
121 return
122
123 dlna_player = DLNAPlayer(
124 provider=self,
125 player_id=udn,
126 description_url=description_url,
127 )
128 # will be updated later when device connects
129 dlna_player._attr_device_info = DeviceInfo(
130 model="unknown",
131 manufacturer="unknown",
132 )
133
134 # Setup will return False if the device should be ignored (e.g., passive speaker)
135 if not await dlna_player.setup():
136 self._ignored_udns.add(udn)
137