ha-dashboard-studio/dashboard_studio/server/main.py

"""Dashboard Studio backend.

Serves the editor UI over Home Assistant ingress, stores dashboards in /data,
and proxies the Home Assistant websocket API using the Supervisor token so the
browser never needs its own long-lived access token.

Without a SUPERVISOR_TOKEN (local development) it runs against a mock Home
Assistant with a handful of fake entities.
"""

import asyncio
import json
import logging
import os
import re
import time
from pathlib import Path

import aiohttp
from aiohttp import web

from immich import Immich
from mock_ha import MockHA

LOG = logging.getLogger("dashboard_studio")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

SUPERVISOR_TOKEN = os.environ.get("SUPERVISOR_TOKEN")
# Inside Home Assistant the Supervisor always writes /data/options.json. Fake data is
# only for local development, never a silent fallback on a real install.
IN_HOME_ASSISTANT = Path("/data/options.json").exists()
MOCK = not SUPERVISOR_TOKEN and not IN_HOME_ASSISTANT
HA_WS_URL = "ws://supervisor/core/websocket"
HA_API_URL = "http://supervisor/core/api"
INGRESS_PROXY_IP = "172.30.32.2"

APP_DIR = Path(__file__).resolve().parent.parent
WEB_DIR = APP_DIR / "web"


def read_version():
    """The app version from config.yaml (copied into the image next to server/ and web/)."""
    try:
        m = re.search(r'^version:\s*"?([^"\n]+)"?', (APP_DIR / "config.yaml").read_text(), re.M)
        return m.group(1).strip() if m else "dev"
    except OSError:
        return "dev"


VERSION = read_version()
DATA_DIR = Path("/data") if Path("/data").is_dir() else Path(__file__).resolve().parent.parent / "data"
STORE_FILE = DATA_DIR / "dashboards.json"
PORT = int(os.environ.get("PORT", "8099"))


def load_options():
    """App options from the Configuration tab (/data/options.json), or env vars when developing."""
    opts = {}
    path = Path("/data/options.json")
    if path.exists():
        try:
            opts = json.loads(path.read_text())
        except (OSError, json.JSONDecodeError):
            LOG.exception("Could not read %s", path)
    return {
        "immich_url": opts.get("immich_url") or os.environ.get("IMMICH_URL", ""),
        "immich_api_key": opts.get("immich_api_key") or os.environ.get("IMMICH_API_KEY", ""),
    }


OPTIONS = load_options()


@web.middleware
async def ingress_only(request, handler):
    # Only the Supervisor ingress proxy may talk to us; it has already
    # authenticated the Home Assistant user.
    if not MOCK and request.remote != INGRESS_PROXY_IP:
        return web.Response(status=403, text="Access only via Home Assistant ingress")
    response = await handler(request)
    # Revalidate the UI files so an add-on update shows up immediately.
    if isinstance(response, web.FileResponse):
        response.headers.setdefault("Cache-Control", "no-cache")
    return response


def load_store():
    if STORE_FILE.exists():
        try:
            return json.loads(STORE_FILE.read_text())
        except (OSError, json.JSONDecodeError):
            LOG.exception("Could not read %s, starting empty", STORE_FILE)
    return {"version": 1, "dashboards": []}


def save_store(data):
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    tmp = STORE_FILE.with_suffix(".tmp")
    tmp.write_text(json.dumps(data, indent=2))
    tmp.replace(STORE_FILE)


# Version history: each dashboard keeps its last saves in history/<id>.json, newest last.
# Saves within a few minutes of each other count as one version, so typing doesn't fill the list.
HISTORY_DIR = DATA_DIR / "history"
HISTORY_MAX = 30
HISTORY_MERGE_S = 5 * 60
SAFE_ID = re.compile(r"^[\w-]{1,64}$")


def history_file(dash_id):
    return HISTORY_DIR / f"{dash_id}.json"


