"""A tiny fake of the Home Assistant websocket API for local development."""
import asyncio
import json
import math
import random
import time
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=["color_temp", "rgb"], color_mode="color_temp",
color_temp_kelvin=2700, rgb_color=[255, 167, 87], min_color_temp_kelvin=2000, max_color_temp_kelvin=6500),
entity("light.reading_lamp", "on", friendly_name="Reading Lamp", brightness=140, rgb_color=[255, 140, 60], supported_color_modes=["rgb"], color_mode="rgb"),
entity("light.kitchen", "off", friendly_name="Kitchen", brightness=None, supported_color_modes=["color_temp"], min_color_temp_kelvin=2200, max_color_temp_kelvin=6500),
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=["color_temp", "rgb"],
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("sensor.robot_vacuum_main_brush_left", "41", friendly_name="Robot Vacuum Main brush left", unit_of_measurement="%"),
entity("sensor.robot_vacuum_side_brush_left", "9", friendly_name="Robot Vacuum Side brush 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.doorbell", "off", friendly_name="Doorbell", device_class="occupancy"),
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"],
preset_modes=["none", "eco", "comfort", "away"], preset_mode="comfort", fan_modes=["auto", "low", "high"], fan_mode="auto", hvac_action="heating", current_humidity=46, target_temp_step=0.5),
entity("sensor.grid_power", "-680", friendly_name="Grid Power", unit_of_measurement="W", device_class="power"),
entity("sensor.home_battery_power", "500", friendly_name="Home Battery Power", unit_of_measurement="W", device_class="power"),
entity("sensor.home_battery_level", "64", friendly_name="Home Battery Level", unit_of_measurement="%", device_class="battery"),
entity("sensor.solar_energy_today", "12.4", friendly_name="Produced today", unit_of_measurement="kWh", device_class="energy"),
entity("sensor.home_energy_today", "8.1", friendly_name="Used today", unit_of_measurement="kWh", device_class="energy"),
entity("binary_sensor.washer_door", "off", friendly_name="Washer Door", device_class="door"),
# A dishwasher as the Home Connect integration exposes it
entity("sensor.dishwasher_operation_state", "run", friendly_name="Dishwasher Operation state", device_class="enum",
options=["inactive", "ready", "delayedstart", "run", "pause", "actionrequired", "finished", "error", "aborting"]),
entity("sensor.dishwasher_program_progress", "40", friendly_name="Dishwasher Program progress", unit_of_measurement="%"),
entity("sensor.dishwasher_program_finish_time", (datetime.now(timezone.utc) + timedelta(minutes=52)).isoformat(), friendly_name="Dishwasher Program finish time", device_class="timestamp"),
entity("sensor.dishwasher_door", "locked", friendly_name="Dishwasher Door", device_class="enum", options=["closed", "locked", "open"]),
entity("select.dishwasher_active_program", "dishcare_dishwasher_program_eco_50", friendly_name="Dishwasher Active program",
options=["dishcare_dishwasher_program_eco_50", "dishcare_dishwasher_program_auto_2", "dishcare_dishwasher_program_quick_45"]),
entity("button.dishwasher_resume_program", "unknown", friendly_name="Dishwasher Resume program"),
entity("button.dishwasher_stop_program", "unknown", friendly_name="Dishwasher Stop program"),
entity("switch.dishwasher_power", "on", friendly_name="Dishwasher Power"),
entity("sensor.dishwasher_rinse_aid_nearly_empty", "present", friendly_name="Dishwasher Rinse aid nearly empty", device_class="enum"),
entity("sensor.washer_energy", "0.4", friendly_name="Washer Energy", unit_of_measurement="kWh", device_class="energy"),
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,"),
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_picture="data:image/svg+xml;utf8,"),
entity("person.noor", "Work", friendly_name="Noor"),
entity("todo.shopping_list", "3", friendly_name="Shopping list"),
entity("cover.living_room_blinds", "open", friendly_name="Living Room Blinds", current_position=60, current_tilt_position=30),
entity("cover.bedroom_curtain", "closed", friendly_name="Bedroom Curtain", current_position=0),
entity("alarm_control_panel.home", "disarmed", friendly_name="Home Alarm", code_format="number", supported_features=7),
entity("update.home_assistant_core_update", "on", friendly_name="Home Assistant Core Update", title="Home Assistant Core",
installed_version="2026.10.1", latest_version="2026.10.3", release_url="https://www.home-assistant.io/blog/"),
entity("timer.kitchen", "idle", friendly_name="Kitchen timer", duration="0:10:00", remaining="0:10:00"),
entity("sun.sun", "above_horizon", friendly_name="Sun",
next_rising=(datetime.now(timezone.utc) + timedelta(hours=9)).isoformat(), next_setting=(datetime.now(timezone.utc) + timedelta(hours=3)).isoformat()),
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("sensor.model_y_time_to_full", "48", friendly_name="Model Y Time to full", unit_of_measurement="min", device_class="duration"),
entity("switch.model_y_charger", "on", friendly_name="Model Y Charger"),
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.states.update({e["entity_id"]: e for e in self._extra_entities()})
self.states["cover.garage_door"]["context"] = {"id": "mock", "user_id": "mock-user-alex"}
self.clients = set()
self.todo = [{"uid": "1", "summary": "Milk", "status": "needs_action"}, {"uid": "2", "summary": "Bread", "status": "needs_action"},
{"uid": "3", "summary": "Coffee beans", "status": "needs_action"}, {"uid": "4", "summary": "Batteries", "status": "completed"}]
self._timer_handle = None
self.notes = {
"welcome": {"notification_id": "welcome", "title": "New devices found", "message": "Home Assistant found **2 new devices**. [Check them](/config/integrations).", "created_at": now_iso()},
"backup": {"notification_id": "backup", "title": "Backup finished", "message": "Last night's backup completed.", "created_at": now_iso()},
}
self.note_subs = set()
def _extra_entities(self):
"""Parcels, air quality, meters, solar forecast, sports and Formula One."""
now = datetime.now().astimezone()
day = now.replace(hour=0, minute=0, second=0, microsecond=0)
at = lambda d, h, m=0: (day + timedelta(days=d, hours=h, minutes=m)).isoformat()
drivers = [("Max", "Verstappen", "VER", "Red Bull", 395), ("Lando", "Norris", "NOR", "McLaren", 372), ("Oscar", "Piastri", "PIA", "McLaren", 341),
("Charles", "Leclerc", "LEC", "Ferrari", 290), ("George", "Russell", "RUS", "Mercedes", 245), ("Lewis", "Hamilton", "HAM", "Ferrari", 198)]
standings = [{"position": str(i + 1), "points": str(p), "wins": str(max(0, 6 - i * 2)), "Driver": {"givenName": g, "familyName": f, "code": c}, "Constructors": [{"name": team}]}
for i, (g, f, c, team, p) in enumerate(drivers)]
teams = [{"position": str(i + 1), "points": str(p), "wins": str(w), "Constructor": {"name": n}} for i, (n, p, w) in enumerate([("McLaren", 713, 9), ("Ferrari", 488, 3), ("Red Bull", 470, 7), ("Mercedes", 402, 2)])]
results = [{"position": str(i + 1), "points": str([25, 18, 15, 12, 10, 8][i]), "Driver": {"givenName": g, "familyName": f, "code": c}, "Constructor": {"name": team},
"status": "Finished", "Time": {"time": ["1:32:10.553", "+2.411", "+8.902", "+15.300", "+21.774", "+30.105"][i]}}
for i, (g, f, c, team, _) in enumerate(drivers)]
return [
entity("sensor.postnl_delivery", "2", friendly_name="PostNL Delivery",
enroute=[{"key": "3SABCD123", "name": "Bol.com", "status_message": "De bezorger is onderweg", "planned_date": at(0, 14), "planned_to": at(0, 16), "url": "https://jouw.postnl.nl/"},
{"key": "3SXYZ987", "name": "Coolblue", "status_message": "Wordt gesorteerd", "planned_date": at(1, 0)}],
delivered=[{"key": "3SQQQ555", "name": "Zalando", "status_message": "Bezorgd", "delivery_date": at(-1, 11, 20)}]),
entity("sensor.living_room_co2", "940", friendly_name="Living Room CO2", unit_of_measurement="ppm", device_class="carbon_dioxide"),
entity("sensor.living_room_pm25", "8", friendly_name="Living Room PM2.5", unit_of_measurement="µg/m³", device_class="pm25"),
entity("sensor.bedroom_co2", "1340", friendly_name="Bedroom CO2", unit_of_measurement="ppm", device_class="carbon_dioxide"),
entity("sensor.bedroom_humidity", "72", friendly_name="Bedroom Humidity", unit_of_measurement="%", device_class="humidity"),
entity("sensor.office_co2", "610", friendly_name="Office CO2", unit_of_measurement="ppm", device_class="carbon_dioxide"),
entity("sensor.office_humidity", "46", friendly_name="Office Humidity", unit_of_measurement="%", device_class="humidity"),
entity("sensor.water_meter", "1234.567", friendly_name="Water Meter", unit_of_measurement="m³", device_class="water", state_class="total_increasing"),
entity("sensor.gas_meter", "4321.12", friendly_name="Gas Meter", unit_of_measurement="m³", device_class="gas", state_class="total_increasing"),
entity("sensor.energy_production_today", "18.4", friendly_name="Estimated energy production - today", unit_of_measurement="kWh", device_class="energy"),
entity("sensor.energy_production_tomorrow", "12.1", friendly_name="Estimated energy production - tomorrow", unit_of_measurement="kWh", device_class="energy"),
entity("sensor.power_production_now", "1240", friendly_name="Estimated power production - now", unit_of_measurement="W", device_class="power"),
entity("sensor.team_tracker_ajax", "PRE", friendly_name="Ajax", sport="soccer", league="Eredivisie", team_name="Ajax", team_abbr="AJA", team_homeaway="home",
opponent_name="PSV", opponent_abbr="PSV", date=at(0, 20), kickoff_in="in a few hours", venue="Johan Cruijff ArenA", location="Amsterdam", tv_network="ESPN",
team_colors=["#d2122e", "#ffffff"], opponent_colors=["#ed1c24", "#ffffff"], team_record="8-2-1", opponent_record="9-1-1", team_rank=2, opponent_rank=1,
team_win_probability=0.42, opponent_win_probability=0.33),
entity("sensor.team_tracker_feyenoord", "IN", friendly_name="Feyenoord", sport="soccer", league="Eredivisie", team_name="Feyenoord", team_abbr="FEY", team_homeaway="away",
opponent_name="FC Utrecht", opponent_abbr="UTR", date=at(0, 14, 30), venue="Stadion Galgenwaard", location="Utrecht", tv_network="ESPN 2",
team_colors=["#e2001a", "#000000"], opponent_colors=["#d40000", "#ffffff"], team_score=2, opponent_score=1, clock="63'", team_record="7-3-1", opponent_record="5-3-3",
team_rank=3, opponent_rank=6, last_play="Goal! Hadj Moussa scores for Feyenoord.", team_win_probability=0.71, opponent_win_probability=0.09),
entity("sensor.f1_next_race", at(5, 18), friendly_name="F1 Next Race", season="2026", round="21", race_name="São Paulo Grand Prix",
circuit_name="Autódromo José Carlos Pace", circuit_locality="São Paulo", circuit_country="Brazil",
first_practice_start=at(3, 15, 30), sprint_qualifying_start=at(3, 19, 30), sprint_start=at(4, 15), qualifying_start=at(4, 19), race_start=at(5, 18)),
entity("sensor.f1_driver_standings", "Max Verstappen", friendly_name="F1 Driver Standings", driver_standings=standings),
entity("sensor.f1_constructor_standings", "McLaren", friendly_name="F1 Constructor Standings", constructor_standings=teams),
entity("sensor.f1_last_race_results", "Mexico City Grand Prix", friendly_name="F1 Last Race Results", race_name="Mexico City Grand Prix", results=results),
]
def statistics(self, ids, start):
start_dt = datetime.fromisoformat(start.replace("Z", "+00:00")).astimezone()
now = datetime.now().astimezone()
out = {}
for eid in ids:
base = {"sensor.gas_meter": 3.2}.get(eid, 0.18)
rows, d = [], start_dt
while d <= now:
seed = (d.toordinal() * 7 + len(eid)) % 10
change = base * (0.6 + seed / 12)
if d.date() == now.date():
change *= now.hour / 24
rows.append({"start": int(d.timestamp() * 1000), "end": int((d + timedelta(days=1)).timestamp() * 1000), "change": round(change, 3)})
d += timedelta(days=1)
out[eid] = rows
return out
def solar_forecast(self):
day = datetime.now().astimezone().replace(hour=0, minute=0, second=0, microsecond=0)
hours = {}
for d, peak in ((0, 2600), (1, 1700)):
for h in range(6, 21):
wh = max(0, peak * math.sin(math.pi * (h - 6) / 14))
hours[(day + timedelta(days=d, hours=h)).isoformat()] = round(wh)
return {"mock_entry": {"wh_hours": hours}}
def broadcast_event(self, event_type, data):
event = {"event_type": event_type, "data": data or {}, "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}))
LOVELACE = {"views": [
{"title": "Home", "path": "home", "icon": "mdi:home", "cards": [
{"type": "entities", "title": "Living room", "entities": ["light.living_room", "light.reading_lamp", {"entity": "sensor.living_room_temperature", "name": "Temperature"}]},
{"type": "thermostat", "entity": "climate.thermostat"},
{"type": "weather-forecast", "entity": "weather.home"},
{"type": "horizontal-stack", "cards": [{"type": "button", "entity": "switch.coffee_machine"}, {"type": "light", "entity": "light.kitchen"}]},
{"type": "media-control", "entity": "media_player.living_room_speaker"},
{"type": "markdown", "content": "## Welcome home"},
]},
{"title": "Energy", "path": "energy", "cards": [
{"type": "gauge", "entity": "sensor.power_usage", "min": 0, "max": 5000},
{"type": "history-graph", "entities": ["sensor.power_usage", "sensor.solar_production"], "hours_to_show": 24},
{"type": "tile", "entity": "sensor.solar_production"},
{"type": "picture-entity", "entity": "camera.front_yard"},
]},
]}
# -- 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 _timer(self, eid, service, data):
a = self.states[eid]["attributes"]
secs = lambda v: sum(int(x) * m for x, m in zip(reversed(str(v).split(":")), (1, 60, 3600)))
fmt = lambda n: f"{n // 3600}:{n % 3600 // 60:02d}:{n % 60:02d}"
if self._timer_handle:
self._timer_handle.cancel()
if service == "start":
dur = secs(data["duration"]) if data.get("duration") else (secs(a["remaining"]) if self.states[eid]["state"] == "paused" else secs(a["duration"]))
total = secs(data["duration"]) if data.get("duration") else secs(a["duration"])
fin = datetime.now(timezone.utc) + timedelta(seconds=dur)
self.set_state(eid, "active", duration=fmt(total), remaining=fmt(dur), finishes_at=fin.isoformat())
self._timer_handle = asyncio.get_event_loop().call_later(dur, lambda: self.set_state(eid, "idle", remaining=fmt(total), finishes_at=None))
elif service == "pause":
left = max(0, int((datetime.fromisoformat(a["finishes_at"]) - datetime.now(timezone.utc)).total_seconds())) if a.get("finishes_at") else secs(a["remaining"])
self.set_state(eid, "paused", remaining=fmt(left), finishes_at=None)
elif service in ("cancel", "finish"):
self.set_state(eid, "idle", remaining=a["duration"], finishes_at=None)
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 == "todo":
item = data.get("item")
if service == "add_item":
self.todo.insert(0, {"uid": str(int(time.time() * 1000)), "summary": item, "status": "needs_action"})
for it in list(self.todo):
if it["uid"] == item or it["summary"] == item:
if service == "update_item":
it["status"] = data.get("status", it["status"])
elif service == "remove_item":
self.todo.remove(it)
self.set_state(eid, str(sum(1 for it in self.todo if it["status"] == "needs_action")))
continue
if domain == "cover":
pos = self.states[eid]["attributes"].get("current_position")
if service == "set_cover_position":
p_ = int(data.get("position", 0)); self.set_state(eid, "open" if p_ > 0 else "closed", current_position=p_)
elif service == "set_cover_tilt_position":
self.set_state(eid, None, current_tilt_position=int(data.get("tilt_position", 0)))
elif service in ("open_cover", "close_cover"):
self.set_state(eid, "open" if service == "open_cover" else "closed", current_position=100 if service == "open_cover" else 0)
continue
if domain == "alarm_control_panel":
if service != "alarm_disarm" and data.get("code") not in (None, "1234") or service == "alarm_disarm" and data.get("code") != "1234":
raise ValueError("Invalid code")
self.set_state(eid, {"alarm_arm_home": "armed_home", "alarm_arm_away": "armed_away", "alarm_arm_night": "armed_night", "alarm_disarm": "disarmed"}[service])
continue
if domain == "timer":
self._timer(eid, service, data)
continue
if eid.startswith("button.dishwasher_"): # Home Connect program buttons
st_ = {"stop": "aborting", "pause": "pause", "resume": "run"}[eid.split("_")[-2]]
self.set_state("sensor.dishwasher_operation_state", st_)
if st_ == "aborting":
asyncio.get_event_loop().call_later(3, lambda: self.set_state("sensor.dishwasher_operation_state", "ready"))
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 "rgb_color" in data:
extra.update(rgb_color=data["rgb_color"], color_mode="rgb")
if "color_temp_kelvin" in data:
extra.update(color_temp_kelvin=data["color_temp_kelvin"], color_mode="color_temp")
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 in ("set_preset_mode", "set_fan_mode"):
key = "preset_mode" if service == "set_preset_mode" else "fan_mode"
self.set_state(eid, None, **{key: data.get(key)})
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"""""".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}")
home, solar, batt = float(self.states["sensor.power_usage"]["state"]), float(self.states["sensor.solar_production"]["state"]), float(self.states["sensor.home_battery_power"]["state"])
self.set_state("sensor.grid_power", f"{home + batt - solar:.0f}")
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 ----------------------------------------------------------
# Home Assistant's logins, as config/auth/list returns them. Alex is the owner; the others aren't administrators.
USERS = {
"mock-user-alex": {"id": "mock-user-alex", "name": "Alex", "username": "alex", "is_owner": True, "is_active": True, "system_generated": False, "group_ids": ["system-admin"]},
"mock-user-sam": {"id": "mock-user-sam", "name": "Sam", "username": "sam", "is_owner": False, "is_active": True, "system_generated": False, "group_ids": ["system-users"]},
"mock-user-noor": {"id": "mock-user-noor", "name": "Noor", "username": "noor", "is_owner": False, "is_active": True, "system_generated": False, "group_ids": ["system-users"]},
"mock-user-tablet": {"id": "mock-user-tablet", "name": "Kitchen tablet", "username": "tablet", "is_owner": False, "is_active": True, "system_generated": False, "group_ids": ["system-users"]},
}
@staticmethod
def USERS_LIST(): # noqa: N802 - mirrors USERS
return [*MockHA.USERS.values(), {"id": "mock-supervisor", "name": "Supervisor", "is_owner": False, "is_active": True, "system_generated": True, "group_ids": ["system-admin"]}]
async def serve(self, ws, allow=None):
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 allow and not allow(req):
await ws.send_json({"id": mid, "type": "result", "success": False, "error": {"code": "unauthorized", "message": "Only Home Assistant administrators can do this"}})
elif 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 == "persistent_notification/subscribe":
self.note_subs.add((ws, mid))
await ws.send_json({"id": mid, "type": "result", "success": True, "result": None})
await ws.send_json({"id": mid, "type": "event", "event": {"type": "current", "notifications": self.notes}})
elif typ == "call_service" and req.get("domain") == "persistent_notification":
gone = dict(self.notes) if req["service"] == "dismiss_all" else {k: v for k, v in self.notes.items() if k == (req.get("service_data") or {}).get("notification_id")}
for k in gone:
self.notes.pop(k, None)
for w2, sid in list(self.note_subs):
asyncio.ensure_future(self._safe_send(w2, {"id": sid, "type": "event", "event": {"type": "removed", "notifications": gone}}))
await ws.send_json({"id": mid, "type": "result", "success": True, "result": {"context": {"id": "mock"}}})
elif typ == "todo/item/list":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": {"items": self.todo}})
elif typ == "call_service" and req.get("domain") == "weather" and req.get("service") == "get_forecasts" and (req.get("service_data") or {}).get("type") == "hourly":
ids = (req.get("target") or {}).get("entity_id") or []
ids = [ids] if isinstance(ids, str) else ids
hour = datetime.now().astimezone().replace(minute=0, second=0, microsecond=0)
conds = ["partlycloudy", "partlycloudy", "cloudy", "rainy", "rainy", "cloudy", "partlycloudy", "sunny"]
forecast = [{"datetime": (hour + timedelta(hours=i + 1)).isoformat(), "condition": conds[i % 8], "temperature": round(13 + 3 * math.sin(i / 4), 1),
"precipitation_probability": [0, 10, 30, 70, 80, 40, 10, 0][i % 8]} for i in range(24)]
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") == "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), ("cloudy", 14, 9), ("pouring", 11, 8), ("partlycloudy", 13, 6), ("sunny", 17, 7)])]
for i, f in enumerate(forecast):
f["precipitation_probability"] = [10, 80, 0, 20, 90, 30, 0][i]
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"))
try:
self.call_service(req["domain"], req["service"], data)
except ValueError as err:
await ws.send_json({"id": mid, "type": "result", "success": False, "error": {"code": "service_validation_error", "message": str(err)}})
continue
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": ["sensor.living_room_co2", "sensor.living_room_pm25", "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", "sensor.bedroom_co2", "sensor.bedroom_humidity"],
"office": ["fan.office", "input_boolean.guest_mode", "sensor.office_co2", "sensor.office_humidity"],
"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]
rows += [{"entity_id": e, "area_id": "kitchen", "device_id": "hc-dishwasher", "disabled_by": None, "hidden_by": None, "entity_category": None}
for e in self.states if e.split(".")[1].startswith("dishwasher_")]
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 == "recorder/statistics_during_period":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": self.statistics(req.get("statistic_ids") or [], req.get("start_time"))})
elif typ == "energy/solar_forecast":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": self.solar_forecast()})
elif typ == "fire_event":
self.broadcast_event(req.get("event_type"), req.get("event_data"))
await ws.send_json({"id": mid, "type": "result", "success": True, "result": {"context": {"id": "mock"}}})
elif typ == "lovelace/dashboards/list":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": [{"url_path": "wall-tablet", "title": "Wall tablet", "mode": "storage", "icon": "mdi:tablet"}]})
elif typ == "lovelace/config":
await ws.send_json({"id": mid, "type": "result", "success": True, "result": self.LOVELACE})
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)
self.note_subs = {x for x in self.note_subs if x[0] is not ws}