from music_assistant import MusicAssistant
-class DLNANotifyServer(UpnpNotifyServer):
+class DLNANotifyServer(UpnpNotifyServer): # type: ignore[misc]
"""Notify server for async_upnp_client which uses the MA webserver."""
def __init__(
# transform aiohttp request to async_upnp_client request
http_request = HttpRequest(
method=request.method,
- url=request.url,
+ url=str(request.url),
headers=request.headers,
body=await request.text(),
)
def _handle_event(
self,
service: UpnpService,
- state_variables: Sequence[UpnpStateVariable],
+ state_variables: Sequence[UpnpStateVariable[Any]],
) -> None:
"""Handle state variable(s) changed event from DLNA device."""
if not state_variables:
# Indicates a failure to resubscribe, check if device is still available
self.force_poll = True
return
-
if service.service_id == "urn:upnp-org:serviceId:AVTransport":
for state_variable in state_variables:
# Force a state refresh when player begins or pauses playback
self.display_name,
state_variable.value,
)
-
self.last_seen = time.time()
self.mass.create_task(self._update_player())
Called when provider is deregistered (e.g. MA exiting or config reloading).
"""
self.mass.streams.unregister_dynamic_route("/notify", "NOTIFY")
- if self.dlnaplayers is None:
- return
+
async with TaskManager(self.mass) as tg:
for dlna_player in self.dlnaplayers.values():
tg.create_task(self._device_disconnect(dlna_player))
'^music_assistant/providers/apple_music/.*$',
'^music_assistant/providers/bluesound/.*$',
'^music_assistant/providers/chromecast/.*$',
- '^music_assistant/providers/dlna/.*$',
'^music_assistant/providers/hass_players/.*$',
'^music_assistant/providers/player_group/.*$',
'^music_assistant/providers/qobuz/.*$',