fasthue/hue.py

#!/usr/bin/env python3
"""A small, dependency-free command line client for a local Philips Hue bridge.

The first time you use it, run ``python hue.py register`` while on the same
network as the bridge.  Press the physical button on the bridge when prompted.
"""

from __future__ import annotations

import argparse
import json
import ssl
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


DISCOVERY_URL = "https://discovery.meethue.com/"
DEFAULT_CONFIG = Path(__file__).resolve().parent / "secrets.yaml"


class HueError(RuntimeError):
    pass


def request_json(
    url: str,
    method: str = "GET",
    payload: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    verify_tls: bool = False,
) -> Any:
    """Make a JSON request. Hue bridges commonly use a local self-signed cert."""
    body = json.dumps(payload).encode() if payload is not None else None
    request = Request(url, data=body, method=method)
    request.add_header("Accept", "application/json")
    if payload is not None:
        request.add_header("Content-Type", "application/json")
    for name, value in (headers or {}).items():
        request.add_header(name, value)
    context = ssl.create_default_context() if verify_tls else ssl._create_unverified_context()
    try:
        with urlopen(request, timeout=10, context=context) as response:
            return json.loads(response.read().decode())
    except HTTPError as exc:
        detail = exc.read().decode(errors="replace")
        raise HueError(f"Bridge returned HTTP {exc.code}: {detail}") from exc
    except (URLError, TimeoutError) as exc:
        raise HueError(f"Could not reach {url}: {exc}") from exc
    except json.JSONDecodeError as exc:
        raise HueError(f"{url} did not return JSON") from exc


def discover(verify_tls: bool) -> list[dict[str, Any]]:
    result = request_json(DISCOVERY_URL, verify_tls=verify_tls)
    if not isinstance(result, list):
        raise HueError("Unexpected response from Hue discovery service")
    return result


def load_config(path: Path) -> dict[str, str]:
    if not path.exists():
        return {}
    try:
        contents = path.read_text()
        if path.suffix in {".yaml", ".yml"}:
            data = {}
            for line in contents.splitlines():
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                key, separator, value = line.partition(":")
                if not separator or not key.strip():
                    raise ValueError("expected simple key: value entries")
                data[key.strip()] = value.strip().strip("'\"")
        else:
            data = json.loads(contents)
        return {key: str(value) for key, value in data.items()}
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        raise HueError(f"Could not read {path}: {exc}") from exc


def save_config(path: Path, data: dict[str, str]) -> None:
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    if path.suffix in {".yaml", ".yml"}:
        path.write_text("".join(f"{key}: {value!r}\n" for key, value in data.items()))
    else:
        path.write_text(json.dumps(data, indent=2) + "\n")
    path.chmod(0o600)


@dataclass
class HueBridge:
    host: str
    app_key: str | None
    verify_tls: bool

    @property
    def api_url(self) -> str:
        return f"https://{self.host}/api"

    @property
    def v2_url(self) -> str:
        return f"https://{self.host}/clip/v2/resource"

    def v2(self, path: str, method: str = "GET", payload: dict[str, Any] | None = None) -> Any:
        if not self.app_key:
            raise HueError("No application key. Run `python hue.py register` first.")
        return request_json(
            self.v2_url + path,
            method=method,
            payload=payload,
            headers={"hue-application-key": self.app_key},
            verify_tls=self.verify_tls,
        )

    def resources(self, resource_type: str) -> list[dict[str, Any]]:
        result = self.v2("/" + resource_type)
        data = result.get("data") if isinstance(result, dict) else None
        if not isinstance(data, list):
            raise HueError(f"Unexpected {resource_type} response from bridge")
        return data


def bridge_from_args(args: argparse.Namespace, require_key: bool = True) -> tuple[HueBridge, Path]:
    config_path = Path(args.config).expanduser()
    config = load_config(config_path)
    host = args.bridge or config.get("bridge")
    if not host:
        bridges = discover(args.verify_tls)
        if len(bridges) == 1:
            host = bridges[0].get("internalipaddress")
            if host:
                print(f"Discovered bridge at {host}", file=sys.stderr)
        elif not bridges:
            raise HueError("No Hue bridge found. Supply --bridge IP_OR_HOSTNAME.")
        else:
            choices = ", ".join(str(item.get("internalipaddress")) for item in bridges)
            raise HueError(f"Found multiple bridges ({choices}). Supply --bridge IP_OR_HOSTNAME.")
    key = args.app_key or config.get("app_key")
    if require_key and not key:
        raise HueError("No application key. Run `python hue.py register` first.")
    return HueBridge(str(host), key, args.verify_tls), config_path


