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

WEB_DIR = Path(__file__).resolve().parent.parent / "web"
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)


async def index(request):
    return web.FileResponse(WEB_DIR / "index.html")


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)
    save_store(data)
    return web.json_response({"ok": True})


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/dashboards", get_dashboards)
    app.router.add_put("/api/dashboards", put_dashboards)
    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 listening on port %s (data in %s)", PORT, DATA_DIR)
    web.run_app(make_app(), port=PORT, access_log=None)