""" 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, )