fasthue/app.py

#!/usr/bin/env python3
"""Phone-first Hue room and zone dashboard."""

from __future__ import annotations

import argparse
import os
from pathlib import Path
from typing import Any

import yaml
from flask import Flask, jsonify, render_template, request
from werkzeug.middleware.proxy_fix import ProxyFix

from hue import DEFAULT_CONFIG, HueBridge, HueError, load_config


ROOT = Path(__file__).resolve().parent
APP_CONFIG = ROOT / "config.yaml"


def read_dashboard(path: Path) -> list[dict[str, Any]]:
    try:
        data = yaml.safe_load(path.read_text())
        buttons = data["buttons"]
        rooms = {item["name"]: {**item, "kind": "room"} for item in data["rooms"]}
        zones = {item["name"]: {**item, "kind": "zone"} for item in data["zones"]}
    except (OSError, KeyError, TypeError, yaml.YAMLError) as exc:
        raise RuntimeError(f"Could not read dashboard configuration {path}: {exc}") from exc

    result = []
    for position, button in sorted(buttons.items(), key=lambda item: int(item[0])):
        # A simple name is retained as a backwards-compatible configuration.
        name = button["name"] if isinstance(button, dict) else button
        group = rooms.get(name) or zones.get(name)
        if not group:
            raise RuntimeError(f"Button {position} refers to unknown room or zone {name!r}")
        result.append({**group, "scenes": button.get("scenes", []) if isinstance(button, dict) else []})
    if len(result) != 8:
        raise RuntimeError(f"Expected exactly 8 buttons, found {len(result)}")
    return result


def make_bridge(verify_tls: bool) -> HueBridge:
    config = load_config(Path(os.environ.get("FASTHUE_SECRETS_FILE", DEFAULT_CONFIG)).expanduser())
    host = os.environ.get("HUE_BRIDGE", config.get("bridge"))
    app_key = os.environ.get("HUE_APP_KEY", config.get("app_key"))
    if not host or not app_key:
        raise HueError("No bridge/key available. Run `python hue.py register` first.")
    return HueBridge(host, app_key, verify_tls)


def grouped_light_id(group: dict[str, Any]) -> str:
    # Current bridges put the group's control service in `services`; older
    # resource shapes exposed it among `children`.
    for resource in group.get("services", []) + group.get("children", []):
        if resource.get("rtype") == "grouped_light":
            return resource["rid"]
    raise HueError(f"{group['metadata']['name']!r} has no grouped-light service")


def light_ids(group: dict[str, Any], devices: dict[str, dict[str, Any]]) -> set[str]:
    """Find actual lights, so an indicator means *any* light is on."""
    ids: set[str] = set()
    for child in group.get("children", []):
        if child.get("rtype") == "light":
            ids.add(child["rid"])
        elif child.get("rtype") == "device":
            device = devices.get(child.get("rid"), {})
            ids.update(service["rid"] for service in device.get("services", []) if service.get("rtype") == "light")
    return ids


def scene_is_active(scene: dict[str, Any]) -> bool:
    """Hue returns `inactive`, `static`, or a dynamic scene mode in status.active."""
    active = scene.get("status", {}).get("active")
    return active is True or (active not in (None, False, "inactive"))


def dashboard_state(buttons: list[dict[str, str]], verify_tls: bool) -> list[dict[str, Any]]:
    bridge = make_bridge(verify_tls)
    groups = bridge.resources("room") + bridge.resources("zone")
    group_by_id = {group["id"]: group for group in groups}
    devices = {device["id"]: device for device in bridge.resources("device")}
    lights = {light["id"]: light for light in bridge.resources("light")}
    grouped = {item["id"]: item for item in bridge.resources("grouped_light")}
    scenes = bridge.resources("scene")

    state = []
    for button in buttons:
        group = group_by_id.get(button["id"])
        if not group:
            raise HueError(f"{button['kind'].title()} {button['name']!r} is no longer present on the bridge")
        ids = light_ids(group, devices)
        if ids:
            any_on = any(lights.get(light_id, {}).get("on", {}).get("on", False) for light_id in ids)
        else:
            # Some bridges expose only the grouped-light service for a group.
            any_on = grouped.get(grouped_light_id(group), {}).get("on", {}).get("on", False)
        active_scene = next(
            (scene.get("metadata", {}).get("name") for scene in scenes
             if scene.get("group", {}).get("rid") == button["id"] and scene_is_active(scene)),
            None,
        )
        state.append({"id": button["id"], "name": button["name"], "on": any_on, "scene": active_scene})
    return state


def next_scene(button: dict[str, Any], bridge: HueBridge) -> dict[str, Any]:
    """Recall the scene after the currently active configured scene."""
    catalog = [scene for scene in bridge.resources("scene") if scene.get("group", {}).get("rid") == button["id"]]
    by_name = {scene.get("metadata", {}).get("name"): scene for scene in catalog}
    try:
        ordered = [by_name[name] for name in button["scenes"]]
    except KeyError as exc:
        raise HueError(f"Configured scene {exc.args[0]!r} is no longer on the bridge") from exc
    if not ordered:
        raise HueError(f"No scenes are configured for {button['name']!r}")

    active_index = next((index for index, scene in enumerate(ordered) if scene_is_active(scene)), None)
    selected = ordered[0] if active_index is None else ordered[(active_index + 1) % len(ordered)]
    bridge.v2(f"/scene/{selected['id']}", "PUT", {"recall": {"action": "active"}})
    return {"name": selected.get("metadata", {}).get("name", "scene")}


