smart-mirror/raspi/Dashboard.py

"""
Dashboard.py
────────────
Flask app factory for the Smart Mirror dashboard.

Routes
------
  GET  /            → Jinja2 rendered dashboard (templates/dashboard.html)
  GET  /api/data    → JSON snapshot from USBRead (status + ring-buffer)
  GET  /api/alerts  → JSON log of dispatched alert events

Start via Gunicorn:
    gunicorn -c gunicorn.conf.py "Dashboard:create_app()"

Or for development:
    python Dashboard.py
"""

import json
import threading
from collections import deque
from datetime import datetime
from pathlib import Path

from flask import Flask, jsonify, render_template

import Notification
import USBRead

# ── Resolve paths relative to this file ──────────────────────────
BASE_DIR      = Path(__file__).parent
SETTINGS_PATH = BASE_DIR / "settings.json"


# ── Settings loader ───────────────────────────────────────────────

def load_settings() -> dict:
    with open(SETTINGS_PATH, encoding="utf-8") as fh:
        return json.load(fh)


# ── Flask app factory ─────────────────────────────────────────────

def create_app() -> Flask:
    settings   = load_settings()
    dash_cfg   = settings["dashboard"]
    sensor_cfg = settings.get("sensors", [])

    # ── Init subsystems ───────────────────────────────────────────
    USBRead.init(settings["usb"])
    USBRead.start_reader()

    Notification.init(settings)
    Notification.attach_to_usb()

    # ── In-memory alert log (visible in dashboard) ────────────────
    alert_log  : deque = deque(maxlen=500)
    alert_lock : threading.Lock = threading.Lock()

    # Wrap Notification.send_alert so every outbound alert is also
    # appended to the dashboard log.
    _orig_send = Notification.send_alert

    def _logging_send(subject: str, body: str) -> None:
        with alert_lock:
            alert_log.append({
                "ts":      datetime.now().isoformat(timespec="milliseconds"),
                "subject": subject,
                "body":    body,
            })
        _orig_send(subject, body)

    Notification.send_alert = _logging_send

    # ── Build Flask app ───────────────────────────────────────────
    app = Flask(
        __name__,
        template_folder = str(BASE_DIR / "templates"),
        static_folder   = str(BASE_DIR / "static"),
    )

    # Pre-serialise sensor config once (passed to template as JSON
    # data-attribute so dashboard.js can read it without an extra request)
    sensors_json = json.dumps(sensor_cfg)

    # ── Routes ────────────────────────────────────────────────────

    @app.route("/")
    def index():
        return render_template(
            "dashboard.html",
            title        = dash_cfg.get("title", "Smart Mirror"),
            sensors      = sensor_cfg,
            sensors_json = sensors_json,
            poll_ms      = dash_cfg.get("poll_interval_ms", 800),
        )

    @app.route("/api/data")
    def api_data():
        return jsonify(USBRead.get_snapshot())

    @app.route("/api/alerts")
    def api_alerts():
        with alert_lock:
            return jsonify(list(alert_log))

    return app


# ── Dev entry point ───────────────────────────────────────────────
if __name__ == "__main__":
    settings = load_settings()
    dash     = settings["dashboard"]
    app      = create_app()
    app.run(
        host  = dash.get("host",  "0.0.0.0"),
        port  = dash.get("port",  80),
        debug = False,
    )