def load_history(dash_id):
    try:
        return json.loads(history_file(dash_id).read_text())
    except (OSError, json.JSONDecodeError):
        return []


def record_history(old, new):
    """Add every dashboard that changed in this save to its history."""
    before = {d.get("id"): d for d in old.get("dashboards", []) if isinstance(d, dict)}
    now = time.time()
    for d in new.get("dashboards", []):
        dash_id = d.get("id") if isinstance(d, dict) else None
        if not isinstance(dash_id, str) or not SAFE_ID.match(dash_id) or before.get(dash_id) == d:
            continue
        entries = load_history(dash_id)
        if not entries and dash_id in before:  # the first time: keep what was there before this save
            try:
                saved_at = min(STORE_FILE.stat().st_mtime, now - HISTORY_MERGE_S - 1)
            except OSError:
                saved_at = now - HISTORY_MERGE_S - 1
            entries.append({"at": saved_at, "data": before[dash_id]})
        if entries and now - entries[-1]["at"] < HISTORY_MERGE_S:
            entries[-1] = {"at": now, "data": d}
        else:
            entries.append({"at": now, "data": d})
        HISTORY_DIR.mkdir(parents=True, exist_ok=True)
        tmp = history_file(dash_id).with_suffix(".tmp")
        tmp.write_text(json.dumps(entries[-HISTORY_MAX:]))
        tmp.replace(history_file(dash_id))


async def index(request):
    # Stamp the version on every asset URL, so an update can never be served from a stale cache.
    html = (WEB_DIR / "index.html").read_text()
    # Locally, files change without a new version number: stamp them with their last change instead.
    stamp = VERSION if not MOCK else f"{VERSION}-{int(max(f.stat().st_mtime for f in WEB_DIR.iterdir()))}"
    html = re.sub(r'(static/[\w.-]+\.(?:js|css|png))"', rf'\1?v={stamp}"', html)
    html = html.replace("__VERSION__", VERSION)
    return web.Response(text=html, content_type="text/html", headers={"Cache-Control": "no-store"})


USERS_FILE = DATA_DIR / "users.json"


def who(request):
    """The Home Assistant user, from the headers the Supervisor's ingress proxy adds."""
    if MOCK:
        return {"id": os.environ.get("MOCK_USER_ID", "mock-user-alex"), "name": "alex", "display": "Alex"}
    h = request.headers
    uid = h.get("X-Remote-User-Id")
    return {"id": uid, "name": h.get("X-Remote-User-Name", ""), "display": h.get("X-Remote-User-Display-Name", "")} if uid else None


def load_users():
    try:
        return json.loads(USERS_FILE.read_text())
    except (OSError, json.JSONDecodeError):
        return {}


async def whoami(request):
    me = who(request)
    if me:  # remember who has used the app, so dashboards can be assigned to them
        users = load_users()
        if users.get(me["id"], {}).get("display") != me["display"] or me["id"] not in users:
            users[me["id"]] = {"name": me["name"], "display": me["display"]}
            DATA_DIR.mkdir(parents=True, exist_ok=True)
            USERS_FILE.write_text(json.dumps(users, indent=2))
    return web.json_response(me or {}, headers={"Cache-Control": "no-store"})


async def users(request):
    return web.json_response([{"id": k, **v} for k, v in load_users().items()], headers={"Cache-Control": "no-store"})


async def changelog(request):
    try:
        text = (APP_DIR / "CHANGELOG.md").read_text()
    except OSError:
        text = ""
    return web.Response(text=text, content_type="text/markdown", headers={"Cache-Control": "no-store"})


async def version(request):
    return web.json_response({"version": VERSION}, headers={"Cache-Control": "no-store"})


def all_widgets(widgets):
    for w in widgets or []:
        if isinstance(w, dict):
            yield w
            yield from all_widgets((w.get("cfg") or {}).get("children"))