def create_app(yaml_path: Path, verify_tls: bool, url_prefix: str = "") -> Flask:
    app = Flask(__name__)
    if url_prefix:
        # Gunicorn only receives traffic over its private socket from Nginx.
        # Nginx strips this prefix and supplies it in X-Forwarded-Prefix.
        app.wsgi_app = ProxyFix(app.wsgi_app, x_prefix=1)
    buttons = read_dashboard(yaml_path)

    @app.get("/")
    def index() -> str:
        return render_template("index.html", buttons=buttons)

    @app.get("/api/groups")
    def groups() -> Any:
        try:
            return jsonify({"groups": dashboard_state(buttons, verify_tls)})
        except (HueError, RuntimeError) as exc:
            return jsonify({"error": str(exc)}), 503

    @app.post("/api/groups/<group_id>/next-scene")
    def activate_next_scene(group_id: str) -> Any:
        try:
            button = next(item for item in buttons if item["id"] == group_id)
            bridge = make_bridge(verify_tls)
            scene = next_scene(button, bridge)
            return jsonify({"id": group_id, "scene": scene["name"]})
        except StopIteration:
            return jsonify({"error": "Unknown dashboard group"}), 404
        except (HueError, RuntimeError) as exc:
            return jsonify({"error": str(exc)}), 503

    @app.post("/api/groups/<group_id>/off")
    def turn_group_off(group_id: str) -> Any:
        try:
            button = next(item for item in buttons if item["id"] == group_id)
            bridge = make_bridge(verify_tls)
            group = next(item for item in bridge.resources(button["kind"]) if item["id"] == group_id)
            bridge.v2(f"/grouped_light/{grouped_light_id(group)}", "PUT", {"on": {"on": False}})
            return jsonify({"id": group_id, "on": False})
        except StopIteration:
            return jsonify({"error": "Unknown dashboard group"}), 404
        except (HueError, RuntimeError) as exc:
            return jsonify({"error": str(exc)}), 503

    @app.post("/api/groups/<group_name>/scene")
    def activate_group_scene(group_name: str) -> Any:
        payload = request.get_json(silent=True) or {}
        scene_name = payload.get("name")
        if not isinstance(scene_name, str) or not scene_name.strip():
            return jsonify({"error": "JSON body must include a non-empty scene name"}), 400
        try:
            bridge = make_bridge(verify_tls)
            groups = bridge.resources("room") + bridge.resources("zone")
            matching_groups = [
                group for group in groups
                if group.get("metadata", {}).get("name", "").casefold() == group_name.casefold()
            ]
            if not matching_groups:
                return jsonify({"error": f"No room or zone named {group_name!r}"}), 404
            if len(matching_groups) > 1:
                return jsonify({"error": f"{group_name!r} matches more than one room or zone"}), 409
            group = matching_groups[0]
            scene = next(
                item for item in bridge.resources("scene")
                if item.get("group", {}).get("rid") == group["id"]
                and item.get("metadata", {}).get("name", "").casefold() == scene_name.casefold()
            )
            bridge.v2(f"/scene/{scene['id']}", "PUT", {"recall": {"action": "active"}})
            return jsonify({"group": group["metadata"]["name"], "scene": scene["metadata"]["name"]})
        except StopIteration:
            return jsonify({"error": f"No scene named {scene_name!r} belongs to this room or zone"}), 404
        except (HueError, RuntimeError) as exc:
            return jsonify({"error": str(exc)}), 503

    return app


def configured_url_prefix(path: Path = APP_CONFIG) -> str:
    """Read and validate the reverse-proxy URL prefix from config.yaml."""
    url_prefix = load_config(path).get("url_prefix", "").rstrip("/")
    if url_prefix and not url_prefix.startswith("/"):
        raise RuntimeError("config.yaml url_prefix must start with /, for example /hue")
    return url_prefix


def create_gunicorn_app() -> Flask:
    """Gunicorn application factory; configuration is read from environment/files."""
    config_path = Path(os.environ.get("FASTHUE_DASHBOARD_CONFIG", ROOT / "hue.yaml"))
    verify_tls = os.environ.get("FASTHUE_VERIFY_TLS", "").casefold() in {"1", "true", "yes"}
    return create_app(config_path, verify_tls, configured_url_prefix())


# Older Gunicorn releases do not support --factory.  Expose a WSGI callable at
# import time so they can load this application with `app:gunicorn_app`.
if __name__ != "__main__":
    gunicorn_app = create_gunicorn_app()


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, default=ROOT / "hue.yaml", help="Dashboard YAML file")
    parser.add_argument("--verify-tls", action="store_true", help="Verify the Hue bridge certificate")
    parser.add_argument("--url-prefix", help="Override the reverse-proxy URL prefix from config.yaml")
    parser.add_argument("--host", default="0.0.0.0", help="Listen address")
    parser.add_argument("--port", type=int, default=5000, help="Listen port")
    args = parser.parse_args()
    url_prefix = args.url_prefix.rstrip("/") if args.url_prefix is not None else configured_url_prefix()
    if url_prefix and not url_prefix.startswith("/"):
        parser.error("--url-prefix must start with /, for example /hue")
    create_app(args.config, args.verify_tls, url_prefix).run(host=args.host, port=args.port, debug=False)


if __name__ == "__main__":
    main()