veenm/ha-dashboard-studio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dashboard Studio 2.2.0: guest invites, personal looks, more themes, Haven widgets and tiles

Browse Source 
- Vacuum card: the robot drawn beside its dock by state, last clean, maintenance
- Formula One card: the official F1 look with a user-supplied logo
- Nine more themes; per-person theme, text size and language on Admin, People
- Guest invites: an expiring link and QR code for one dashboard, served on a
  LAN-only guest port, with the Home Assistant connection filtered both ways
- Haven 2.2: home screen widgets, three quick settings tiles, several homes,
  and self-updates from the Haven that Dashboard Studio serves

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Mel van Veen
2026-10-27 00:31:50 +01:00
parent f4c98e27af
commit a5823b4726
26 changed files with 1514 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dashboard_studio/server/main.py
+305 -26
View File
@@ -9,10 +9,12 @@ Assistant with a handful of fake entities.
"""
import asyncio
import ipaddress
import json
import logging
import os
import re
import secrets
import time
from pathlib import Path
@@ -187,11 +189,12 @@ async def ha_call(app, msg):
USERS_TTL = 60
USERS_CACHE = {}
async def ha_users(app):
"""Home Assistant's people (not its system users), with whether each is an administrator. Cached for a minute."""
cached = app.get("users_cache")
cached = USERS_CACHE.get("users")
if cached and time.time() - cached[0] < USERS_TTL:
return cached[1]
rows = MockHA.USERS_LIST() if MOCK else await ha_call(app, {"type": "config/auth/list"})
@@ -199,7 +202,7 @@ async def ha_users(app):
"admin": bool(u.get("is_owner")) or "system-admin" in (u.get("group_ids") or []),
"active": u.get("is_active", True)}
for u in rows if not u.get("system_generated")}
app["users_cache"] = (time.time(), users)
USERS_CACHE["users"] = (time.time(), users)
return users
@@ -349,6 +352,11 @@ async def service_worker(request):
return web.Response(text=js, content_type="application/javascript", headers={"Cache-Control": "no-store"})
async def haven_version(request):
"""Which Haven this Dashboard Studio serves, so Haven can offer itself the update."""
return web.FileResponse(WEB_DIR / "haven.json", headers={"Content-Type": "application/json", "Cache-Control": "no-store"})
async def haven_app(request):
"""Haven, the Android app for dashboards (built from haven/ in the repository), to download on a phone."""
return web.FileResponse(WEB_DIR / "haven.apk", headers={
@@ -384,6 +392,7 @@ async def get_dashboards(request):
"dashboards": [d for d in store.get("dashboards", []) if d.get("id") in ids],
"assign": {"devices": {request.query["device"]: dev} if dev else {}, "users": {}},
"access": {"me": access_of(store, me.get("id"))},
"look": (store.get("looks") or {}).get(me.get("id")) or {},
"role": "basic",
}, headers={"Cache-Control": "no-store"})
@@ -450,12 +459,15 @@ async def activity(request):
async def camera(request):
entity_id = request.match_info["entity_id"]
return await camera_image(request.app, request.match_info["entity_id"])
async def camera_image(app, 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"]
return web.Response(body=app["mock"].camera_svg(entity_id), content_type="image/svg+xml")
session = app["http"]
async with session.get(
f"{HA_API_URL}/camera_proxy/{entity_id}",
headers={"Authorization": f"Bearer {SUPERVISOR_TOKEN}"},
@@ -495,28 +507,39 @@ def denied(req):
"error": {"code": "unauthorized", "message": "Only Home Assistant administrators can do this"}}
async def websocket(request):
client = web.WebSocketResponse(heartbeat=30, max_msg_size=0)
await client.prepare(request)
class GuardedSocket:
"""The browser's socket as the mock Home Assistant sees it, with an outbound filter in between."""
def __init__(self, ws, shape):
self.ws, self.shape = ws, shape
allow = None if await role_of(request) == "admin" else allowed_for_basic
def __aiter__(self):
return self.ws.__aiter__()
async def send_json(self, data):
data = self.shape(data)
if data is not None:
await self.ws.send_json(data)
def __getattr__(self, name):
return getattr(self.ws, name)
async def serve_ha(app, client, allow=None, shape=None):
"""Connect a browser socket to Home Assistant. `allow` vets what the browser asks; `shape` filters what comes back."""
if MOCK:
await request.app["mock"].serve(client, allow)
return client
await app["mock"].serve(GuardedSocket(client, shape) if shape else client, allow)
return
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"]
return
try:
upstream = await session.ws_connect(HA_WS_URL, heartbeat=30, max_msg_size=0)
upstream = await app["http"].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
return
try:
msg = await upstream.receive_json() # auth_required
await upstream.send_json({"type": "auth", "access_token": SUPERVISOR_TOKEN})
@@ -524,34 +547,275 @@ async def websocket(request):
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
return
await client.send_json({"type": "ready", "ha_version": msg.get("ha_version")})
async def pump(src, dst, check=None):
async for m in src:
async def inbound():
async for m in client:
if m.type == aiohttp.WSMsgType.TEXT:
if check:
if allow:
try:
req = json.loads(m.data)
except json.JSONDecodeError:
continue
if not isinstance(req, dict) or not check(req):
if not isinstance(req, dict) or not allow(req):
await client.send_json(denied(req if isinstance(req, dict) else {}))
continue
await dst.send_str(m.data)
await upstream.send_str(m.data)
elif m.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE):
break
tasks = [asyncio.create_task(pump(client, upstream, allow)), asyncio.create_task(pump(upstream, client))]
async def outbound():
async for m in upstream:
if m.type == aiohttp.WSMsgType.TEXT:
if shape:
out = shape(json.loads(m.data))
if out is not None:
await client.send_json(out)
else:
await client.send_str(m.data)
elif m.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE):
break
tasks = [asyncio.create_task(inbound()), asyncio.create_task(outbound())]
_, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for t in pending:
t.cancel()
finally:
await upstream.close()
await client.close()
async def websocket(request):
client = web.WebSocketResponse(heartbeat=30, max_msg_size=0)
await client.prepare(request)
await serve_ha(request.app, client, None if await role_of(request) == "admin" else allowed_for_basic)
return client
# ---------------------------------------------------------------- guest links
# An admin makes an invite: a link with a code to one dashboard, until a set time. It is served on a second
# port that only answers the local network, and it shows that dashboard in kiosk mode. A guest can read and
# control only the entities on that dashboard: everything else is filtered out in both directions.
INVITES_FILE = DATA_DIR / "invites.json"
GUEST_PORT = int(os.environ.get("GUEST_PORT", "8199"))
ENTITY_ID = re.compile(r"^[a-z_]+\.[a-z0-9_]+$")
def load_invites():
now = time.time()
return {k: v for k, v in load_json(INVITES_FILE, {}).items() if v.get("until", 0) > now - 7 * 86400} # ended ones linger a week
def entities_in(obj, found=None):
"""Every entity ID (and service name) mentioned anywhere in a dashboard."""
found = set() if found is None else found
if isinstance(obj, dict):
for v in obj.values():
entities_in(v, found)
elif isinstance(obj, list):
for v in obj:
entities_in(v, found)
elif isinstance(obj, str) and ENTITY_ID.match(obj):
found.add(obj)
return found
def guest_policy(allowed):
"""What a guest may ask Home Assistant, and what it may hear back."""
asked = {}
def ids_of(req):
out = []
for src in (req.get("target") or {}, req.get("service_data") or {}):
v = src.get("entity_id")
out += [v] if isinstance(v, str) else list(v or [])
return out
def allow(req):
t = req.get("type")
asked[req.get("id")] = t
if t in ("get_states", "get_config", "ping", "unsubscribe_events", "config/area_registry/list", "config/entity_registry/list", "config/device_registry/list"):
return True
if t == "subscribe_events":
return req.get("event_type") in ("state_changed", "dashboard_studio_announce")
if t == "call_service":
ids = ids_of(req)
if req.get("domain") in BLOCKED_DOMAINS or req.get("domain") == "homeassistant" and req.get("service") not in ("toggle", "turn_on", "turn_off"):
return False
return all(i in allowed for i in ids) if ids else f"{req.get('domain')}.{req.get('service')}" in allowed
if t == "todo/item/list":
return req.get("entity_id") in allowed
if t == "history/history_during_period":
return all(i in allowed for i in req.get("entity_ids") or [])
if t == "recorder/statistics_during_period":
return all(i in allowed for i in req.get("statistic_ids") or [])
return False
def shape(msg):
if not isinstance(msg, dict):
return msg
kind = asked.get(msg.get("id"))
if msg.get("type") == "result" and isinstance(msg.get("result"), list):
if kind == "get_states":
msg["result"] = [x for x in msg["result"] if x.get("entity_id") in allowed]
elif kind == "config/entity_registry/list":
msg["result"] = [x for x in msg["result"] if x.get("entity_id") in allowed]
if msg.get("type") == "event":
ev = msg.get("event") or {}
if ev.get("event_type") == "state_changed" and (ev.get("data") or {}).get("entity_id") not in allowed:
return None
return msg
return allow, shape
def lan_only(handler):
"""Guest links answer the local network only."""
async def wrapped(request):
try:
ip = ipaddress.ip_address((request.remote or "").split("%")[0])
except ValueError:
return web.Response(status=403, text="Guest links work on the local network only")
if not (ip.is_private or ip.is_loopback or ip.is_link_local):
return web.Response(status=403, text="Guest links work on the local network only")
return await handler(request)
return wrapped
ENDED_PAGE = """<!doctype html><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>Invite ended</title><body style="margin:0;min-height:100vh;display:grid;place-items:center;background:#111;color:#eee;font:17px/1.5 system-ui,sans-serif;text-align:center;padding:24px">
<div><h1 style="font-size:24px;margin:0 0 8px">This invite has ended</h1><p style="margin:0;color:#aaa">Ask your host for a new one.</p></div>"""
def invite_of(request):
inv = load_invites().get(request.match_info.get("token", ""))
if not inv or inv.get("until", 0) < time.time():
return None, None
dash = next((d for d in load_store().get("dashboards", []) if d.get("id") == inv.get("dashboard")), None)
return (inv, dash) if dash else (None, None)
@lan_only
async def guest_page(request):
inv, _ = invite_of(request)
if not inv:
return web.Response(text=ENDED_PAGE, content_type="text/html", status=410)
if not request.path.endswith("/"):
raise web.HTTPFound(request.path + "/")
return await index(request)
@lan_only
async def guest_api(request):
inv, dash = invite_of(request)
if not inv:
return web.json_response({"error": "This invite has ended"}, status=410)
what = request.match_info["what"]
if what == "whoami":
return web.json_response({"id": "guest", "name": "guest", "display": inv.get("name", ""), "admin": False, "guest": True, "until": inv["until"]})
if what == "dashboards":
return web.json_response({"version": 1, "dashboards": [dash], "assign": {"devices": {}, "users": {}},
"access": {"me": {"all": False, "dashboards": [dash["id"]], "home": dash["id"]}}, "role": "guest", "look": {}},
headers={"Cache-Control": "no-store"})
if what == "version":
return web.json_response({"version": VERSION})
if what == "devices/seen":
return web.json_response({"ok": True})
return web.Response(status=404)
@lan_only
async def guest_camera(request):
inv, dash = invite_of(request)
if not inv or request.match_info["entity_id"] not in entities_in(dash):
return web.Response(status=403)
return await camera_image(request.app["main"], request.match_info["entity_id"])
@lan_only
async def guest_static(request):
inv, _ = invite_of(request)
path = (WEB_DIR / request.match_info["path"]).resolve()
if not inv or WEB_DIR.resolve() not in path.parents or not path.is_file():
return web.Response(status=404)
return web.FileResponse(path, headers={"Cache-Control": "no-cache"})
@lan_only
async def guest_socket(request):
inv, dash = invite_of(request)
client = web.WebSocketResponse(heartbeat=30, max_msg_size=0)
await client.prepare(request)
if not inv:
await client.close()
return client
allow, shape = guest_policy(entities_in(dash))
# the connection ends when the invite does
ender = asyncio.get_event_loop().call_later(max(1, inv["until"] - time.time()), lambda: asyncio.ensure_future(client.close()))
try:
await serve_ha(request.app["main"], client, allow, shape)
finally:
ender.cancel()
return client
def make_guest_app(main):
app = web.Application()
app["main"] = main
app.router.add_get("/g/{token}", guest_page)
app.router.add_get("/g/{token}/", guest_page)
app.router.add_get("/g/{token}/static/{path:.+}", guest_static)
app.router.add_get("/g/{token}/api/camera/{entity_id}", guest_camera)
app.router.add_get("/g/{token}/ws", guest_socket)
app.router.add_route("*", "/g/{token}/api/{what:.+}", guest_api)
return app
@admin_only
async def list_invites(request):
names = {d.get("id"): d.get("name", "") for d in load_store().get("dashboards", [])}
port = await guest_port(request.app)
rows = [{"token": k, **v, "dashboardName": names.get(v.get("dashboard"), ""), "ended": v.get("until", 0) < time.time()}
for k, v in sorted(load_invites().items(), key=lambda kv: -kv[1].get("created", 0))]
return web.json_response({"invites": rows, "port": port}, headers={"Cache-Control": "no-store"})
@admin_only
async def create_invite(request):
body = await request.json()
hours = float(body.get("hours") or 0)
until = float(body.get("until") or 0) or time.time() + hours * 3600
if not body.get("dashboard") or until <= time.time() or until > time.time() + 31 * 86400:
return web.json_response({"error": "Choose a dashboard and an end within a month"}, status=400)
invites = load_invites()
token = secrets.token_urlsafe(12)
invites[token] = {"dashboard": str(body["dashboard"]), "name": str(body.get("name") or "")[:60], "until": until,
"created": time.time(), "by": (who(request) or {}).get("display", "")}
save_json(INVITES_FILE, invites)
return web.json_response({"token": token, **invites[token]})
@admin_only
async def delete_invite(request):
invites = load_invites()
invites.pop(request.match_info["token"], None)
save_json(INVITES_FILE, invites)
return web.json_response({"ok": True})
async def guest_port(app):
"""The host port the guest links are reachable on, from the app's Network settings (None while off)."""
if MOCK:
return GUEST_PORT
try:
async with app["http"].get("http://supervisor/addons/self/info", headers={"Authorization": f"Bearer {SUPERVISOR_TOKEN}"}) as r:
net = ((await r.json()).get("data") or {}).get("network") or {}
return net.get(f"{GUEST_PORT}/tcp")
except Exception: # noqa: BLE001
return None
async def on_startup(app):
app["http"] = aiohttp.ClientSession()
await app["immich"].start()
@@ -575,6 +839,7 @@ def make_app():
app.router.add_get("/manifest.webmanifest", manifest)
app.router.add_get("/sw.js", service_worker)
app.router.add_get("/haven.apk", haven_app)
app.router.add_get("/haven.json", haven_version)
app.router.add_get("/api/changelog", changelog)
app.router.add_get("/api/whoami", whoami)
app.router.add_get("/api/people", people)
@@ -582,6 +847,9 @@ def make_app():
app.router.add_get("/api/devices", get_devices)
app.router.add_delete("/api/devices/{device}", forget_device)
app.router.add_get("/api/activity", activity)
app.router.add_get("/api/invites", list_invites)
app.router.add_post("/api/invites", create_invite)
app.router.add_delete("/api/invites/{token}", delete_invite)
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)
@@ -599,5 +867,16 @@ def make_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)
LOG.info("Dashboard Studio %s listening on port %s, guest links on %s (data in %s)", VERSION, PORT, GUEST_PORT, DATA_DIR)
async def main():
app = make_app()
runner = web.AppRunner(app, access_log=None)
await runner.setup()
await web.TCPSite(runner, port=PORT).start()
guest = web.AppRunner(make_guest_app(app), access_log=None)
await guest.setup()
await web.TCPSite(guest, port=GUEST_PORT).start()
await asyncio.Event().wait()
asyncio.run(main())