service_zigbee/lib_denon.py

"""Small HTTP client for controlling a Denon receiver."""

from __future__ import annotations

import argparse
import socket
import ssl
import sys
from collections.abc import Callable
from urllib.error import URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

import lib_heos


_host = "192.168.178.177"
_port = 8080
_timeout = 3.0
_verify_tls = True


def configure(
    host: str,
    port: int = 8080,
    timeout: float = 3.0,
    verify_tls: bool = True,
) -> None:
    """Configure the receiver used by the parameterless action functions."""
    global _host, _port, _timeout, _verify_tls
    _host = host
    _port = port
    _timeout = timeout
    _verify_tls = verify_tls


def _command(command: str) -> None:
    """Send one command through Denon's HTTP control endpoint."""
    encoded_command = quote(command, safe="")
    url = f"http://{_host}:{_port}/goform/formiPhoneAppDirect.xml?{encoded_command}"
    request = Request(url, headers={"User-Agent": "service-zigbee/1.0"})
    context = None if _verify_tls else ssl._create_unverified_context()
    with urlopen(request, timeout=_timeout, context=context) as response:
        # Reading the response completes the request and lets urllib reuse/close
        # the connection cleanly. HTTP error responses raise an exception.
        response.read()


def on() -> None:
    """Turn the receiver on."""
    _command("PWON")
    print("On")


def off() -> None:
    """Put the receiver into standby."""
    _command("PWSTANDBY")


def _power_state() -> str:
    """Return ON or STANDBY using Denon's control-protocol status query."""
    try:
        with socket.create_connection((_host, 23), timeout=_timeout) as connection:
            connection.settimeout(_timeout)
            connection.sendall(b"PW?\r")
            with connection.makefile("r", encoding="ascii", newline="\r") as stream:
                while line := stream.readline():
                    response = line.strip()
                    if response.startswith("PW"):
                        state = response[2:]
                        if state in {"ON", "STANDBY"}:
                            return state
    except OSError as error:
        raise RuntimeError(f"Cannot query Denon power state: {error}") from error
    raise RuntimeError("Denon did not return a valid power state")


def next() -> None:
    """Play the next HEOS favorite according to the receiver's power state."""
    lib_heos.configure(_host, timeout=_timeout)
    if _power_state() == "STANDBY":
        lib_heos.play_favorite(1)
        return
    # Deliberately do nothing when the receiver is on but its current media is
    # not one of the configured HEOS favorites.
    lib_heos.next_favorite()


def previous() -> None:
    """Play the previous HEOS favorite according to the receiver's power state."""
    lib_heos.configure(_host, timeout=_timeout)
    if _power_state() == "STANDBY":
        lib_heos.play_favorite(1)
        return
    # Deliberately do nothing when the receiver is on but its current media is
    # not one of the configured HEOS favorites.
    lib_heos.previous_favorite()


def volume_up() -> None:
    """Increase the volume by 5 points."""
    for _ in range(5):
        _command("MVUP")


def volume_down() -> None:
    """Decrease the volume by 5 points."""
    for _ in range(5):
        _command("MVDOWN")


def volume_mute() -> None:
    """Mute the volume."""
    _command("MUON")


def main() -> int:
    """Run a receiver action from the command line."""
    parser = argparse.ArgumentParser(description="Control the Denon receiver")
    parser.add_argument(
        "action",
        choices=(
            "on",
            "off",
            "next",
            "previous",
            "volume-up",
            "volume-down",
            "volume-mute",
        ),
    )
    parser.add_argument("--host", default=_host, help=f"receiver address (default: {_host})")
    parser.add_argument("--port", type=int, default=_port, help=f"HTTP port (default: {_port})")
    parser.add_argument(
        "--timeout",
        type=float,
        default=_timeout,
        help=f"request timeout in seconds (default: {_timeout:g})",
    )
    tls_group = parser.add_mutually_exclusive_group()
    tls_group.add_argument(
        "--verify-tls",
        action="store_true",
        default=_verify_tls,
        help="verify the receiver's HTTPS certificate",
    )
    tls_group.add_argument(
        "--no-verify-tls",
        action="store_false",
        dest="verify_tls",
        help="accept a self-signed HTTPS certificate",
    )
    args = parser.parse_args()
    if not 1 <= args.port <= 65535:
        parser.error("--port must be between 1 and 65535")
    if args.timeout <= 0:
        parser.error("--timeout must be positive")

    configure(args.host, args.port, args.timeout, args.verify_tls)
    actions: dict[str, Callable[[], None]] = {
        "on": on,
        "off": off,
        "next": next,
        "previous": previous,
        "volume-up": volume_up,
        "volume-down": volume_down,
        "volume-mute": volume_mute,
    }
    try:
        actions[args.action]()
    except (OSError, URLError, RuntimeError, lib_heos.HeosError) as error:
        print(f"Denon {args.action} failed: {error}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())