heos/zidoo.py

"""Minimal client for the Zidoo media player's own HTTP API

The AVR only tells HEOS which of its inputs is selected, never what a
device plugged into one is actually showing, so a Zidoo's "now playing"
has to be asked for directly, over its own control API on port 9529.

Its video player only answers getPlayStatus while a video is actually
loaded -- with nothing playing, that route does not exist yet, which
looks the same here as the box being off or unreachable. Both are simply
"nothing to show" rather than an error: this is a nice-to-have on top of
an AVR input, not something the rest of the panel depends on.
"""

import json
import re
import urllib.parse
import urllib.request


# What the card has room for, out of the Zidoo's own wording: the longest
# match wins, so "Dolby Digital Plus" is never read as plain "Dolby Digital".
DYNAMIC_RANGES = {"dolby vision": "Dolby Vision", "dovi": "Dolby Vision", "hdr10+": "HDR10+",
                  "hdr10": "HDR10", "hdr": "HDR", "hlg": "HLG", "sdr": "SDR"}
AUDIO_CODECS = {"dolby digital plus": "DD+", "e-ac3": "DD+", "eac3": "DD+",
                "dolby digital": "DD", "ac3": "DD", "truehd": "TrueHD",
                "dts:x": "DTS:X", "dts-x": "DTS:X", "dts-hd ma": "DTS-HD MA",
                "dts-hd master audio": "DTS-HD MA", "dts-hd": "DTS-HD", "dts": "DTS",
                "aac": "AAC", "flac": "FLAC", "lpcm": "PCM", "pcm": "PCM", "opus": "Opus"}


def _find(text, names):
    """The short name of the longest of `names` found in `text`, or None."""
    lowered = text.lower()
    hits = [key for key in names if re.search(rf"(?<![\w+]){re.escape(key)}(?![\w+])", lowered)]
    return names[max(hits, key=len)] if hits else None


def short_output(output):
    """"3840X2160P @ 23.976Hz YUV422 12Bits SDR" -> "2160p 422 12b SDR":
    the picture's height, its chroma subsampling and bit depth, and its
    dynamic range. Anything not in that shape is passed on as it is."""
    if not output:
        return None
    size = re.search(r"\d+\s*[xX]\s*(\d+)\s*([PpIi])?", output)
    if not size:
        return output
    height = size.group(1) + (size.group(2) or "p").lower()
    # YUV422 or YCbCr444 down to the digits, which are what tell them apart;
    # RGB has no subsampling to give, so it keeps its name.
    chroma = re.search(r"(?:YUV|YCbCr)\s*(4[24][024])|\b(RGB)\b", output, re.I)
    chroma = (chroma.group(1) or chroma.group(2).upper()) if chroma else None
    bits = re.search(r"(\d+)\s*-?\s*bits?\b", output, re.I)
    bits = f"{bits.group(1)}b" if bits else None
    return " ".join(filter(None, (height, chroma, bits, _find(output, DYNAMIC_RANGES))))


def short_audio(audio):
    """"English Dolby Digital Plus with Dolby Atmos 48.0KHz 768kbps (3/3)"
    -> "DD+ Atmos": the codec, and Atmos or DTS:X's objects on top of it. A
    codec it does not know keeps the Zidoo's own words, minus the sample
    rate, bitrate and track count."""
    if not audio or audio.lower().startswith("off"):
        return None
    codec = _find(audio, AUDIO_CODECS)
    if codec is None:
        return re.sub(r"\s*(\d[\d.]*\s*k?hz|\d+\s*kbps|\(\d+/\d+\))", "", audio, flags=re.I).strip() or None
    atmos = "atmos" in audio.lower()
    return f"{codec} Atmos" if atmos else codec


class ZidooError(RuntimeError):
    """The Zidoo did not do what it was asked -- off, unreachable, or with
    nothing loaded to do it to."""


