mirror of
https://github.com/Matysh/houseplan-card
synced 2026-07-31 16:38:31 +00:00
Scripted demo-stand vacuum: the checklist needs a robot no hardware can provide
The public stand cannot run any Tier-A map integration (no radios, no robots), which left the whole vacuum section of docs/TESTING.md untestable outside the owner's home. demo_robot fakes exactly the surface the card consumes: a Tasshack-shaped map sensor (dict vacuum_position, rooms named after the plan rooms so auto-calibration has something to match, flipped Y so the mirror default is meaningful) and a vacuum that drives a serpentine route through every room in ~3.5 minutes and docks itself.
This commit is contained in:
@@ -0,0 +1,13 @@
|
||||
# Demo-stand extras
|
||||
|
||||
Things that live on the public stand (demo.houseplan.tech) but are not part
|
||||
of the shipped integration.
|
||||
|
||||
- `demo_robot/` — the scripted robot vacuum (docs/VACUUM.md, "demo stand gets
|
||||
a scripted synthetic robot"). Deployed by copying the folder to
|
||||
`custom_components/demo_robot` in both stand seeds and adding `demo_robot:`
|
||||
to configuration.yaml; the dev stand picks it up automatically from this
|
||||
path (`/opt/hp/bin/hp-update-dev.sh`). The rest of the stand-only config
|
||||
(template LQI sensors, alarm helpers, the smoke automation) lives in the
|
||||
seeds on the stand host — see `docs/TESTING-DEMO.md` and the memory note
|
||||
`houseplan-demo-stand`.
|
||||
@@ -0,0 +1,181 @@
|
||||
"""Scripted robot vacuum for the public demo stand.
|
||||
|
||||
The real feature (docs/VACUUM.md) targets Tier-A map integrations — Xiaomi
|
||||
Cloud Map Extractor, Tasshack dreame-vacuum, Valetudo. None of them can run
|
||||
on the stand (no hardware), so the stand ships this simulator instead: a
|
||||
`vacuum` entity with start/pause/return_to_base and a `sensor` whose
|
||||
attributes follow the Tasshack shape (`vacuum_position` as a dict, `rooms`
|
||||
with names matching the plan rooms, `map_index`). Room names match the demo
|
||||
house on purpose — that is what makes «Настроить автоматически» demonstrable.
|
||||
|
||||
Coordinates are millimetres in 0..8000 with the Y axis flipped versus the
|
||||
screen (like every real robot map seen so far) so the mirror toggle in the
|
||||
fit panel has meaning on the stand too.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from datetime import timedelta
|
||||
from typing import Any, Callable
|
||||
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import discovery
|
||||
from homeassistant.helpers.event import async_track_time_interval
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
DOMAIN = "demo_robot"
|
||||
TICK_S = 2 # telemetry cadence
|
||||
SPEED_MM = 600 # per tick ≈ 0.3 m/s, a realistic robot pace
|
||||
DOCK = (3900.0, 2600.0)
|
||||
|
||||
# Robot-map rooms, mm, Y up (flipped vs screen). Bboxes mirror the plan's
|
||||
# room polygons; the names MUST equal the plan room names — auto-calibration
|
||||
# matches by name.
|
||||
ROOMS: dict[str, tuple[str, float, float, float, float]] = {
|
||||
"1": ("Гостиная", 480, 4320, 4160, 7360),
|
||||
"2": ("Кухня", 4160, 5120, 7520, 7360),
|
||||
"3": ("Спальня", 4160, 2400, 7520, 5120),
|
||||
"4": ("Кабинет", 480, 2400, 2720, 4320),
|
||||
"5": ("Прихожая", 2720, 2400, 4160, 4320),
|
||||
}
|
||||
# cleaning order: start at the dock in the hallway, end next to it
|
||||
CLEAN_ORDER = ("5", "4", "1", "2", "3")
|
||||
INSET = 350.0 # keep lanes away from walls
|
||||
LANE_MM = 650.0 # serpentine lane spacing
|
||||
|
||||
|
||||
def _serpentine(x0: float, y0: float, x1: float, y1: float) -> list[tuple[float, float]]:
|
||||
"""Boustrophedon lanes across a room bbox, like a real vacuum drives."""
|
||||
x0, y0, x1, y1 = x0 + INSET, y0 + INSET, x1 - INSET, y1 - INSET
|
||||
pts: list[tuple[float, float]] = []
|
||||
y = y0
|
||||
left_to_right = True
|
||||
while y <= y1 + 1:
|
||||
xs = (x0, x1) if left_to_right else (x1, x0)
|
||||
pts.append((xs[0], y))
|
||||
pts.append((xs[1], y))
|
||||
left_to_right = not left_to_right
|
||||
y += LANE_MM
|
||||
return pts
|
||||
|
||||
|
||||
def build_route() -> list[tuple[float, float]]:
|
||||
route: list[tuple[float, float]] = [DOCK]
|
||||
for rid in CLEAN_ORDER:
|
||||
_, x0, y0, x1, y1 = ROOMS[rid]
|
||||
route.extend(_serpentine(x0, y0, x1, y1))
|
||||
route.append(DOCK)
|
||||
return route
|
||||
|
||||
|
||||
class RobotSim:
|
||||
"""One shared simulation both entities read from."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant) -> None:
|
||||
self.hass = hass
|
||||
self.route = build_route()
|
||||
self.state = "docked" # docked|cleaning|paused|returning
|
||||
self.pos: tuple[float, float] = DOCK
|
||||
self.heading = 0.0
|
||||
self._seg = 0 # index of the segment being driven
|
||||
self._seg_done = 0.0 # mm driven inside the current segment
|
||||
self._back: list[tuple[float, float]] | None = None # manual return path
|
||||
self._listeners: list[Callable[[], None]] = []
|
||||
|
||||
# ---- entity plumbing ----
|
||||
def add_listener(self, cb: Callable[[], None]) -> Callable[[], None]:
|
||||
self._listeners.append(cb)
|
||||
|
||||
def _remove() -> None:
|
||||
if cb in self._listeners:
|
||||
self._listeners.remove(cb)
|
||||
return _remove
|
||||
|
||||
def _notify(self) -> None:
|
||||
for cb in list(self._listeners):
|
||||
cb()
|
||||
|
||||
# ---- commands (called by the vacuum entity) ----
|
||||
def start(self) -> None:
|
||||
if self.state == "paused":
|
||||
self.state = "returning" if self._back else "cleaning"
|
||||
else:
|
||||
self.state = "cleaning"
|
||||
self._seg = 0
|
||||
self._seg_done = 0.0
|
||||
self._back = None
|
||||
self.pos = DOCK
|
||||
self._notify()
|
||||
|
||||
def pause(self) -> None:
|
||||
if self.state in ("cleaning", "returning"):
|
||||
self.state = "paused"
|
||||
self._notify()
|
||||
|
||||
def stop(self) -> None:
|
||||
self.pause()
|
||||
|
||||
def return_to_base(self) -> None:
|
||||
if self.state == "docked":
|
||||
return
|
||||
# retrace the visited waypoints backwards — a straight line to the
|
||||
# dock would cut through walls
|
||||
visited = self.route[: self._seg + 1]
|
||||
self._back = [self.pos] + list(reversed(visited))
|
||||
self._seg = 0
|
||||
self._seg_done = 0.0
|
||||
self.state = "returning"
|
||||
self._notify()
|
||||
|
||||
# ---- physics ----
|
||||
def _polyline(self) -> list[tuple[float, float]]:
|
||||
return self._back if self._back is not None else self.route
|
||||
|
||||
def tick(self, _now: Any = None) -> None:
|
||||
if self.state not in ("cleaning", "returning"):
|
||||
return
|
||||
line = self._polyline()
|
||||
# manual return drives twice as fast, like the real thing hurrying home
|
||||
budget = SPEED_MM * (2 if self._back is not None else 1)
|
||||
while budget > 0 and self._seg < len(line) - 1:
|
||||
ax, ay = line[self._seg]
|
||||
bx, by = line[self._seg + 1]
|
||||
seg_len = math.hypot(bx - ax, by - ay)
|
||||
left = seg_len - self._seg_done
|
||||
if left <= budget:
|
||||
budget -= left
|
||||
self._seg += 1
|
||||
self._seg_done = 0.0
|
||||
continue
|
||||
self._seg_done += budget
|
||||
t = self._seg_done / seg_len
|
||||
self.pos = (ax + (bx - ax) * t, ay + (by - ay) * t)
|
||||
self.heading = math.degrees(math.atan2(by - ay, bx - ax)) % 360
|
||||
budget = 0
|
||||
if self._seg >= len(line) - 1:
|
||||
# arrived: either the route is finished or the manual return is
|
||||
self.pos = line[-1]
|
||||
self.state = "docked"
|
||||
self._back = None
|
||||
elif self._back is None and self._seg >= len(line) - 2:
|
||||
# the last route leg is the drive home
|
||||
self.state = "returning"
|
||||
self._notify()
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
sim = RobotSim(hass)
|
||||
hass.data[DOMAIN] = sim
|
||||
|
||||
@callback
|
||||
def _tick(now: Any) -> None:
|
||||
sim.tick(now)
|
||||
|
||||
async_track_time_interval(hass, _tick, timedelta(seconds=TICK_S))
|
||||
hass.async_create_task(
|
||||
discovery.async_load_platform(hass, "vacuum", DOMAIN, {}, config)
|
||||
)
|
||||
hass.async_create_task(
|
||||
discovery.async_load_platform(hass, "sensor", DOMAIN, {}, config)
|
||||
)
|
||||
return True
|
||||
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"domain": "demo_robot",
|
||||
"name": "Demo Robot Vacuum",
|
||||
"codeowners": [],
|
||||
"dependencies": [],
|
||||
"documentation": "https://github.com/Matysh/houseplan-card/tree/dev/demo/stand/demo_robot",
|
||||
"iot_class": "calculated",
|
||||
"requirements": [],
|
||||
"version": "1.0.0"
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
"""The map sensor of the stand simulator — Tasshack-shaped attributes.
|
||||
|
||||
`vacuum_position` is a plain dict on purpose: the real Tasshack integration
|
||||
keeps a Point OBJECT in memory (houseplan's trail recorder handles both), but
|
||||
a dict is what any template/attribute tooling on the stand can also read.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.sensor import SensorEntity
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import DOMAIN, ROOMS, RobotSim
|
||||
|
||||
|
||||
async def async_setup_platform(hass: HomeAssistant, config, add_entities, discovery_info=None) -> None:
|
||||
add_entities([DemoRobotMap(hass.data[DOMAIN])])
|
||||
|
||||
|
||||
class DemoRobotMap(SensorEntity):
|
||||
_attr_should_poll = False
|
||||
_attr_name = "Карта робота-пылесоса"
|
||||
_attr_unique_id = "demo_robot_map"
|
||||
_attr_icon = "mdi:map"
|
||||
|
||||
def __init__(self, sim: RobotSim) -> None:
|
||||
self._sim = sim
|
||||
self.entity_id = "sensor.demo_robot_map"
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
self.async_on_remove(self._sim.add_listener(self.async_write_ha_state))
|
||||
|
||||
@property
|
||||
def native_value(self) -> str:
|
||||
return self._sim.state
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
x, y = self._sim.pos
|
||||
return {
|
||||
"vacuum_position": {
|
||||
"x": round(x, 1),
|
||||
"y": round(y, 1),
|
||||
"a": round(self._sim.heading, 1),
|
||||
},
|
||||
"rooms": {
|
||||
rid: {
|
||||
"name": name,
|
||||
"x0": x0, "y0": y0, "x1": x1, "y1": y1,
|
||||
"x": (x0 + x1) / 2, "y": (y0 + y1) / 2,
|
||||
}
|
||||
for rid, (name, x0, y0, x1, y1) in ROOMS.items()
|
||||
},
|
||||
"map_index": 1,
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
"""The vacuum entity of the stand simulator."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.vacuum import (
|
||||
StateVacuumEntity,
|
||||
VacuumActivity,
|
||||
VacuumEntityFeature,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import DOMAIN, RobotSim
|
||||
|
||||
ACTIVITY = {
|
||||
"docked": VacuumActivity.DOCKED,
|
||||
"cleaning": VacuumActivity.CLEANING,
|
||||
"paused": VacuumActivity.PAUSED,
|
||||
"returning": VacuumActivity.RETURNING,
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_platform(hass: HomeAssistant, config, add_entities, discovery_info=None) -> None:
|
||||
add_entities([DemoRobotVacuum(hass.data[DOMAIN])])
|
||||
|
||||
|
||||
class DemoRobotVacuum(StateVacuumEntity):
|
||||
_attr_should_poll = False
|
||||
_attr_name = "Робот-пылесос"
|
||||
_attr_unique_id = "demo_robot_vacuum"
|
||||
_attr_supported_features = (
|
||||
VacuumEntityFeature.START
|
||||
| VacuumEntityFeature.STOP
|
||||
| VacuumEntityFeature.PAUSE
|
||||
| VacuumEntityFeature.RETURN_HOME
|
||||
| VacuumEntityFeature.STATE
|
||||
)
|
||||
|
||||
def __init__(self, sim: RobotSim) -> None:
|
||||
self._sim = sim
|
||||
self.entity_id = "vacuum.demo_robot"
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
self.async_on_remove(self._sim.add_listener(self.async_write_ha_state))
|
||||
|
||||
@property
|
||||
def activity(self) -> VacuumActivity:
|
||||
return ACTIVITY.get(self._sim.state, VacuumActivity.IDLE)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
# the card resolves the active map name through the vacuum entity
|
||||
return {"selected_map": "1"}
|
||||
|
||||
async def async_start(self) -> None:
|
||||
self._sim.start()
|
||||
|
||||
async def async_pause(self) -> None:
|
||||
self._sim.pause()
|
||||
|
||||
async def async_stop(self, **kwargs: Any) -> None:
|
||||
self._sim.stop()
|
||||
|
||||
async def async_return_to_base(self, **kwargs: Any) -> None:
|
||||
self._sim.return_to_base()
|
||||
Reference in New Issue
Block a user