ignace/service_hue_collector
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1sT poc

Browse Source
This commit is contained in:
ignace
2026-08-30 14:26:11 +02:00
parent 11fa106723
commit f8d5f5466c
14 changed files with 514 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+6
View File
@@ -0,0 +1,6 @@
HUE_BRIDGE_HOST=192.168.1.2
HUE_APPLICATION_KEY=replace-with-your-hue-application-key
HUE_VERIFY_TLS=false
COLLECT_INTERVAL_SECONDS=30
LOG_LEVEL=INFO
.gitignore
+5
View File
@@ -0,0 +1,5 @@
.env
__pycache__/
.pytest_cache/
*.py[cod]
Dockerfile
+11
View File
@@ -0,0 +1,11 @@
FROM python:3.13-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY hue_collector ./hue_collector
USER 65532:65532
EXPOSE 8000
CMD ["python", "-m", "hue_collector"]
README.md
+62 -1
View File
@@ -1,2 +1,63 @@
# service_hue_collector
# Hue collector
A local-only monitoring stack for a Philips Hue bridge. The Python service reads the
Hue v2 API, exposes Prometheus metrics, VictoriaMetrics stores them, and Grafana ships
with a provisioned dashboard.
## Metrics
- For every room and zone: available lights, shining lights, percentages shining,
off, and unavailable, plus average brightness of the shining lights.
- For every temperature and illuminance sensor: availability and its latest reading,
labelled with its room and zones.
A device is available when its Hue `zigbee_connectivity` resource is connected. A
light is shining when it is available and reports `on=true`. If a device has no
connectivity resource (for example some bridge-owned resources), it is considered
available.
## Setup
1. Find your bridge IP in the Hue app under **Settings → My Hue system → System
information**.
2. Create an application key while physically near the bridge:
```bash
curl -k -X POST https://BRIDGE_IP/api \
-H 'Content-Type: application/json' \
-d '{"devicetype":"local-hue-collector"}'
```
Press the bridge link button immediately before running the command. Copy the
returned `username`; that is the application key.
3. Configure and start the stack:
```bash
cp .env.example .env
# Edit .env with the bridge IP and application key.
docker compose up --build -d
```
4. Open [Grafana](http://localhost:3000) and sign in with `admin` / `admin`. The
dashboard is in the **Hue** folder. Change this development password if the port
will ever be exposed beyond localhost.
Useful local endpoints:
- Collector metrics: <http://localhost:8000/metrics>
- VictoriaMetrics UI: <http://localhost:8428/vmui/>
- Grafana: <http://localhost:3000>
All published ports bind to `127.0.0.1`. Named Docker volumes retain metrics and
Grafana state across restarts. TLS verification is off by default because Hue bridges
normally use a self-signed certificate; set `HUE_VERIFY_TLS=true` if yours has a
trusted certificate.
## Development
```bash
python -m venv .venv
. .venv/bin/activate
pip install -r requirements.txt
python -m unittest discover -s tests
HUE_BRIDGE_HOST=... HUE_APPLICATION_KEY=... python -m hue_collector
```
config/grafana/dashboards/hue.json
+49
View File
@@ -0,0 +1,49 @@
{
"annotations": {"list": []},
"editable": true,
"graphTooltip": 1,
"panels": [
{
"type": "timeseries", "title": "Lights shining", "id": 1,
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 0},
"fieldConfig": {"defaults": {"unit": "percent", "min": 0, "max": 100}, "overrides": []},
"targets": [
{"expr": "hue_group_lights_shining_percent", "legendFormat": "shining — {{group_type}}: {{group}}", "refId": "A"},
{"expr": "hue_group_lights_off_percent", "legendFormat": "off — {{group_type}}: {{group}}", "refId": "B"},
{"expr": "hue_group_lights_unavailable_percent", "legendFormat": "unavailable — {{group_type}}: {{group}}", "refId": "C"}
]
},
{
"type": "timeseries", "title": "Average intensity of shining lights", "id": 2,
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 0},
"fieldConfig": {"defaults": {"unit": "percent", "min": 0, "max": 100}, "overrides": []},
"targets": [{"expr": "hue_group_average_brightness_percent", "legendFormat": "{{group_type}}: {{group}}", "refId": "A"}]
},
{
"type": "timeseries", "title": "Temperature", "id": 3,
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 8},
"fieldConfig": {"defaults": {"unit": "celsius"}, "overrides": []},
"targets": [{"expr": "hue_temperature_celsius", "legendFormat": "{{sensor}} ({{room}})", "refId": "A"}]
},
{
"type": "timeseries", "title": "Light intensity", "id": 4,
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 8},
"fieldConfig": {"defaults": {"unit": "lux"}, "overrides": []},
"targets": [{"expr": "hue_light_level_lux", "legendFormat": "{{sensor}} ({{room}})", "refId": "A"}]
},
{
"type": "table", "title": "Unavailable sensors", "id": 5,
"gridPos": {"h": 7, "w": 24, "x": 0, "y": 16},
"targets": [{"expr": "hue_sensor_available == 0", "format": "table", "instant": true, "refId": "A"}]
}
],
"refresh": "30s",
"schemaVersion": 42,
"tags": ["hue"],
"templating": {"list": []},
"time": {"from": "now-24h", "to": "now"},
"timezone": "browser",
"title": "Philips Hue",
"uid": "philips-hue-local",
"version": 1
}
config/grafana/provisioning/dashboards/default.yml
+11
View File
@@ -0,0 +1,11 @@
apiVersion: 1
providers:
- name: Hue
orgId: 1
folder: Hue
type: file
disableDeletion: true
updateIntervalSeconds: 30
options:
path: /var/lib/grafana/dashboards
config/grafana/provisioning/datasources/victoriametrics.yml
+10
View File
@@ -0,0 +1,10 @@
apiVersion: 1
datasources:
- name: VictoriaMetrics
uid: victoriametrics
type: prometheus
access: proxy
url: http://victoriametrics:8428
isDefault: true
editable: false
config/victoriametrics/scrape.yml
+8
View File
@@ -0,0 +1,8 @@
global:
scrape_interval: 30s
scrape_configs:
- job_name: hue
static_configs:
- targets: [hue-collector:8000]
docker-compose.yml
+52
View File
@@ -0,0 +1,52 @@
services:
hue-collector:
build: .
restart: unless-stopped
environment:
HUE_BRIDGE_HOST: ${HUE_BRIDGE_HOST:-}
HUE_APPLICATION_KEY: ${HUE_APPLICATION_KEY:-}
HUE_VERIFY_TLS: ${HUE_VERIFY_TLS:-false}
COLLECT_INTERVAL_SECONDS: ${COLLECT_INTERVAL_SECONDS:-30}
LOG_LEVEL: ${LOG_LEVEL:-INFO}
ports:
- "127.0.0.1:8000:8000"
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/healthz')"]
interval: 30s
timeout: 5s
retries: 3
victoriametrics:
image: victoriametrics/victoria-metrics:v1.126.0
restart: unless-stopped
command:
- -storageDataPath=/victoria-metrics-data
- -promscrape.config=/etc/victoriametrics/scrape.yml
- -retentionPeriod=1y
volumes:
- victoria-metrics-data:/victoria-metrics-data
- ./config/victoriametrics/scrape.yml:/etc/victoriametrics/scrape.yml:ro
ports:
- "127.0.0.1:8428:8428"
depends_on:
- hue-collector
grafana:
image: grafana/grafana:12.3.3
restart: unless-stopped
environment:
GF_SECURITY_ADMIN_USER: admin
GF_SECURITY_ADMIN_PASSWORD: admin
GF_USERS_ALLOW_SIGN_UP: "false"
volumes:
- grafana-data:/var/lib/grafana
- ./config/grafana/provisioning:/etc/grafana/provisioning:ro
- ./config/grafana/dashboards:/var/lib/grafana/dashboards:ro
ports:
- "127.0.0.1:3000:3000"
depends_on:
- victoriametrics
volumes:
victoria-metrics-data:
grafana-data:
hue_collector/__init__.py
+2
View File
@@ -0,0 +1,2 @@
"""Philips Hue Prometheus collector."""
hue_collector/__main__.py
+4
View File
@@ -0,0 +1,4 @@
from .collector import main
main()
hue_collector/collector.py
+243
View File
@@ -0,0 +1,243 @@
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_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_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()
requirements.txt
+2
View File
@@ -0,0 +1,2 @@
requests==2.32.5
tests/test_collector.py
+49
View File
@@ -0,0 +1,49 @@
import unittest
from hue_collector.collector import Collector
class FakeClient:
data = {
"device": [
{"id": "d1", "metadata": {"name": "Motion"}},
{"id": "d2", "metadata": {"name": "Lamp"}},
{"id": "d3", "metadata": {"name": "Offline lamp"}},
],
"room": [{"id": "r1", "metadata": {"name": "Office"}, "children": [{"rid": "d1"}, {"rid": "d2"}, {"rid": "d3"}]}],
"zone": [{"id": "z1", "metadata": {"name": "Downstairs"}, "children": [{"rid": "d2"}]}],
"light": [
{"id": "l1", "owner": {"rid": "d2"}, "on": {"on": True}, "dimming": {"brightness": 60}},
{"id": "l2", "owner": {"rid": "d3"}, "on": {"on": True}, "dimming": {"brightness": 100}},
],
"temperature": [{"id": "t1", "owner": {"rid": "d1"}, "temperature": {"temperature": 21.5}}],
"light_level": [{"id": "s1", "owner": {"rid": "d1"}, "light_level": {"light_level": 123.4}}],
"zigbee_connectivity": [
{"owner": {"rid": "d1"}, "status": "connected"},
{"owner": {"rid": "d2"}, "status": "connected"},
{"owner": {"rid": "d3"}, "status": "disconnected"},
],
}
def get(self, resource):
return self.data[resource]
class CollectorTests(unittest.TestCase):
def setUp(self):
self.metrics = Collector(FakeClient()).collect()
def test_group_counts_only_connected_lights(self):
self.assertIn('hue_group_lights_available{group="Office",group_id="r1",group_type="room"} 1', self.metrics)
self.assertIn('hue_group_lights_shining_percent{group="Office",group_id="r1",group_type="room"} 100.0', self.metrics)
self.assertIn('hue_group_lights_off_percent{group="Office",group_id="r1",group_type="room"} 0.0', self.metrics)
self.assertIn('hue_group_lights_unavailable_percent{group="Office",group_id="r1",group_type="room"} 50.0', self.metrics)
self.assertIn('hue_group_average_brightness_percent{group="Office",group_id="r1",group_type="room"} 60.0', self.metrics)
def test_sensor_values_have_location(self):
self.assertIn('hue_temperature_celsius{room="Office",sensor="Motion",sensor_id="t1",sensor_type="temperature",zones=""} 21.5', self.metrics)
self.assertIn('hue_light_level_lux{room="Office",sensor="Motion",sensor_id="s1",sensor_type="light_level",zones=""} 123.4', self.metrics)
if __name__ == "__main__":
unittest.main()