POC event handler

lib_denon.py

@@ -0,0 +1,129 @@
+"""Small HTTP client for controlling a Denon receiver."""
+
+from __future__ import annotations
+
+import argparse
+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
+
+
+_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 next() -> None:
+    """Select the next preset or source."""
+
+
+def previous() -> None:
+    """Select the previous preset or source."""
+
+
+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 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"),
+    )
+    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,
+    }
+    try:
+        actions[args.action]()
+    except (OSError, URLError) as error:
+        print(f"Denon {args.action} failed: {error}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())