veenm
4187c8f38d
- App store icon (icon.png) and favicon, drawn by tools/make_icon.py - Clock can show a weather entity with today's high and low from weather.get_forecasts - Side menu gets an icon-only rail style (now the default): full height, centered icons, active marker, names as hover/focus tooltips; phones keep a labelled bottom tab bar Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
284 lines
18 KiB
Python
284 lines
18 KiB
Python
"""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(entity_id, state, **attrs):
|
||
ts = now_iso()
|
||
return {
|
||
"entity_id": entity_id,
|
||
"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("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"),
|
||
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
|
||
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 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]
|
||
for eid in targets:
|
||
if eid not in self.states:
|
||
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 == "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)
|