veenm/ha-dashboard-studio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
78aa85a82f63183b00edb1c8d2ba680560e9d980
ha-dashboard-studio/dashboard_studio/server/mock_ha.py
T
veenm 78aa85a82f Dashboard Studio 1.10.0: phone dashboards, room card, changelog 
- Phone dashboards: a 4-column layout designed at phone width, with
  conversion both ways; on phones the top bar is hidden and comes back
  with a touch-and-hold on an empty spot
- Room card built from Home Assistant areas (area, device and entity
  registries): climate, lights on with an all-lights toggle, thermostat,
  open doors and windows, motion, media, and device buttons
- CHANGELOG.md for Home Assistant's Changelog tab, served at
  /api/changelog and shown under About > What's new

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-26 00:06:01 +01:00

341 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""A tiny fake of the Home Assistant websocket API for local development."""
import asyncio
import json
import math
import random
from datetime import datetime, timedelta, timezone
from aiohttp import WSMsgType
def now_iso():
return datetime.now(timezone.utc).isoformat()
def entity(eid, state, /, **attrs):
ts = now_iso()
return {
"entity_id": eid,
"state": state,
"attributes": attrs,
"last_changed": ts,
"last_updated": ts,
"context": {"id": "mock"},
}
class MockHA:
def __init__(self):
today = datetime.now().astimezone().date()
def pickup(kind, days):
# Mirrors the Afvalwijzer integration: ISO date state plus collection-date attributes.
if days is None:
return entity(f"sensor.afvalwijzer_1234ab_12_{kind}", "geen", friendly_name=f"Afvalwijzer {kind}", last_update=now_iso(),
days_until_collection_date=None, is_collection_date_today=False, is_collection_date_tomorrow=False, is_collection_date_day_after_tomorrow=False)
return entity(f"sensor.afvalwijzer_1234ab_12_{kind}", (today + timedelta(days=days)).isoformat(), friendly_name=f"Afvalwijzer {kind}", last_update=now_iso(),
days_until_collection_date=days, is_collection_date_today=days == 0, is_collection_date_tomorrow=days == 1, is_collection_date_day_after_tomorrow=days == 2)
waste = [pickup("gft", 1), pickup("pmd", 1), pickup("papier", 6), pickup("restafval", 8), pickup("textiel", 23), pickup("kerstbomen", None),
entity("sensor.afvalwijzer_1234ab_12_tomorrow", "gft, pmd", friendly_name="Afvalwijzer Morgen"),
entity("sensor.afvalwijzer_1234ab_12_next_in_days", "1", friendly_name="Afvalwijzer Dagen tot volgende ophaling")]
self.states = {
e["entity_id"]: e
for e in waste + [
entity("light.living_room", "on", friendly_name="Living Room", brightness=180, supported_color_modes=["brightness"]),
entity("light.reading_lamp", "on", friendly_name="Reading Lamp", brightness=140, rgb_color=[255, 140, 60], supported_color_modes=["rgb"]),
entity("light.kitchen", "off", friendly_name="Kitchen", brightness=None, supported_color_modes=["brightness"]),
entity("light.bedroom", "on", friendly_name="Bedroom", brightness=90, supported_color_modes=["brightness"]),
entity("light.downstairs", "on", friendly_name="Downstairs", brightness=160, supported_color_modes=["brightness"],
entity_id=["light.living_room", "light.kitchen", "light.reading_lamp"]),
entity("vacuum.robot", "docked", friendly_name="Robot Vacuum", battery_level=82, fan_speed="standard",
fan_speed_list=["quiet", "standard", "turbo", "max"], status="Charging"),
entity("sensor.robot_vacuum_cleaned_area", "23", friendly_name="Robot Vacuum Cleaned area", unit_of_measurement="m²"),
entity("sensor.robot_vacuum_cleaning_time", "34", friendly_name="Robot Vacuum Cleaning time", unit_of_measurement="min", device_class="duration"),
entity("sensor.robot_vacuum_filter_left", "72", friendly_name="Robot Vacuum Filter left", unit_of_measurement="%"),
entity("switch.coffee_machine", "off", friendly_name="Coffee Machine"),
entity("fan.office", "off", friendly_name="Office Fan"),
entity("sensor.living_room_temperature", "21.4", friendly_name="Living Room Temperature", unit_of_measurement="°C", device_class="temperature"),
entity("sensor.outdoor_temperature", "12.8", friendly_name="Outdoor Temperature", unit_of_measurement="°C", device_class="temperature"),
entity("sensor.living_room_humidity", "48", friendly_name="Living Room Humidity", unit_of_measurement="%", device_class="humidity"),
entity("sensor.power_usage", "640", friendly_name="Power Usage", unit_of_measurement="W", device_class="power"),
entity("sensor.solar_production", "1820", friendly_name="Solar Production", unit_of_measurement="W", device_class="power"),
entity("sensor.phone_battery", "76", friendly_name="Phone Battery", unit_of_measurement="%", device_class="battery"),
entity("binary_sensor.front_door", "off", friendly_name="Front Door", device_class="door"),
entity("binary_sensor.hallway_motion", "off", friendly_name="Hallway Motion", device_class="motion"),
entity("climate.thermostat", "heat", friendly_name="Thermostat", current_temperature=20.5, temperature=21, min_temp=7, max_temp=30, hvac_modes=["off", "heat", "auto"]),
entity("cover.garage_door", "closed", friendly_name="Garage Door"),
entity("lock.front_door", "locked", friendly_name="Front Door Lock"),
entity("media_player.living_room_speaker", "playing", friendly_name="Living Room Speaker", media_title="Blue in Green", media_artist="Miles Davis", media_album_name="Kind of Blue", volume_level=0.35, entity_picture="data:image/svg+xml;utf8,<svg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 100 100'><defs><linearGradient id='g' x1='0' y1='0' x2='1' y2='1'><stop offset='0' stop-color='%23234'/><stop offset='1' stop-color='%23c58'/></linearGradient></defs><rect width='100' height='100' fill='url(%23g)'/><circle cx='50' cy='50' r='22' fill='none' stroke='%23fff' stroke-width='3' opacity='.7'/></svg>"),
entity("media_player.kitchen_speaker", "paused", friendly_name="Kitchen Speaker", media_title="Bohemian Rhapsody", media_artist="Queen", volume_level=0.2),
entity("media_player.tv", "off", friendly_name="TV"),
entity("sensor.washer_status", "running", friendly_name="Washer Status"),
entity("sensor.washer_remaining_time", "42", friendly_name="Washer Remaining Time", unit_of_measurement="min", device_class="duration"),
entity("sensor.washer_progress", "64", friendly_name="Washer Progress", unit_of_measurement="%"),
entity("sensor.washer_program", "cotton_40", friendly_name="Washer Program"),
entity("sensor.washer_power", "450", friendly_name="Washer Power", unit_of_measurement="W", device_class="power"),
entity("sensor.dryer_status", "off", friendly_name="Dryer Status"),
entity("sensor.dryer_remaining_time", "0", friendly_name="Dryer Remaining Time", unit_of_measurement="min", device_class="duration"),
entity("scene.movie_night", "2026-10-24T20:00:00+00:00", friendly_name="Movie Night"),
entity("script.good_morning", "off", friendly_name="Good Morning"),
entity("weather.home", "partlycloudy", friendly_name="Home", temperature=13, humidity=71, wind_speed=14),
entity("camera.front_yard", "idle", friendly_name="Front Yard"),
entity("person.alex", "home", friendly_name="Alex", user_id="mock-user-alex"),
entity("person.sam", "not_home", friendly_name="Sam", user_id="mock-user-sam"),
entity("input_boolean.guest_mode", "off", friendly_name="Guest Mode"),
entity("calendar.family", "off", friendly_name="Family"),
entity("calendar.work", "on", friendly_name="Work"),
entity("sensor.model_y_battery", "68", friendly_name="Model Y Battery", unit_of_measurement="%", device_class="battery"),
entity("sensor.model_y_range", "341", friendly_name="Model Y Range", unit_of_measurement="km", device_class="distance"),
entity("binary_sensor.model_y_charging", "on", friendly_name="Model Y Charging", device_class="battery_charging"),
entity("sensor.model_y_charging_power", "7.4", friendly_name="Model Y Charging Power", unit_of_measurement="kW", device_class="power"),
entity("binary_sensor.model_y_plug", "on", friendly_name="Model Y Plug", device_class="plug"),
entity("number.model_y_charge_limit", "80", friendly_name="Model Y Charge Limit", unit_of_measurement="%"),
entity("lock.model_y_doors", "locked", friendly_name="Model Y Doors"),
entity("climate.model_y_climate", "off", friendly_name="Model Y Climate", current_temperature=14, temperature=21, hvac_modes=["off", "heat_cool"]),
entity("device_tracker.model_y_location", "home", friendly_name="Model Y Location"),
entity("sensor.model_y_odometer", "24816", friendly_name="Model Y Odometer", unit_of_measurement="km", device_class="distance"),
]
}
self.clients = set()
# -- state handling ---------------------------------------------------
def set_state(self, entity_id, state=None, **attrs):
old = self.states[entity_id]
new = json.loads(json.dumps(old))
if state is not None and state != old["state"]:
new["state"] = state
new["last_changed"] = now_iso()
new["attributes"].update(attrs)
new["last_updated"] = now_iso()
self.states[entity_id] = new
if entity_id.startswith("light.") and "entity_id" not in new["attributes"]:
self._sync_groups(entity_id)
event = {"event_type": "state_changed", "data": {"entity_id": entity_id, "old_state": old, "new_state": new}, "time_fired": now_iso()}
for ws, sub_ids in list(self.clients):
for sid in sub_ids:
asyncio.ensure_future(self._safe_send(ws, {"id": sid, "type": "event", "event": event}))
async def _safe_send(self, ws, payload):
try:
await ws.send_json(payload)
except Exception: # client went away
pass
def _vacuum(self, eid, service, data):
if service == "start":
self.set_state(eid, "cleaning", status="Cleaning")
elif service == "pause":
self.set_state(eid, "paused", status="Paused")
elif service in ("return_to_base", "stop"):
self.set_state(eid, "returning", status="Returning home")
asyncio.get_event_loop().call_later(6, lambda: self.set_state(eid, "docked", status="Charging"))
elif service == "set_fan_speed":
self.set_state(eid, None, fan_speed=data.get("fan_speed"))
def _sync_groups(self, changed):
for gid, g in list(self.states.items()):
members = g["attributes"].get("entity_id")
if not gid.startswith("light.") or not isinstance(members, list) or changed not in members:
continue
on = [self.states[m] for m in members if m in self.states and self.states[m]["state"] == "on"]
bright = round(sum(m["attributes"].get("brightness") or 0 for m in on) / len(on)) if on else None
self.set_state(gid, "on" if on else "off", brightness=bright)
def call_service(self, domain, service, data):
targets = data.get("entity_id") or (data.get("target") or {}).get("entity_id") or []
if isinstance(targets, str):
targets = [targets]
expanded = []
for eid in targets:
members = (self.states.get(eid) or {}).get("attributes", {}).get("entity_id")
expanded += members if eid.startswith("light.") and isinstance(members, list) else [eid]
for eid in expanded:
if eid not in self.states:
continue
if domain == "vacuum" or eid.startswith("vacuum."):
self._vacuum(eid, service, data)
continue
st = self.states[eid]["state"]
if eid.startswith("climate.") and service in ("turn_on", "turn_off", "toggle"):
on = service == "turn_on" or (service == "toggle" and st == "off")
self.set_state(eid, (self.states[eid]["attributes"].get("hvac_modes") or ["off", "heat"])[-1] if on else "off")
elif service in ("turn_on", "turn_off", "toggle"):
on = service == "turn_on" or (service == "toggle" and st != "on")
extra = {}
if eid.startswith("light."):
if "brightness" in data:
extra["brightness"] = data["brightness"]
elif "brightness_pct" in data:
extra["brightness"] = round(data["brightness_pct"] * 2.55)
elif on and not self.states[eid]["attributes"].get("brightness"):
extra["brightness"] = 255
if not on:
extra["brightness"] = None
self.set_state(eid, "on" if on else "off", **extra)
elif service == "set_temperature":
self.set_state(eid, None, temperature=data.get("temperature"))
elif service == "set_hvac_mode":
self.set_state(eid, data.get("hvac_mode"))
elif service in ("open_cover", "close_cover"):
self.set_state(eid, "open" if service == "open_cover" else "closed")
elif service == "toggle_cover" or (domain == "cover" and service == "toggle"):
self.set_state(eid, "closed" if st == "open" else "open")
elif service in ("lock", "unlock"):
self.set_state(eid, "locked" if service == "lock" else "unlocked")
elif service == "media_play_pause":
self.set_state(eid, "paused" if st == "playing" else "playing")
elif service == "volume_set":
self.set_state(eid, None, volume_level=data.get("volume_level"))
elif domain in ("scene", "script") and service in ("turn_on",):
self.set_state(eid, now_iso() if domain == "scene" else "off")
def calendar_events(self, entity_ids):
"""A repeatable week of events relative to today."""
today = datetime.now().astimezone().replace(hour=0, minute=0, second=0, microsecond=0)
plan = {
"calendar.family": [(0, 18, 30, 90, "Dinner with grandparents", "Home"), (1, None, 0, 0, "Bin collection", ""), (2, 16, 0, 60, "Swimming lessons", "Sports centre"), (5, None, 0, 0, "Emma’s birthday", ""), (6, 10, 0, 120, "Market", "Town square")],
"calendar.work": [(0, 9, 0, 30, "Stand-up", ""), (0, 14, 0, 60, "Design review", "Room 3"), (1, 11, 0, 45, "1:1 with Sam", ""), (3, 13, 30, 90, "Quarterly planning", "Main office")],
}
out = {}
for eid in entity_ids:
events = []
for day, hour, minute, dur, summary, location in plan.get(eid, []):
d = today + timedelta(days=day)
if hour is None:
events.append({"start": d.date().isoformat(), "end": (d + timedelta(days=1)).date().isoformat(), "summary": summary})
else:
start = d.replace(hour=hour, minute=minute)
events.append({"start": start.isoformat(), "end": (start + timedelta(minutes=dur)).isoformat(), "summary": summary, "location": location})
out[eid] = {"events": events}
return out
def history(self, entity_ids, start):
out = {}
start_dt = datetime.fromisoformat(start.replace("Z", "+00:00"))
span = (datetime.now(timezone.utc) - start_dt).total_seconds()
for eid in entity_ids:
st = self.states.get(eid)
if not st:
continue
try:
base = float(st["state"])
except ValueError:
out[eid] = [{"s": st["state"], "lu": start_dt.timestamp()}]
continue
pts = []
n = 96
seed = sum(map(ord, eid))
for i in range(n):
t = start_dt.timestamp() + span * i / (n - 1)
wave = math.sin(i / 10 + seed) * 0.08 + math.sin(i / 3.3 + seed) * 0.03
pts.append({"s": f"{base * (1 + wave):.1f}", "lu": t})
pts[-1]["s"] = st["state"]
out[eid] = pts
return out
def camera_svg(self, entity_id):
stamp = datetime.now().strftime("%H:%M:%S")
return f"""<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 640 360">
<defs><linearGradient id="g" x1="0" y1="0" x2="0" y2="1"><stop offset="0" stop-color="#4b6584"/><stop offset="1" stop-color="#1e272e"/></linearGradient></defs>
<rect width="640" height="360" fill="url(#g)"/>
<rect x="0" y="250" width="640" height="110" fill="#2d3e2f"/>
<polygon points="180,250 260,170 340,250" fill="#57606f"/><rect x="200" y="250" width="120" height="70" fill="#747d8c"/>
<text x="16" y="30" fill="#fff" font-family="monospace" font-size="18">{entity_id} · {stamp}</text>
</svg>""".encode()
async def simulate(self):
while True:
await asyncio.sleep(3)
for eid, spread in (
("sensor.power_usage", 60),
("sensor.solar_production", 80),
("sensor.living_room_temperature", 0.1),
("sensor.outdoor_temperature", 0.2),
):
v = float(self.states[eid]["state"]) + random.uniform(-spread, spread)
self.set_state(eid, f"{max(v, 0):.1f}" if spread < 1 else f"{max(v, 0):.0f}")
if self.states["binary_sensor.model_y_charging"]["state"] == "on":
lvl = float(self.states["sensor.model_y_battery"]["state"])
if lvl < 80:
self.set_state("sensor.model_y_battery", f"{lvl + 0.2:.1f}")
if self.states["sensor.washer_status"]["state"] == "running":
left = max(0, int(self.states["sensor.washer_remaining_time"]["state"]) - 1)
self.set_state("sensor.washer_remaining_time", str(left))
self.set_state("sensor.washer_progress", str(min(100, 100 - round(left * 36 / 42))))
if left == 0:
self.set_state("sensor.washer_status", "finished")
self.set_state("sensor.washer_power", "1")
if random.random() < 0.15:
eid = "binary_sensor.hallway_motion"
self.set_state(eid, "off" if self.states[eid]["state"] == "on" else "on")
# -- websocket ----------------------------------------------------------
async def serve(self, ws):
subs = set()
entry = (ws, frozenset())
await ws.send_json({"type": "ready", "ha_version": "mock"})
try:
async for msg in ws:
if msg.type != WSMsgType.TEXT:
continue
req = json.loads(msg.data)
mid, typ = req.get("id"), req.get("type")
if typ == "get_states":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": list(self.states.values())})
elif typ == "subscribe_events":
subs.add(mid)
self.clients.discard(entry)
entry = (ws, frozenset(subs))
self.clients.add(entry)
await ws.send_json({"id": mid, "type": "result", "success": True, "result": None})
elif typ == "call_service" and req.get("domain") == "weather" and req.get("service") == "get_forecasts":
ids = (req.get("target") or {}).get("entity_id") or []
ids = [ids] if isinstance(ids, str) else ids
day = datetime.now().astimezone().replace(hour=12, minute=0, second=0, microsecond=0)
forecast = [{"datetime": (day + timedelta(days=i)).isoformat(), "condition": c, "temperature": hi, "templow": lo}
for i, (c, hi, lo) in enumerate([("partlycloudy", 15, 8), ("rainy", 12, 7), ("sunny", 16, 6)])]
await ws.send_json({"id": mid, "type": "result", "success": True, "result": {"context": {"id": "mock"}, "response": {e: {"forecast": forecast} for e in ids}}})
elif typ == "call_service" and req.get("domain") == "calendar" and req.get("service") == "get_events":
ids = (req.get("target") or {}).get("entity_id") or []
ids = [ids] if isinstance(ids, str) else ids
await ws.send_json({"id": mid, "type": "result", "success": True, "result": {"context": {"id": "mock"}, "response": self.calendar_events(ids)}})
elif typ == "call_service":
data = dict(req.get("service_data") or {})
if req.get("target"):
data.setdefault("entity_id", req["target"].get("entity_id"))
self.call_service(req["domain"], req["service"], data)
await ws.send_json({"id": mid, "type": "result", "success": True, "result": {"context": {"id": "mock"}}})
elif typ == "history/history_during_period":
start = req.get("start_time") or (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat()
await ws.send_json({"id": mid, "type": "result", "success": True, "result": self.history(req.get("entity_ids", []), start)})
elif typ == "config/area_registry/list":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": [
{"area_id": "living_room", "name": "Living Room", "icon": None, "temperature_entity_id": "sensor.living_room_temperature", "humidity_entity_id": "sensor.living_room_humidity"},
{"area_id": "kitchen", "name": "Kitchen", "icon": None},
{"area_id": "bedroom", "name": "Bedroom", "icon": None},
{"area_id": "office", "name": "Office", "icon": None},
{"area_id": "garage", "name": "Garage", "icon": None},
]})
elif typ == "config/entity_registry/list":
areas = {
"living_room": ["light.living_room", "light.reading_lamp", "light.downstairs", "media_player.living_room_speaker", "media_player.tv", "sensor.living_room_temperature", "sensor.living_room_humidity", "binary_sensor.hallway_motion", "climate.thermostat"],
"kitchen": ["light.kitchen", "switch.coffee_machine", "media_player.kitchen_speaker", "sensor.washer_status"],
"bedroom": ["light.bedroom", "binary_sensor.front_door"],
"office": ["fan.office", "input_boolean.guest_mode"],
"garage": ["cover.garage_door", "lock.front_door"],
}
rows = [{"entity_id": e, "area_id": a, "device_id": None, "disabled_by": None, "hidden_by": None, "entity_category": None} for a, es in areas.items() for e in es]
await ws.send_json({"id": mid, "type": "result", "success": True, "result": rows})
elif typ == "config/device_registry/list":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": []})
elif typ == "ping":
await ws.send_json({"id": mid, "type": "pong"})
else:
await ws.send_json({"id": mid, "type": "result", "success": False, "error": {"code": "unknown_command", "message": typ}})
finally:
self.clients.discard(entry)
Reference in New Issue View Git Blame Copy Permalink