def display_resources(bridge: HueBridge) -> None:
    rooms = bridge.resources("room")
    zones = bridge.resources("zone")
    groups = {item["id"]: item.get("metadata", {}).get("name", "unnamed") for item in rooms + zones}
    print("Rooms:")
    for item in rooms:
        print(f"  {item.get('metadata', {}).get('name', 'unnamed')}  [{item['id']}]")
    print("Zones:")
    for item in zones:
        print(f"  {item.get('metadata', {}).get('name', 'unnamed')}  [{item['id']}]")
    print("Scenes:")
    for item in bridge.resources("scene"):
        name = item.get("metadata", {}).get("name", "unnamed")
        group = item.get("group", {})
        group_name = groups.get(group.get("rid"), group.get("rid", "unknown group"))
        print(f"  {name}  ({group_name})  [{item['id']}]")


def select_named(resources: list[dict[str, Any]], name: str, label: str) -> dict[str, Any]:
    matches = [item for item in resources if item.get("metadata", {}).get("name", "").casefold() == name.casefold()]
    if not matches:
        raise HueError(f"No {label} named {name!r}.")
    if len(matches) > 1:
        ids = ", ".join(item["id"] for item in matches)
        raise HueError(f"More than one {label} is named {name!r}: {ids}")
    return matches[0]


def child_grouped_light(group: dict[str, Any]) -> str:
    for resource in group.get("services", []) + group.get("children", []):
        if resource.get("rtype") == "grouped_light" and resource.get("rid"):
            return str(resource["rid"])
    raise HueError(f"{group.get('metadata', {}).get('name', 'Group')!r} has no grouped-light service")


def set_group_power(bridge: HueBridge, target: str, on: bool) -> None:
    rooms, zones = bridge.resources("room"), bridge.resources("zone")
    kind, separator, name = target.partition(":")
    if separator:
        if kind.casefold() == "room":
            group = select_named(rooms, name, "room")
        elif kind.casefold() == "zone":
            group = select_named(zones, name, "zone")
        else:
            raise HueError("Prefix a target with room: or zone:, or use an unambiguous name.")
    else:
        matches = [item for item in rooms + zones if item.get("metadata", {}).get("name", "").casefold() == target.casefold()]
        if not matches:
            raise HueError(f"No room or zone named {target!r}.")
        if len(matches) > 1:
            raise HueError(f"{target!r} is ambiguous; use room:{target} or zone:{target}.")
        group = matches[0]
    bridge.v2(f"/grouped_light/{child_grouped_light(group)}", "PUT", {"on": {"on": on}})
    print(f"Turned {'on' if on else 'off'}: {group.get('metadata', {}).get('name', target)}")


def activate_scene(bridge: HueBridge, name: str) -> None:
    scene = select_named(bridge.resources("scene"), name, "scene")
    bridge.v2(f"/scene/{scene['id']}", "PUT", {"recall": {"action": "active"}})
    print(f"Activated scene: {scene.get('metadata', {}).get('name', name)}")


def register(args: argparse.Namespace) -> None:
    bridge, config_path = bridge_from_args(args, require_key=False)
    print("Press the round button on the Hue Bridge now, then press Enter here.")
    input()
    result = request_json(
        bridge.api_url,
        method="POST",
        payload={"devicetype": args.device_type, "generateclientkey": True},
        verify_tls=args.verify_tls,
    )
    try:
        success = next(item["success"] for item in result if "success" in item)
        app_key = success["username"]
    except (KeyError, StopIteration, TypeError) as exc:
        raise HueError(f"Registration was not accepted: {result}") from exc
    existing = load_config(config_path)
    save_config(config_path, {
        **existing,
        "bridge": bridge.host,
        "app_key": app_key,
    })
    print(f"Saved bridge address and application key in {config_path}")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--bridge", help="Bridge IP address or hostname; auto-discovered if omitted")
    parser.add_argument("--app-key", help="Hue application key; defaults to saved key")
    parser.add_argument("--config", default=str(DEFAULT_CONFIG), help="Credential file (default: %(default)s)")
    parser.add_argument("--verify-tls", action="store_true", help="Verify the bridge TLS certificate")
    commands = parser.add_subparsers(dest="command", required=True)
    registration = commands.add_parser("register", help="Create and save an application key")
    registration.add_argument("--device-type", default="fasthue#python", help="Visible app name on the bridge")
    commands.add_parser("discover", help="Show bridges found via Hue discovery")
    commands.add_parser("list", help="List rooms, zones, and scenes")
    scene = commands.add_parser("scene", help="Activate a scene by name")
    scene.add_argument("name")
    for action in ("on", "off"):
        power = commands.add_parser(action, help=f"Turn a room or zone {action}")
        power.add_argument("target", help="Name, or room:NAME / zone:NAME when ambiguous")
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        if args.command == "discover":
            for item in discover(args.verify_tls):
                print(f"{item.get('internalipaddress', 'unknown')}  id={item.get('id', 'unknown')}")
        elif args.command == "register":
            register(args)
        else:
            bridge, _ = bridge_from_args(args)
            if args.command == "list":
                display_resources(bridge)
            elif args.command == "scene":
                activate_scene(bridge, args.name)
            else:
                set_group_power(bridge, args.target, args.command == "on")
        return 0
    except HueError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 2
    except KeyboardInterrupt:
        print("Cancelled.", file=sys.stderr)
        return 130


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