service_hue_collector/hue_collector/collector.py

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any

import requests
import urllib3


@dataclass(frozen=True)
class Settings:
    host: str
    application_key: str
    verify_tls: bool = False
    interval: int = 30
    listen_host: str = "0.0.0.0"
    listen_port: int = 8000
    data_dir: str = "/data"

    @classmethod
    def from_env(cls) -> "Settings":
        host = os.environ.get("HUE_BRIDGE_HOST", "").strip()
        key = os.environ.get("HUE_APPLICATION_KEY", "").strip()
        if not host or not key:
            raise ValueError("HUE_BRIDGE_HOST and HUE_APPLICATION_KEY must be set")
        return cls(
            host=host.removeprefix("https://").removeprefix("http://").rstrip("/"),
            application_key=key,
            verify_tls=os.environ.get("HUE_VERIFY_TLS", "false").lower() in {"1", "true", "yes"},
            interval=int(os.environ.get("COLLECT_INTERVAL_SECONDS", "30")),
            listen_host=os.environ.get("LISTEN_HOST", "0.0.0.0"),
            listen_port=int(os.environ.get("LISTEN_PORT", "8000")),
            data_dir=os.environ.get("DATA_DIR", "/data"),
        )


class HueClient:
    def __init__(self, settings: Settings):
        self.base_url = f"https://{settings.host}/clip/v2/resource"
        self.verify = settings.verify_tls
        self.session = requests.Session()
        self.session.headers["hue-application-key"] = settings.application_key
        if not self.verify:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

    def get(self, resource: str) -> list[dict[str, Any]]:
        response = self.session.get(f"{self.base_url}/{resource}", verify=self.verify, timeout=10)
        response.raise_for_status()
        body = response.json()
        if body.get("errors"):
            raise RuntimeError(f"Hue API error for {resource}: {body['errors']}")
        return body.get("data", [])


class Collector:
    RESOURCE_TYPES = ("device", "room", "zone", "light", "temperature", "light_level", "zigbee_connectivity")

    def __init__(self, client: HueClient):
        self.client = client

    def collect(self) -> dict[str, list[dict[str, Any]]]:
        resources = {kind: self.client.get(kind) for kind in self.RESOURCE_TYPES}
        devices = {item["id"]: item for item in resources["device"]}
        connected = {
            item.get("owner", {}).get("rid"): item.get("status") == "connected"
            for item in resources["zigbee_connectivity"]
        }
        light_by_device = self._by_owner(resources["light"])
        room_for_device, zones_for_device = self._locations(resources["room"], resources["zone"])

        groups: list[dict[str, Any]] = []
        for kind in ("room", "zone"):
            for group in resources[kind]:
                all_lights = [light for child in group.get("children", []) for light in light_by_device.get(child.get("rid"), [])]
                available = [
                    light for child in group.get("children", [])
                    for light in light_by_device.get(child.get("rid"), [])
                    if connected.get(child.get("rid"), True)
                ]
                shining = [light for light in available if light.get("on", {}).get("on", False)]
                brightness = [light.get("dimming", {}).get("brightness") for light in shining]
                brightness = [value for value in brightness if value is not None]
                groups.append({
                    "id": group["id"], "type": kind,
                    "name": group.get("metadata", {}).get("name", group["id"]),
                    "available": len(available), "on": len(shining),
                    "off": len(available) - len(shining),
                    "unavailable": len(all_lights) - len(available),
                    "brightness": sum(brightness) / len(brightness) if brightness else 0.0,
                })

        sensors: list[dict[str, Any]] = []
        for resource_type, field, unit in (("temperature", "temperature", "°C"), ("light_level", "light_level", "lux")):
            for sensor in resources[resource_type]:
                device_id = sensor.get("owner", {}).get("rid", "")
                device = devices.get(device_id, {})
                value = sensor.get(resource_type, {}).get(field)
                valid = not (resource_type == "light_level" and not sensor.get("light_level", {}).get("light_level_valid", True))
                available = connected.get(device_id, True)
                sensors.append({
                    "id": sensor["id"], "type": resource_type,
                    "name": sensor.get("metadata", {}).get("name") or device.get("metadata", {}).get("name", sensor["id"]),
                    "room": room_for_device.get(device_id, ""), "zones": zones_for_device.get(device_id, ""),
                    "available": available, "value": value if available and valid else None, "unit": unit,
                })
        return {"groups": groups, "sensors": sensors}

    @staticmethod
    def _by_owner(items: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
        result: dict[str, list[dict[str, Any]]] = {}
        for item in items:
            result.setdefault(item.get("owner", {}).get("rid", ""), []).append(item)
        return result

    @staticmethod
    def _locations(rooms: list[dict[str, Any]], zones: list[dict[str, Any]]) -> tuple[dict[str, str], dict[str, str]]:
        room_for_device: dict[str, str] = {}
        zones_for_device: dict[str, list[str]] = {}
        for room in rooms:
            for child in room.get("children", []):
                room_for_device[child.get("rid", "")] = room.get("metadata", {}).get("name", room["id"])
        for zone in zones:
            for child in zone.get("children", []):
                zones_for_device.setdefault(child.get("rid", ""), []).append(zone.get("metadata", {}).get("name", zone["id"]))
        return room_for_device, {key: ", ".join(sorted(value)) for key, value in zones_for_device.items()}