async def manifest(request):
    """The web app manifest, so the dashboard can be installed on a phone's home screen.
    Widgets marked as a shortcut appear when you long-press the app icon (Android)."""
    shortcuts = []
    for d in load_store().get("dashboards", []):
        for t in d.get("tabs", []):
            for w in all_widgets(t.get("widgets")):
                if w.get("shortcut") and len(shortcuts) < 4:
                    shortcuts.append({"name": str(w["shortcut"])[:40], "url": f"./?ds_action={w.get('id')}",
                                      "icons": [{"src": "static/icon-192.png", "sizes": "192x192", "type": "image/png"}]})
    body = {
        "name": "Dashboard Studio", "short_name": "Dashboards", "id": "./", "start_url": "./", "scope": "./",
        "display": "standalone", "background_color": "#0f1526", "theme_color": "#0f1526",
        "icons": [{"src": "static/icon-192.png", "sizes": "192x192", "type": "image/png"},
                  {"src": "static/icon-512.png", "sizes": "512x512", "type": "image/png"},
                  {"src": "static/app-icon.svg", "sizes": "any", "type": "image/svg+xml"}],
        "shortcuts": shortcuts,
    }
    return web.json_response(body, content_type="application/manifest+json", headers={"Cache-Control": "no-store"})


async def service_worker(request):
    """Network first; the last good copy of the page, its files and the dashboards when the network fails."""
    js = (WEB_DIR / "sw.js").read_text().replace("__VERSION__", VERSION)
    return web.Response(text=js, content_type="application/javascript", headers={"Cache-Control": "no-store"})


async def android_app(request):
    """The Android app (built from android/ in the repository), to download on a phone."""
    return web.FileResponse(WEB_DIR / "dashboard-studio.apk", headers={
        "Content-Type": "application/vnd.android.package-archive",
        "Content-Disposition": 'attachment; filename="Dashboard Studio.apk"', "Cache-Control": "no-store"})


async def get_dashboards(request):
    return web.json_response(load_store())


async def put_dashboards(request):
    try:
        data = await request.json()
    except json.JSONDecodeError:
        return web.json_response({"error": "invalid json"}, status=400)
    if not isinstance(data, dict) or not isinstance(data.get("dashboards"), list):
        return web.json_response({"error": "expected {dashboards: [...]}"}, status=400)
    try:
        record_history(load_store(), data)
    except OSError:
        LOG.exception("Could not write the version history")
    save_store(data)
    return web.json_response({"ok": True})


def widget_count(d):
    return sum(len(t.get("widgets", [])) for t in d.get("tabs", []) if isinstance(t, dict))


async def get_history(request):
    dash_id = request.match_info["dash_id"]
    if not SAFE_ID.match(dash_id):
        return web.Response(status=400)
    entries = load_history(dash_id)
    return web.json_response([
        {"at": e["at"], "name": e["data"].get("name", ""), "tabs": len(e["data"].get("tabs", [])), "widgets": widget_count(e["data"])}
        for e in reversed(entries)
    ], headers={"Cache-Control": "no-store"})


async def get_history_version(request):
    dash_id = request.match_info["dash_id"]
    if not SAFE_ID.match(dash_id):
        return web.Response(status=400)
    try:
        at = float(request.match_info["at"])
    except ValueError:
        return web.Response(status=400)
    for e in load_history(dash_id):
        if abs(e["at"] - at) < 0.001:
            return web.json_response(e["data"], headers={"Cache-Control": "no-store"})
    return web.Response(status=404)


async def camera(request):
    entity_id = request.match_info["entity_id"]
    if not entity_id.startswith("camera.") or "/" in entity_id:
        return web.Response(status=400)
    if MOCK:
        return web.Response(body=request.app["mock"].camera_svg(entity_id), content_type="image/svg+xml")
    session = request.app["http"]
    async with session.get(
        f"{HA_API_URL}/camera_proxy/{entity_id}",
        headers={"Authorization": f"Bearer {SUPERVISOR_TOKEN}"},
    ) as resp:
        body = await resp.read()
        return web.Response(
            status=resp.status,
            body=body,
            content_type=resp.content_type or "image/jpeg",
            headers={"Cache-Control": "no-store"},
        )