class ZidooClient:
    # The video player answers getPlayStatus and seekTo, but has no route of
    # its own for starting or stopping what it is showing -- the box's remote
    # takes that, and its media keys reach the player like a press on the
    # handset would. One key per state rather than a single toggle, for the
    # same reason a room sends the state it wants (see controller.py).
    KEYS = {"play": "Key.MediaPlay", "pause": "Key.MediaPause"}

    def __init__(self, host, port=9529, timeout=1.5):
        self.base_url = f"http://{host}:{port}"
        self.timeout = timeout
        self._film_for = (None, {})   # (file path, _film()'s answer) last looked up

    def _get(self, route, **params):
        """The raw body of a GET, or None if the Zidoo did not answer."""
        query = f"?{urllib.parse.urlencode(params)}" if params else ""
        try:
            with urllib.request.urlopen(f"{self.base_url}/{route}{query}", timeout=self.timeout) as response:
                return response.read()
        except OSError:    # URLError, and a timeout mid-read, which urlopen does not wrap
            return None

    def _get_json(self, route, **params):
        body = self._get(route, **params)
        try:
            return json.loads(body) if body is not None else None
        except ValueError:
            return None

    def now_playing(self):
        """{"song", "artist", "image"} for whatever video is loaded, in the
        same shape a room's now-playing card already expects -- or None.
        A film has no artist, so that line under the title carries its year
        instead -- or, with no year to give, the video output and audio track. "image" is always None: the poster lives on the Zidoo, which
        the phone cannot load over plain http, so "poster_id" names it
        instead for app.py to serve. "position_ms" and "duration_ms" come
        along whenever the length is known, as a room's do. "play_state" is
        "play" while the film runs (the video's "status" is 1), "pause"
        otherwise."""
        payload = self._get_json("ZidooVideoPlay/getPlayStatus")
        if not payload or payload.get("status") != 200:
            return None
        video = payload.get("video") or {}
        title = video.get("title")
        if not title:
            return None
        film = self._film(video.get("path"))
        year = film.get("year")
        # Nothing to say under the title -- an episode, or a file outside the
        # library -- so say how it is playing: the picture going out to the
        # TV, then the audio track.
        playing_as = " · ".join(filter(None, (short_output(video.get("output")),
                                              short_audio(video.get("audioInfo")))))
        track = {"song": title, "artist": str(year) if year else playing_as or None, "image": None,
                 "poster_id": film.get("id"),
                 "play_state": "play" if video.get("status") == 1 else "pause"}
        if video.get("duration"):
            track["position_ms"] = video.get("currentPosition") or 0
            track["duration_ms"] = video["duration"]
        return track

    def play_state(self, state):
        """Start or stop the loaded video. Like seek(), this is something
        someone asked for, so a Zidoo that does not answer is an error rather
        than nothing to show. The key press is acknowledged at once but the
        player takes a moment to act on it, so the next now_playing() can
        still report the state it is on its way out of."""
        key = self.KEYS.get(state)
        if key is None:
            raise ZidooError(f"The Zidoo has no key for '{state}' -- only play and pause")
        payload = self._get_json("ZidooControlCenter/RemoteControl/sendkey", key=key)
        if not payload or payload.get("status") != 200:
            raise ZidooError(f"The Zidoo did not {state} -- is it on and reachable?")

    def seek(self, position_ms):
        """Jump the loaded video to position_ms. Unlike now_playing(), this
        is something someone asked for, so a Zidoo that does not answer is
        an error rather than simply nothing to show."""
        # "positon" [sic] is the Zidoo's own spelling.
        payload = self._get_json("ZidooVideoPlay/seekTo", positon=int(position_ms))
        if not payload or payload.get("status") != 200:
            raise ZidooError("The Zidoo did not seek -- is a film still loaded on it?")

    def _film(self, path):
        """The poster wall's entry for the film a file belongs to -- its
        "id" (which is also its poster's) and "year" among the rest -- or {}
        when the file is not in the library. Looked up once per file rather
        than on every poll: the answer comes wrapped in the film's whole cast
        and crew, and does not change halfway through it."""
        if not path:
            return {}
        if self._film_for[0] != path:
            payload = self._get_json("ZidooPoster/v2/getAggregationOfFile", path=path)
            if payload is None:
                return {}    # unreachable for now -- ask again next poll
            # "type" names the key holding the item itself: "movie" for a
            # film. Anything else without one simply gets no poster or year.
            self._film_for = (path, payload.get(payload.get("type")) or {})
        return self._film_for[1]

    def poster(self, poster_id, width=200, height=300):
        """(bytes, mimetype) for a film's poster, or None. The Zidoo sends
        no Content-Type, and answers an id it has no poster for with a JSON
        error instead, so the image is told apart by its first bytes."""
        body = self._get("ZidooPoster/getFile/getPoster", id=poster_id, w=width, h=height)
        if body is None:
            return None
        if body.startswith(b"\x89PNG"):
            return body, "image/png"
        if body.startswith(b"\xff\xd8"):
            return body, "image/jpeg"
        return None