workable POC
This commit is contained in:
@@ -0,0 +1,282 @@
|
||||
#!/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 secrets
|
||||
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,
|
||||
"api_key": existing.get("api_key", secrets.token_urlsafe(32)),
|
||||
})
|
||||
print(f"Saved bridge address, application key, and dashboard API 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())
|
||||
Reference in New Issue
Block a user