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