from __future__ import annotations import json import logging import os import ssl import threading import time from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any import requests import urllib3 LOG = logging.getLogger("hue-collector") def _label(value: Any) -> str: return str(value).replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"') def _sample(name: str, value: float | int, **labels: str) -> str: rendered = ",".join(f'{key}="{_label(val)}"' for key, val in sorted(labels.items())) return f"{name}{{{rendered}}} {value}" if rendered else f"{name} {value}" @dataclass(frozen=True) class Settings: host: str application_key: str verify_tls: bool = False interval: float = 30 listen_host: str = "0.0.0.0" listen_port: int = 8000 @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=float(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")), ) 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) -> str: 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"]) lines = [ "# HELP hue_collection_success Whether the most recent Hue collection succeeded.", "# TYPE hue_collection_success gauge", "hue_collection_success 1", "# HELP hue_group_lights_available Number of connected lights in a room or zone.", "# TYPE hue_group_lights_available gauge", "# HELP hue_group_lights_shining Number of connected lights that are on.", "# TYPE hue_group_lights_shining gauge", "# HELP hue_group_lights_off Number of connected lights that are off.", "# TYPE hue_group_lights_off gauge", "# HELP hue_group_lights_unavailable Number of configured lights that are disconnected.", "# TYPE hue_group_lights_unavailable gauge", "# HELP hue_group_lights_shining_percent Percentage of connected lights that are on.", "# TYPE hue_group_lights_shining_percent gauge", "# HELP hue_group_lights_off_percent Percentage of connected lights that are off.", "# TYPE hue_group_lights_off_percent gauge", "# HELP hue_group_lights_unavailable_percent Percentage of configured lights that are disconnected.", "# TYPE hue_group_lights_unavailable_percent gauge", "# HELP hue_group_average_brightness_percent Average brightness of connected lights that are on.", "# TYPE hue_group_average_brightness_percent gauge", ] 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"), []) ] lights = [ 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 lights if light.get("on", {}).get("on", False)] off = len(lights) - len(shining) unavailable = len(all_lights) - len(lights) labels = {"group_type": kind, "group": group.get("metadata", {}).get("name", group["id"]), "group_id": group["id"]} lines.append(_sample("hue_group_lights_available", len(lights), **labels)) lines.append(_sample("hue_group_lights_shining", len(shining), **labels)) lines.append(_sample("hue_group_lights_off", off, **labels)) lines.append(_sample("hue_group_lights_unavailable", unavailable, **labels)) lines.append(_sample("hue_group_lights_shining_percent", 100 * len(shining) / len(lights) if lights else 0, **labels)) lines.append(_sample("hue_group_lights_off_percent", 100 * off / len(lights) if lights else 0, **labels)) lines.append(_sample("hue_group_lights_unavailable_percent", 100 * unavailable / len(all_lights) if all_lights else 0, **labels)) brightness = [light.get("dimming", {}).get("brightness") for light in shining] brightness = [value for value in brightness if value is not None] lines.append(_sample("hue_group_average_brightness_percent", sum(brightness) / len(brightness) if brightness else 0, **labels)) lines.extend(self._sensor_metrics(resources, devices, connected, room_for_device, zones_for_device)) return "\n".join(lines) + "\n" @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: name = room.get("metadata", {}).get("name", room["id"]) for child in room.get("children", []): room_for_device[child.get("rid", "")] = name for zone in zones: name = zone.get("metadata", {}).get("name", zone["id"]) for child in zone.get("children", []): zones_for_device.setdefault(child.get("rid", ""), []).append(name) return room_for_device, {key: ", ".join(sorted(value)) for key, value in zones_for_device.items()} def _sensor_metrics(self, resources, devices, connected, room_for_device, zones_for_device) -> list[str]: lines = [ "# HELP hue_sensor_available Whether the sensor's Hue device is connected.", "# TYPE hue_sensor_available gauge", "# HELP hue_temperature_celsius Temperature measured by a Hue sensor.", "# TYPE hue_temperature_celsius gauge", "# HELP hue_light_level_lux Illuminance measured by a Hue sensor.", "# TYPE hue_light_level_lux gauge", ] for resource_type, metric, field in ( ("temperature", "hue_temperature_celsius", "temperature"), ("light_level", "hue_light_level_lux", "light_level"), ): for sensor in resources[resource_type]: device_id = sensor.get("owner", {}).get("rid", "") device = devices.get(device_id, {}) labels = { "sensor": sensor.get("metadata", {}).get("name") or device.get("metadata", {}).get("name", sensor["id"]), "sensor_id": sensor["id"], "sensor_type": resource_type, "room": room_for_device.get(device_id, ""), "zones": zones_for_device.get(device_id, ""), } available = connected.get(device_id, True) lines.append(_sample("hue_sensor_available", int(available), **labels)) value = sensor.get(resource_type, {}).get(field) if resource_type == "light_level" and not sensor.get("light_level", {}).get("light_level_valid", True): value = None if available and value is not None: lines.append(_sample(metric, value, **labels)) return lines class State: def __init__(self): self.metrics = "hue_collection_success 0\n" self.healthy = False self.lock = threading.Lock() def run_collection_loop(collector: Collector, state: State, interval: float) -> None: while True: try: metrics = collector.collect() with state.lock: state.metrics, state.healthy = metrics, True except Exception: LOG.exception("Hue collection failed") with state.lock: state.metrics = "hue_collection_success 0\n" time.sleep(interval) def handler_for(state: State): class Handler(BaseHTTPRequestHandler): def do_GET(self): if self.path == "/metrics": with state.lock: body, status = state.metrics.encode(), 200 content_type = "text/plain; version=0.0.4; charset=utf-8" elif self.path == "/healthz": with state.lock: healthy = state.healthy body, status, content_type = (b"ok\n", 200, "text/plain") if healthy else (b"not ready\n", 503, "text/plain") else: body, status, content_type = b"not found\n", 404, "text/plain" self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, fmt, *args): LOG.debug(fmt, *args) return Handler def main() -> None: logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(message)s") settings = Settings.from_env() state = State() collector = Collector(HueClient(settings)) threading.Thread(target=run_collection_loop, args=(collector, state, settings.interval), daemon=True).start() LOG.info("Serving metrics on %s:%s", settings.listen_host, settings.listen_port) ThreadingHTTPServer((settings.listen_host, settings.listen_port), handler_for(state)).serve_forever() if __name__ == "__main__": main()