omarchy-screencast/daemon/screencast/backends/dlna.py

"""DLNA / UPnP MediaRenderer.

A renderer is told to fetch a URL (SetAVTransportURI + Play) and polled for what
it is doing. Most TVs handle a live MPEG-TS stream this way; some refuse
anything without a duration, which shows up as a Play error.
"""
from __future__ import annotations

import xml.etree.ElementTree as ET
from html import escape

import aiohttp

from ..devices import Device
from ..util import log
from .base import Backend, BackendError

SOAP_ENV = (
    '<?xml version="1.0" encoding="utf-8"?>'
    '<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"'
    ' s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body>{body}</s:Body></s:Envelope>'
)

PROTOCOL_INFO = {
    "mpegts": "http-get:*:video/mp2t:DLNA.ORG_OP=00;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=8D500000000000000000000000000000",
    "mp4": "http-get:*:video/mp4:DLNA.ORG_OP=00;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=8D500000000000000000000000000000",
    "webm": "http-get:*:video/webm:*",
}


def didl(url: str, title: str, container: str) -> str:
    return (
        '<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/"'
        ' xmlns:dc="http://purl.org/dc/elements/1.1/"'
        ' xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/">'
        '<item id="0" parentID="-1" restricted="1">'
        f"<dc:title>{escape(title)}</dc:title>"
        "<upnp:class>object.item.videoItem.movie</upnp:class>"
        f'<res protocolInfo="{PROTOCOL_INFO.get(container, PROTOCOL_INFO["mpegts"])}">{escape(url)}</res>'
        "</item></DIDL-Lite>"
    )


class DlnaBackend(Backend):
    kind = "dlna"
    # MPEG-TS: a renderer can start on any packet, which is what a live stream
    # needs (see the note in cast.py about fragmented MP4).
    containers = ("mpegts",)

    def __init__(self, daemon) -> None:
        super().__init__(daemon)
        self._session: aiohttp.ClientSession | None = None

    async def _http(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10))
        return self._session

    async def _soap(self, device: Device, service: str, action: str, args: dict) -> dict:
        control = device.extra.get("control") or {}
        url = control.get(service)
        stype = control.get(f"{service}_type")
        if not url or not stype:
            raise BackendError(f"{device.name} has no {service} service")
        # stype comes out of the device's own description XML.
        body = f'<u:{action} xmlns:u="{escape(stype, quote=True)}">' + "".join(
            f"<{k}>{escape(str(v))}</{k}>" for k, v in args.items()
        ) + f"</u:{action}>"
        payload = SOAP_ENV.format(body=body).encode()
        headers = {
            "Content-Type": 'text/xml; charset="utf-8"',
            "SOAPAction": f'"{stype}#{action}"',
            "Connection": "close",
        }
        session = await self._http()
        try:
            async with session.post(url, data=payload, headers=headers) as resp:
                text = (await resp.content.read(256 << 10)).decode("utf-8", "replace")
                if resp.status >= 400:
                    raise BackendError(f"{action} failed: {_fault(text) or resp.status}")
        except aiohttp.ClientError as exc:
            raise BackendError(f"{device.name} did not answer: {exc}") from exc
        return _soap_values(text)

    async def play(self, device: Device, url: str, *, container: str, title: str) -> None:
        await self._soap(device, "avtransport", "SetAVTransportURI",
                         {"InstanceID": 0, "CurrentURI": url,
                          "CurrentURIMetaData": didl(url, title, container)})
        await self._soap(device, "avtransport", "Play", {"InstanceID": 0, "Speed": "1"})
        log.info("dlna: %s playing %s", device.name, url)

    async def stop(self, device: Device) -> None:
        try:
            await self._soap(device, "avtransport", "Stop", {"InstanceID": 0})
        except BackendError as exc:
            log.debug("dlna stop: %s", exc)

    async def transport(self, device: Device, action: str) -> None:
        if action == "pause":
            await self._soap(device, "avtransport", "Pause", {"InstanceID": 0})
        elif action == "play":
            await self._soap(device, "avtransport", "Play", {"InstanceID": 0, "Speed": "1"})
        elif action == "stop":
            await self.stop(device)
        else:
            raise BackendError(f"unknown transport action {action}")

    async def set_volume(self, device: Device, volume: float) -> None:
        await self._soap(device, "rendering", "SetVolume",
                         {"InstanceID": 0, "Channel": "Master",
                          "DesiredVolume": int(max(0.0, min(1.0, volume)) * 100)})

    async def set_muted(self, device: Device, muted: bool) -> None:
        await self._soap(device, "rendering", "SetMute",
                         {"InstanceID": 0, "Channel": "Master", "DesiredMute": 1 if muted else 0})

    async def poll(self, device: Device) -> dict:
        out: dict = {}
        try:
            info = await self._soap(device, "avtransport", "GetTransportInfo", {"InstanceID": 0})
            state = (info.get("CurrentTransportState") or "").lower()
            out["playerState"] = {"playing": "playing", "paused_playback": "paused",
                                  "stopped": "idle", "transitioning": "buffering"}.get(state, state)
            out["connected"] = True
        except BackendError:
            out["connected"] = False
        if (device.extra.get("control") or {}).get("rendering"):
            try:
                vol = await self._soap(device, "rendering", "GetVolume",
                                       {"InstanceID": 0, "Channel": "Master"})
                out["volume"] = int(vol.get("CurrentVolume", 0)) / 100.0
            except BackendError:
                pass
        return out

    async def close(self) -> None:
        if self._session is not None and not self._session.closed:
            await self._session.close()


def _soap_values(text: str) -> dict:
    try:
        root = ET.fromstring(text)
    except ET.ParseError:
        return {}
    values = {}
    for body in root.iter():
        if body.tag.endswith("Body"):
            for resp in body:
                for child in resp:
                    values[child.tag.split("}")[-1]] = child.text or ""
    return values


def _fault(text: str) -> str:
    try:
        root = ET.fromstring(text)
    except ET.ParseError:
        return ""
    for el in root.iter():
        if el.tag.endswith("errorDescription"):
            return el.text or ""
    return ""