/
/
/
1"""Various helpers and utils for the DLNA Player Provider."""
2
3from __future__ import annotations
4
5import xml.etree.ElementTree as ET
6from typing import TYPE_CHECKING
7
8from aiohttp.web import Request, Response
9from async_upnp_client.const import HttpRequest
10from async_upnp_client.event_handler import UpnpEventHandler, UpnpNotifyServer
11
12if TYPE_CHECKING:
13 from async_upnp_client.client import UpnpRequester
14
15 from music_assistant import MusicAssistant
16
17
18class DLNANotifyServer(UpnpNotifyServer): # type: ignore[misc,unused-ignore]
19 """Notify server for async_upnp_client which uses the MA webserver."""
20
21 def __init__(
22 self,
23 requester: UpnpRequester,
24 mass: MusicAssistant,
25 ) -> None:
26 """Initialize."""
27 self.mass = mass
28 self.event_handler = UpnpEventHandler(self, requester)
29 self.mass.streams.register_dynamic_route("/notify", self._handle_request, method="NOTIFY")
30
31 async def _handle_request(self, request: Request) -> Response:
32 """Handle incoming requests."""
33 if request.method != "NOTIFY":
34 return Response(status=405)
35
36 # transform aiohttp request to async_upnp_client request
37 http_request = HttpRequest(
38 method=request.method,
39 url=str(request.url),
40 headers=request.headers,
41 body=await request.text(),
42 )
43
44 try:
45 status = await self.event_handler.handle_notify(http_request)
46 except ET.ParseError as err:
47 self.mass.logger.debug(
48 "Ignoring malformed XML in DLNA notify from %s: %s",
49 request.remote,
50 err,
51 )
52 return Response(status=400)
53
54 return Response(status=status)
55
56 @property
57 def callback_url(self) -> str:
58 """Return callback URL on which we are callable."""
59 return f"{self.mass.streams.base_url}/notify"
60