323 lines
16 KiB
Python
323 lines
16 KiB
Python
"""Read-only household dashboard. Credentials stay behind the trusted edge proxy."""
|
|
import copy
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import random
|
|
import threading
|
|
import time
|
|
from contextlib import closing
|
|
from datetime import date, datetime, timedelta
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import caldav
|
|
import requests
|
|
import websocket
|
|
from requests.auth import HTTPBasicAuth
|
|
|
|
ZONE = ZoneInfo("America/Chicago")
|
|
STATIC = Path(__file__).parent / "static"
|
|
REFRESH_SECONDS = 300
|
|
STALE_SECONDS = 900
|
|
CALENDARS = [("Personal Commitments", "#FF6B6B"), ("Work", "#4ECDC4"),
|
|
("Focus", "#45B7D1"), ("Meal Plan", "#FFA07A"), ("Exercise", "#98D8C8")]
|
|
TASK_LISTS = ["Inbox", "Next", "Waiting", "Someday"]
|
|
SOURCES = {**{f"calendar:{n}": n for n, _ in CALENDARS},
|
|
**{f"tasks:{n}": n for n in TASK_LISTS},
|
|
"weather:hourly": "Hourly weather", "weather:daily": "Daily weather"}
|
|
REQUIRED = set(SOURCES) - {"weather:daily"}
|
|
FILES = {"/": ("index.html", "text/html"), "/app.js": ("app.js", "text/javascript"),
|
|
"/style.css": ("style.css", "text/css"), "/skylight.css": ("skylight.css", "text/css"), "/favicon.svg": ("favicon.svg", "image/svg+xml")}
|
|
VERSION = hashlib.sha256(Path(__file__).read_bytes() + b"".join((STATIC / f).read_bytes() for f, _ in FILES.values())).hexdigest()[:12]
|
|
|
|
|
|
def local_iso(value):
|
|
if isinstance(value, datetime):
|
|
return (value.replace(tzinfo=ZONE) if value.tzinfo is None else value.astimezone(ZONE)).isoformat()
|
|
return value.isoformat()
|
|
|
|
|
|
def serialize_event(component, name, color):
|
|
if str(component.get("STATUS", "")).upper() == "CANCELLED":
|
|
return None
|
|
start = component.decoded("DTSTART")
|
|
if not isinstance(start, date):
|
|
raise ValueError("Invalid event date")
|
|
all_day = not isinstance(start, datetime)
|
|
duration = timedelta(days=1) if all_day else timedelta(0)
|
|
end = component.decoded("DTEND") if "DTEND" in component else start + component.decoded("DURATION", duration)
|
|
if isinstance(end, datetime) != isinstance(start, datetime):
|
|
raise ValueError("Invalid event interval")
|
|
if all_day:
|
|
before, equal = end < start, end == start
|
|
else:
|
|
start_epoch = (start.replace(tzinfo=ZONE) if start.tzinfo is None else start).timestamp()
|
|
end_epoch = (end.replace(tzinfo=ZONE) if end.tzinfo is None else end).timestamp()
|
|
before, equal = end_epoch < start_epoch, end_epoch == start_epoch
|
|
if before:
|
|
raise ValueError("Invalid event interval")
|
|
return {"id": str(component.get("UID", "")), "title": str(component.get("SUMMARY", "Untitled"))[:500],
|
|
"start": local_iso(start), "end": local_iso(end), "allDay": all_day, "point": not all_day and equal,
|
|
"location": str(component.get("LOCATION", ""))[:500], "calendar": name, "color": color}
|
|
|
|
|
|
def serialize_task(component):
|
|
if str(component.get("STATUS", "")).upper() in ("COMPLETED", "CANCELLED") or int(component.get("PERCENT-COMPLETE", 0)) == 100:
|
|
return None
|
|
return {"id": str(component.get("UID", "")), "title": str(component.get("SUMMARY", "Untitled"))[:500],
|
|
"description": str(component.get("DESCRIPTION", ""))[:2000],
|
|
"due": local_iso(component.decoded("DUE")) if "DUE" in component else None}
|
|
|
|
|
|
class Store:
|
|
def __init__(self):
|
|
self.lock = threading.Lock()
|
|
self.records = {key: {"data": None, "updated": None, "error": None, "invalid": 0} for key in SOURCES}
|
|
|
|
def publish(self, key, data, invalid=0, now=None):
|
|
with self.lock:
|
|
self.records[key] = {"data": copy.deepcopy(data), "updated": time.time() if now is None else now,
|
|
"error": None, "invalid": invalid}
|
|
|
|
def fail(self, key, error):
|
|
with self.lock:
|
|
self.records[key]["error"] = error
|
|
|
|
def snapshot(self, now=None):
|
|
now = time.time() if now is None else now
|
|
with self.lock:
|
|
records = copy.deepcopy(self.records)
|
|
events, tasks, sources, updated = [], {n: [] for n in TASK_LISTS}, {}, {}
|
|
for key, record in records.items():
|
|
stamp = record["updated"]
|
|
age = max(0, int(now - stamp)) if stamp is not None else None
|
|
state = "loading" if stamp is None else "fresh"
|
|
if record["error"]:
|
|
state = "unavailable" if stamp is None else "error"
|
|
if age is not None and age > STALE_SECONDS:
|
|
state = "stale"
|
|
if state == "fresh" and record["invalid"]:
|
|
state = "degraded"
|
|
sources[key] = {"label": SOURCES[key], "updated": datetime.fromtimestamp(stamp, ZONE).isoformat() if stamp is not None else None,
|
|
"ageSeconds": age, "status": state, "invalidItems": record["invalid"]}
|
|
if stamp is not None:
|
|
updated[key] = sources[key]["updated"]
|
|
if key.startswith("calendar:"):
|
|
events.extend(record["data"] or [])
|
|
elif key.startswith("tasks:"):
|
|
tasks[key.split(":", 1)[1]] = record["data"] or []
|
|
hourly = records["weather:hourly"]["data"]
|
|
weather = copy.deepcopy(hourly) if hourly is not None else None
|
|
if weather is not None:
|
|
weather["daily"] = records["weather:daily"]["data"] or []
|
|
# Keep the old display compatible during rollout; current clients use per-source metadata.
|
|
calendar_updates = [v for k, v in updated.items() if k.startswith(("calendar:", "tasks:"))]
|
|
if calendar_updates:
|
|
updated["calendar"] = min(calendar_updates)
|
|
if "weather:hourly" in updated:
|
|
updated["weather"] = updated["weather:hourly"]
|
|
return {"version": VERSION, "timezone": str(ZONE), "generatedAt": datetime.fromtimestamp(now, ZONE).isoformat(),
|
|
"staleAfterSeconds": STALE_SECONDS, "events": sorted(events, key=lambda e: (e["start"], e["title"])),
|
|
"tasks": tasks, "weather": weather, "updated": updated, "sources": sources,
|
|
"errors": [k for k, s in sources.items() if s["status"] != "fresh"]}
|
|
|
|
|
|
class DavReader:
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.discovery_lock = threading.Lock()
|
|
self.urls = {}
|
|
self.discovered = 0
|
|
|
|
def client(self):
|
|
c = self.config
|
|
return caldav.DAVClient(url=c["dav_url"], auth=HTTPBasicAuth(c["dav_user"], c["dav_password"]), timeout=15)
|
|
|
|
def collection_url(self, name):
|
|
with self.discovery_lock:
|
|
if name not in self.urls or time.monotonic() - self.discovered > 1800:
|
|
self.discovered = time.monotonic()
|
|
try:
|
|
with self.client() as client:
|
|
urls = {calendar.name: str(calendar.url) for calendar in client.principal().calendars()}
|
|
self.urls = urls
|
|
except Exception:
|
|
if name not in self.urls:
|
|
raise
|
|
logging.warning("DAV discovery unavailable; retaining known collection URLs")
|
|
if name not in self.urls:
|
|
raise ValueError("Collection unavailable")
|
|
return self.urls[name]
|
|
|
|
def fetch(self, name, color=None):
|
|
url = self.collection_url(name)
|
|
today = datetime.now(ZONE).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
with self.client() as client:
|
|
calendar = caldav.Calendar(client=client, url=url)
|
|
if color:
|
|
# One extra day covers midnight between five-minute polls.
|
|
results = calendar.search(start=today, end=today + timedelta(days=8), event=True, expand=True)
|
|
else:
|
|
results = calendar.search(todo=True, include_completed=True)
|
|
items, invalid = [], 0
|
|
for result in results:
|
|
try:
|
|
item = serialize_event(result.icalendar_component, name, color) if color else serialize_task(result.icalendar_component)
|
|
if item is not None:
|
|
items.append(item)
|
|
except (ValueError, TypeError, KeyError, AttributeError, OverflowError):
|
|
invalid += 1
|
|
return items, invalid
|
|
|
|
|
|
def receive(ws, deadline):
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TimeoutError("Forecast deadline exceeded")
|
|
ws.settimeout(min(15, remaining))
|
|
message = json.loads(ws.recv())
|
|
if not isinstance(message, dict):
|
|
raise ValueError("Invalid weather message")
|
|
return message
|
|
|
|
|
|
def fetch_weather(config, kind="hourly"):
|
|
base, token = config["ha_url"].rstrip("/"), config["ha_access_token"]
|
|
attrs, condition = {}, None
|
|
if kind == "hourly":
|
|
with requests.get(base + "/api/states/weather.forecast_home", headers={"Authorization": "Bearer " + token}, timeout=(5, 15)) as response:
|
|
response.raise_for_status()
|
|
state = response.json()
|
|
attrs, condition = state["attributes"], state["state"]
|
|
deadline = time.monotonic() + 30
|
|
with closing(websocket.create_connection(base.replace("http://", "ws://").replace("https://", "wss://") + "/api/websocket", timeout=10)) as ws:
|
|
if receive(ws, deadline).get("type") != "auth_required":
|
|
raise ValueError("Weather handshake failed")
|
|
ws.send(json.dumps({"type": "auth", "access_token": token}))
|
|
if receive(ws, deadline).get("type") != "auth_ok":
|
|
raise ValueError("Weather authentication failed")
|
|
ws.send(json.dumps({"id": 1, "type": "weather/subscribe_forecast", "entity_id": "weather.forecast_home", "forecast_type": kind}))
|
|
while True:
|
|
message = receive(ws, deadline)
|
|
if message.get("id") != 1:
|
|
continue
|
|
if message.get("type") == "result" and not message.get("success"):
|
|
raise ValueError("Weather forecast unavailable")
|
|
if message.get("type") == "event":
|
|
forecast = message["event"]["forecast"]
|
|
break
|
|
if not isinstance(forecast, list) or not forecast:
|
|
raise ValueError("Empty forecast")
|
|
valid, invalid = [], 0
|
|
for item in forecast:
|
|
try:
|
|
if not math.isfinite(float(item["temperature"])):
|
|
raise ValueError("Invalid temperature")
|
|
stamp = datetime.fromisoformat(item["datetime"])
|
|
if stamp.tzinfo is None:
|
|
raise ValueError("Forecast timezone missing")
|
|
# Expose only display fields, never arbitrary upstream attributes.
|
|
clean = {"datetime": stamp.isoformat(), "temperature": float(item["temperature"]), "condition": str(item.get("condition", "cloudy"))}
|
|
for key in ("templow", "precipitation_probability"):
|
|
if item.get(key) is not None:
|
|
value = float(item[key])
|
|
if math.isfinite(value):
|
|
clean[key] = value
|
|
valid.append(clean)
|
|
except (ValueError, TypeError, KeyError):
|
|
invalid += 1
|
|
if not valid:
|
|
raise ValueError("No usable forecast")
|
|
valid.sort(key=lambda item: datetime.fromisoformat(item["datetime"]).timestamp())
|
|
if kind == "daily":
|
|
return valid, invalid
|
|
return {"forecast": valid, "unit": attrs.get("temperature_unit", "°F"), "condition": condition}, invalid
|
|
|
|
|
|
def refresh_source(key, fetch, store, stop):
|
|
failures = 0
|
|
while not stop.is_set():
|
|
started = time.monotonic()
|
|
try:
|
|
data, invalid = fetch()
|
|
store.publish(key, data, invalid)
|
|
failures = 0
|
|
logging.info("Source refreshed: %s (%d invalid items)", key, invalid)
|
|
except Exception as exc:
|
|
# Log types only: exception bodies and upstream responses may contain secrets.
|
|
store.fail(key, type(exc).__name__)
|
|
failures += 1
|
|
logging.warning("Source refresh failed: %s (%s)", key, type(exc).__name__)
|
|
interval = REFRESH_SECONDS if not failures else min(REFRESH_SECONDS, 30 * 2 ** min(failures - 1, 4))
|
|
delay = max(1, interval - (time.monotonic() - started))
|
|
stop.wait(delay + random.uniform(0, 5))
|
|
|
|
|
|
def validate_config(config):
|
|
for key in ("dav_url", "dav_user", "dav_password", "ha_url", "ha_access_token", "proxy_token"):
|
|
if not isinstance(config.get(key), str) or not config[key].strip():
|
|
raise ValueError("Required dashboard credential missing: " + key)
|
|
for key in ("dav_url", "ha_url"):
|
|
url = urlparse(config[key])
|
|
if url.scheme not in ("http", "https") or not url.hostname or url.username or url.password:
|
|
raise ValueError("Invalid dashboard upstream URL: " + key)
|
|
if len(config["proxy_token"]) < 32:
|
|
raise ValueError("Proxy credential too short")
|
|
|
|
|
|
class Application:
|
|
def __init__(self, config, store):
|
|
self.proxy_token, self.store = config["proxy_token"], store
|
|
self.assets = {path: ((STATIC / filename).read_bytes(), mime) for path, (filename, mime) in FILES.items()}
|
|
html, mime = self.assets["/"]
|
|
html = html.replace(b'APP_VERSION', VERSION.encode()).replace(b'"/app.js"', f'"/app.js?v={VERSION}"'.encode()).replace(b'"/style.css"', f'"/style.css?v={VERSION}"'.encode())
|
|
html = html.replace(b'"/skylight.css"', f'"/skylight.css?v={VERSION}"'.encode())
|
|
self.assets["/"] = html, mime
|
|
|
|
def __call__(self, environ, start_response):
|
|
path, method = environ.get("PATH_INFO", "/"), environ.get("REQUEST_METHOD", "GET")
|
|
code, mime, content = 200, "application/json", b""
|
|
if method not in ("GET", "HEAD"):
|
|
code, content = 405, b'{"error":"Method not allowed"}'
|
|
elif path == "/health/live":
|
|
content = b'{"ok":true}'
|
|
elif not hmac.compare_digest(environ.get("HTTP_X_DASHBOARD_KEY", "").encode(), self.proxy_token.encode()):
|
|
code, content = 403, b'{"error":"Forbidden"}'
|
|
elif path in ("/api/dashboard", "/health"):
|
|
data = self.store.snapshot()
|
|
if path == "/health":
|
|
ready = all(data["sources"][key]["status"] == "fresh" for key in REQUIRED)
|
|
code = 200 if ready else 503
|
|
data = {"ok": ready, "sources": data["sources"]}
|
|
content = json.dumps(data, allow_nan=False).encode()
|
|
elif path in self.assets:
|
|
content, mime = self.assets[path]
|
|
else:
|
|
code, content = 404, b'{"error":"Not found"}'
|
|
names = {200: "OK", 403: "Forbidden", 404: "Not Found", 405: "Method Not Allowed", 503: "Service Unavailable"}
|
|
headers = [("Content-Type", mime + "; charset=utf-8"), ("Content-Length", str(len(content))),
|
|
("Cache-Control", "no-store"), ("X-Content-Type-Options", "nosniff"), ("Referrer-Policy", "no-referrer"),
|
|
("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'none'")]
|
|
if code == 405:
|
|
headers.append(("Allow", "GET, HEAD"))
|
|
start_response(f"{code} {names[code]}", headers)
|
|
return [] if method == "HEAD" else [content]
|
|
|
|
|
|
def create_app():
|
|
config = json.loads(Path(os.environ.get("CONFIG_FILE", "/run/secrets/dashboard.json")).read_text())
|
|
validate_config(config)
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
|
|
store, stop, dav = Store(), threading.Event(), DavReader(config)
|
|
jobs = [(f"calendar:{name}", lambda n=name, c=color: dav.fetch(n, c)) for name, color in CALENDARS]
|
|
jobs.extend((f"tasks:{name}", lambda n=name: dav.fetch(n)) for name in TASK_LISTS)
|
|
jobs.extend((f"weather:{kind}", lambda k=kind: fetch_weather(config, k)) for kind in ("hourly", "daily"))
|
|
for key, fetch in jobs:
|
|
threading.Thread(target=refresh_source, args=(key, fetch, store, stop), daemon=True, name=key).start()
|
|
return Application(config, store)
|