async def websocket(request):
    client = web.WebSocketResponse(heartbeat=30, max_msg_size=0)
    await client.prepare(request)

    if MOCK:
        await request.app["mock"].serve(client)
        return client
    if not SUPERVISOR_TOKEN:
        await client.send_json({"type": "error", "message": "Dashboard Studio has no access to Home Assistant (SUPERVISOR_TOKEN is missing). Restart the app; if this stays, reinstall it."})
        await client.close()
        return client

    session = request.app["http"]
    try:
        upstream = await session.ws_connect(HA_WS_URL, heartbeat=30, max_msg_size=0)
    except aiohttp.ClientError as err:
        LOG.error("Cannot reach Home Assistant websocket: %s", err)
        await client.send_json({"type": "error", "message": "Cannot reach Home Assistant"})
        await client.close()
        return client

    try:
        msg = await upstream.receive_json()  # auth_required
        await upstream.send_json({"type": "auth", "access_token": SUPERVISOR_TOKEN})
        msg = await upstream.receive_json()
        if msg.get("type") != "auth_ok":
            LOG.error("Home Assistant rejected the Supervisor token: %s", msg)
            await client.send_json({"type": "error", "message": "Home Assistant auth failed"})
            return client
        await client.send_json({"type": "ready", "ha_version": msg.get("ha_version")})

        async def pump(src, dst):
            async for m in src:
                if m.type == aiohttp.WSMsgType.TEXT:
                    await dst.send_str(m.data)
                elif m.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE):
                    break

        tasks = [asyncio.create_task(pump(client, upstream)), asyncio.create_task(pump(upstream, client))]
        _, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        for t in pending:
            t.cancel()
    finally:
        await upstream.close()
        await client.close()
    return client


async def on_startup(app):
    app["http"] = aiohttp.ClientSession()
    await app["immich"].start()
    if MOCK:
        app["mock"] = MockHA()
        app["mock_task"] = asyncio.create_task(app["mock"].simulate())
        LOG.warning("No SUPERVISOR_TOKEN: running against mock Home Assistant")


async def on_cleanup(app):
    await app["http"].close()
    await app["immich"].close()
    if MOCK:
        app["mock_task"].cancel()


def make_app():
    app = web.Application(middlewares=[ingress_only], client_max_size=8 * 1024 * 1024)
    app.router.add_get("/", index)
    app.router.add_get("/api/version", version)
    app.router.add_get("/manifest.webmanifest", manifest)
    app.router.add_get("/sw.js", service_worker)
    app.router.add_get("/android.apk", android_app)
    app.router.add_get("/api/changelog", changelog)
    app.router.add_get("/api/whoami", whoami)
    app.router.add_get("/api/users", users)
    app.router.add_get("/api/dashboards", get_dashboards)
    app.router.add_put("/api/dashboards", put_dashboards)
    app.router.add_get("/api/history/{dash_id}", get_history)
    app.router.add_get("/api/history/{dash_id}/{at}", get_history_version)
    app.router.add_get("/api/camera/{entity_id}", camera)
    app.router.add_get("/ws", websocket)
    app["immich"] = Immich(OPTIONS["immich_url"], OPTIONS["immich_api_key"], mock=MOCK)
    app.router.add_routes(app["immich"].routes())
    app.router.add_static("/static", WEB_DIR)
    app.on_startup.append(on_startup)
    app.on_cleanup.append(on_cleanup)
    return app


if __name__ == "__main__":
    if IN_HOME_ASSISTANT and not SUPERVISOR_TOKEN:
        LOG.error("SUPERVISOR_TOKEN is missing; the app can't talk to Home Assistant")
    LOG.info("Dashboard Studio %s listening on port %s (data in %s)", VERSION, PORT, DATA_DIR)
    web.run_app(make_app(), port=PORT, access_log=None)