from __future__ import annotations import logging import os import threading import time from io import BytesIO from flask import Flask, Response, abort, render_template_string, request, send_file, url_for from .collector import Collector, HueClient, Settings from .rrd_store import RRDStore LOG = logging.getLogger("hue-collector") PERIODS = ("6h", "24h", "7d", "30d") PAGE = ''' Hue history

Philips Hue

{{ status }}{% if error %} — {{ error }}{% endif %}

Rooms and zones

{% for item in groups %}

{{ item.type|title }}: {{ item.name }}

Light history
{% else %}

No groups collected yet.

{% endfor %}

Sensors

{% for item in sensors %}

{{ item.name }}

{{ item.room or item.zones or 'No room' }} · {% if item.value is none %}unavailable{% else %}{{ '%.1f'|format(item.value) }} {{ item.unit }}{% endif %}

Sensor history
{% else %}

No sensors collected yet.

{% endfor %}''' class State: def __init__(self): self.snapshot = {"groups": [], "sensors": []} self.last_success: float | None = None self.error: str | None = None self.lock = threading.Lock() def collection_loop(collector: Collector, store: RRDStore, state: State, interval: int) -> None: while True: started = time.monotonic() try: snapshot = collector.collect() store.update(snapshot) with state.lock: state.snapshot, state.last_success, state.error = snapshot, time.time(), None except Exception as exc: LOG.exception("Hue collection failed") with state.lock: state.error = str(exc) time.sleep(max(1, interval - (time.monotonic() - started))) def create_app(settings: Settings | None = None, *, start_collector: bool = True) -> Flask: settings = settings or Settings.from_env() app = Flask(__name__) state = State() store = RRDStore(settings.data_dir, settings.interval) app.extensions["hue_state"] = state app.extensions["hue_rrd_store"] = store if start_collector: collector = Collector(HueClient(settings)) threading.Thread(target=collection_loop, args=(collector, store, state, settings.interval), daemon=True).start() @app.get("/") def dashboard(): period = request.args.get("range", "24h") if period not in PERIODS: period = "24h" with state.lock: groups = sorted(state.snapshot["groups"], key=lambda item: (item["type"], item["name"].lower())) sensors = sorted(state.snapshot["sensors"], key=lambda item: (item["type"], item["name"].lower())) last_success, error = state.last_success, state.error status = time.strftime("Last update: %Y-%m-%d %H:%M:%S", time.localtime(last_success)) if last_success else "Waiting for first collection" return render_template_string(PAGE, groups=groups, sensors=sensors, status=status, error=error, periods=PERIODS, period=period) @app.get("/graph//.png") def graph(kind: str, item_id: str): if kind not in {"group", "sensor"}: abort(404) with state.lock: collection = state.snapshot["groups" if kind == "group" else "sensors"] item = next((value for value in collection if value["id"] == item_id), None) if item is None: abort(404) try: image = store.graph(kind, item, request.args.get("range", "24h")) except (ValueError, FileNotFoundError): abort(404) return send_file(BytesIO(image), mimetype="image/png", max_age=30) @app.get("/healthz") def health(): with state.lock: healthy = state.last_success is not None return Response("ok\n" if healthy else "not ready\n", status=200 if healthy else 503, mimetype="text/plain") return app def main() -> None: logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(message)s") settings = Settings.from_env() app = create_app(settings) LOG.info("Serving Hue dashboard on http://%s:%s", settings.listen_host, settings.listen_port) app.run(host=settings.listen_host, port=settings.listen_port, debug=False, use_reloader=False)