diff --git a/src/haclient/domains/__init__.py b/src/haclient/domains/__init__.py index 174589d..ba3d1c6 100644 --- a/src/haclient/domains/__init__.py +++ b/src/haclient/domains/__init__.py @@ -17,6 +17,7 @@ from haclient.domains.sensor import Sensor from haclient.domains.switch import Switch from haclient.domains.timer import Timer +from haclient.domains.vacuum import Vacuum from haclient.domains.valve import Valve __all__ = [ @@ -34,5 +35,6 @@ "Sensor", "Switch", "Timer", + "Vacuum", "Valve", ] diff --git a/src/haclient/domains/vacuum.py b/src/haclient/domains/vacuum.py new file mode 100644 index 0000000..6f12389 --- /dev/null +++ b/src/haclient/domains/vacuum.py @@ -0,0 +1,405 @@ +"""``vacuum`` domain implementation.""" + +from __future__ import annotations + +import logging +from typing import Any + +from haclient.core.plugins import DomainSpec, register_domain +from haclient.entity.base import Entity + +_LOGGER = logging.getLogger(__name__) + +# Home Assistant ``VacuumEntityFeature`` bitmask. +# See homeassistant/components/vacuum/const.py. +_FEATURE_TURN_ON = 1 +_FEATURE_TURN_OFF = 2 +_FEATURE_PAUSE = 4 +_FEATURE_STOP = 8 +_FEATURE_RETURN_HOME = 16 +_FEATURE_FAN_SPEED = 32 +_FEATURE_BATTERY = 64 +_FEATURE_STATUS = 128 +_FEATURE_SEND_COMMAND = 256 +_FEATURE_LOCATE = 512 +_FEATURE_CLEAN_SPOT = 1024 +_FEATURE_MAP = 2048 +_FEATURE_STATE = 4096 +_FEATURE_START = 8192 + +# Canonical Home Assistant vacuum states. +_STATE_CLEANING = "cleaning" +_STATE_DOCKED = "docked" +_STATE_IDLE = "idle" +_STATE_PAUSED = "paused" +_STATE_RETURNING = "returning" +_STATE_ERROR = "error" + + +class Vacuum(Entity): + """A Home Assistant vacuum entity. + + Provides intent-specific actions (``start``, ``pause``, ``stop``, + ``return_to_base``, ``locate``, ``set_fan_speed``, ``send_command``, + ``clean_spot``) and structured state introspection (``is_cleaning``, + ``is_docked``, ``is_idle``, ``is_paused``, ``is_returning``, + ``is_error``, ``battery_level``, ``fan_speed``, ``fan_speed_list``) + rather than exposing raw service calls. + + Methods that depend on optional vacuum capabilities degrade safely: + if the underlying hardware does not advertise the relevant + ``VacuumEntityFeature`` bit in ``supported_features``, the call + becomes a no-op that logs a debug message instead of raising. This + keeps user code portable across heterogeneous fleets. Callers that + need to know whether an action will actually be dispatched can + pre-check with the ``supports_*`` properties. + """ + + domain = "vacuum" + + # -- Listener decorators ------------------------------------------ + + def on_start(self, func: Any) -> Any: + """Register a listener for when the vacuum starts cleaning. + + Parameters + ---------- + func : callable + Sync or async zero-argument callable invoked on every + transition into the ``cleaning`` state. + + Returns + ------- + callable + The same *func*, returned for decorator use. + """ + return self._register_state_transition_listener(_STATE_CLEANING, func) + + def on_dock(self, func: Any) -> Any: + """Register a listener for when the vacuum returns to the dock. + + Parameters + ---------- + func : callable + Sync or async zero-argument callable invoked on every + transition into the ``docked`` state. + + Returns + ------- + callable + The same *func*, returned for decorator use. + """ + return self._register_state_transition_listener(_STATE_DOCKED, func) + + def on_error(self, func: Any) -> Any: + """Register a listener for when the vacuum enters the error state. + + Parameters + ---------- + func : callable + Sync or async zero-argument callable invoked on every + transition into the ``error`` state. + + Returns + ------- + callable + The same *func*, returned for decorator use. + """ + return self._register_state_transition_listener(_STATE_ERROR, func) + + def on_battery_change(self, func: Any) -> Any: + """Register a listener for battery level changes. + + Parameters + ---------- + func : callable + Callable receiving the new ``battery_level`` value (0--100). + + Returns + ------- + callable + The same *func*, returned for decorator use. + """ + return self._register_attr_listener("battery_level", func) + + def on_fan_speed_change(self, func: Any) -> Any: + """Register a listener for fan-speed changes. + + Parameters + ---------- + func : callable + Callable receiving the new ``fan_speed`` value. + + Returns + ------- + callable + The same *func*, returned for decorator use. + """ + return self._register_attr_listener("fan_speed", func) + + # -- State properties --------------------------------------------- + + @property + def is_cleaning(self) -> bool: + """Whether the vacuum is currently cleaning.""" + return self.state == _STATE_CLEANING + + @property + def is_docked(self) -> bool: + """Whether the vacuum is currently docked.""" + return self.state == _STATE_DOCKED + + @property + def is_idle(self) -> bool: + """Whether the vacuum is currently idle.""" + return self.state == _STATE_IDLE + + @property + def is_paused(self) -> bool: + """Whether the vacuum is currently paused.""" + return self.state == _STATE_PAUSED + + @property + def is_returning(self) -> bool: + """Whether the vacuum is currently returning to the dock.""" + return self.state == _STATE_RETURNING + + @property + def is_error(self) -> bool: + """Whether the vacuum is currently in an error state.""" + return self.state == _STATE_ERROR + + @property + def battery_level(self) -> int | None: + """Battery charge percentage (0--100) or ``None`` if unsupported.""" + value = self.attributes.get("battery_level") + return int(value) if isinstance(value, (int, float)) else None + + @property + def fan_speed(self) -> str | None: + """Current fan speed label, or ``None`` if unsupported.""" + value = self.attributes.get("fan_speed") + return str(value) if isinstance(value, str) else None + + @property + def fan_speed_list(self) -> list[str]: + """Available fan-speed labels, or an empty list if unsupported.""" + value = self.attributes.get("fan_speed_list") + if not isinstance(value, list): + return [] + return [str(item) for item in value if isinstance(item, str)] + + def _has_feature(self, mask: int) -> bool: + """Return ``True`` when ``supported_features`` advertises *mask*. + + Parameters + ---------- + mask : int + ``VacuumEntityFeature`` bit to test for. + + Returns + ------- + bool + ``True`` if the entity advertises *mask* in its + ``supported_features`` bitmask, otherwise ``False``. + """ + features = self.attributes.get("supported_features") + if not isinstance(features, int): + return False + return bool(features & mask) + + @property + def supports_start(self) -> bool: + """Whether the vacuum advertises ``START`` support.""" + return self._has_feature(_FEATURE_START) + + @property + def supports_pause(self) -> bool: + """Whether the vacuum advertises ``PAUSE`` support.""" + return self._has_feature(_FEATURE_PAUSE) + + @property + def supports_stop(self) -> bool: + """Whether the vacuum advertises ``STOP`` support.""" + return self._has_feature(_FEATURE_STOP) + + @property + def supports_return_home(self) -> bool: + """Whether the vacuum advertises ``RETURN_HOME`` support.""" + return self._has_feature(_FEATURE_RETURN_HOME) + + @property + def supports_locate(self) -> bool: + """Whether the vacuum advertises ``LOCATE`` support.""" + return self._has_feature(_FEATURE_LOCATE) + + @property + def supports_fan_speed(self) -> bool: + """Whether the vacuum advertises ``FAN_SPEED`` support.""" + return self._has_feature(_FEATURE_FAN_SPEED) + + @property + def supports_send_command(self) -> bool: + """Whether the vacuum advertises ``SEND_COMMAND`` support.""" + return self._has_feature(_FEATURE_SEND_COMMAND) + + @property + def supports_clean_spot(self) -> bool: + """Whether the vacuum advertises ``CLEAN_SPOT`` support.""" + return self._has_feature(_FEATURE_CLEAN_SPOT) + + # -- Actions ------------------------------------------------------ + + async def start(self) -> None: + """Start (or resume) cleaning. + + Degrades safely: if the vacuum does not advertise the ``START`` + feature, this method logs a debug message and returns without + raising. Callers can pre-check with `supports_start`. + """ + if not self.supports_start: + _LOGGER.debug( + "start() unsupported for %s; skipping (no VacuumEntityFeature.START)", + self.entity_id, + ) + return + await self._call_service("start") + + async def pause(self) -> None: + """Pause the current cleaning cycle. + + Degrades safely: if the vacuum does not advertise the ``PAUSE`` + feature, this method logs a debug message and returns without + raising. Callers can pre-check with `supports_pause`. + """ + if not self.supports_pause: + _LOGGER.debug( + "pause() unsupported for %s; skipping (no VacuumEntityFeature.PAUSE)", + self.entity_id, + ) + return + await self._call_service("pause") + + async def stop(self) -> None: + """Stop the current cleaning cycle. + + Degrades safely: if the vacuum does not advertise the ``STOP`` + feature, this method logs a debug message and returns without + raising. Callers can pre-check with `supports_stop`. + """ + if not self.supports_stop: + _LOGGER.debug( + "stop() unsupported for %s; skipping (no VacuumEntityFeature.STOP)", + self.entity_id, + ) + return + await self._call_service("stop") + + async def return_to_base(self) -> None: + """Send the vacuum back to its dock. + + Degrades safely: if the vacuum does not advertise the + ``RETURN_HOME`` feature, this method logs a debug message and + returns without raising. Callers can pre-check with + `supports_return_home`. + """ + if not self.supports_return_home: + _LOGGER.debug( + "return_to_base() unsupported for %s; skipping " + "(no VacuumEntityFeature.RETURN_HOME)", + self.entity_id, + ) + return + await self._call_service("return_to_base") + + async def locate(self) -> None: + """Make the vacuum emit a locator sound, if supported. + + Degrades safely: if the vacuum does not advertise the ``LOCATE`` + feature, this method logs a debug message and returns without + raising. Callers can pre-check with `supports_locate`. + """ + if not self.supports_locate: + _LOGGER.debug( + "locate() unsupported for %s; skipping (no VacuumEntityFeature.LOCATE)", + self.entity_id, + ) + return + await self._call_service("locate") + + async def clean_spot(self) -> None: + """Perform a spot-cleaning cycle, if supported. + + Degrades safely: if the vacuum does not advertise the + ``CLEAN_SPOT`` feature, this method logs a debug message and + returns without raising. Callers can pre-check with + `supports_clean_spot`. + """ + if not self.supports_clean_spot: + _LOGGER.debug( + "clean_spot() unsupported for %s; skipping (no VacuumEntityFeature.CLEAN_SPOT)", + self.entity_id, + ) + return + await self._call_service("clean_spot") + + async def set_fan_speed(self, speed: str) -> None: + """Set the vacuum's fan speed, if supported. + + Parameters + ---------- + speed : str + Fan-speed label. Should be one of the values reported in + the entity's ``fan_speed_list`` attribute. + + Notes + ----- + Degrades safely: if the vacuum does not advertise the + ``FAN_SPEED`` feature, this method logs a debug message and + returns without raising. Callers can pre-check with + `supports_fan_speed`. + """ + if not self.supports_fan_speed: + _LOGGER.debug( + "set_fan_speed() unsupported for %s; skipping (no VacuumEntityFeature.FAN_SPEED)", + self.entity_id, + ) + return + await self._call_service("set_fan_speed", {"fan_speed": str(speed)}) + + async def send_command( + self, + command: str, + params: dict[str, Any] | None = None, + ) -> None: + """Send a vendor-specific command to the vacuum, if supported. + + Parameters + ---------- + command : str + Vendor-specific command name to send. + params : dict or None, optional + Optional parameters dictionary forwarded verbatim to Home + Assistant alongside the command. + + Notes + ----- + Degrades safely: if the vacuum does not advertise the + ``SEND_COMMAND`` feature, this method logs a debug message and + returns without raising. Callers can pre-check with + `supports_send_command`. + """ + if not self.supports_send_command: + _LOGGER.debug( + "send_command() unsupported for %s; skipping (no VacuumEntityFeature.SEND_COMMAND)", + self.entity_id, + ) + return + data: dict[str, Any] = {"command": str(command)} + if params is not None: + data["params"] = dict(params) + await self._call_service("send_command", data) + + +SPEC: DomainSpec[Vacuum] = register_domain(DomainSpec(name="vacuum", entity_cls=Vacuum)) +"""The `DomainSpec` registered with the shared `DomainRegistry`.""" diff --git a/tests/test_domains.py b/tests/test_domains.py index 79bed88..2f8532d 100644 --- a/tests/test_domains.py +++ b/tests/test_domains.py @@ -1120,3 +1120,198 @@ def _co2(old: Any, new: Any) -> None: assert aqi_events == [("42", "55")] assert pm25_events == [(10.0, 14.0)] assert co2_events == [(800, 950)] + + +# --------------------------------------------------------------------------- +# Vacuum +# --------------------------------------------------------------------------- + + +# All ``VacuumEntityFeature`` bits OR'd together (START|PAUSE|STOP| +# RETURN_HOME|FAN_SPEED|LOCATE|SEND_COMMAND|CLEAN_SPOT). +_VACUUM_FULL_FEATURES = 8192 | 4 | 8 | 16 | 32 | 512 | 256 | 1024 + + +async def test_vacuum_actions_full_featured(client: HAClient, fake_ha: FakeHA) -> None: + """A full-featured vacuum dispatches every intent-specific service.""" + robo = client.vacuum("roborock") + robo._apply_state( + {"state": "cleaning", "attributes": {"supported_features": _VACUUM_FULL_FEATURES}} + ) + + await robo.start() + await robo.pause() + await robo.stop() + await robo.return_to_base() + await robo.locate() + await robo.clean_spot() + await robo.set_fan_speed("turbo") + await robo.send_command("set_zone", {"zone": [1, 2, 3, 4]}) + + services = [c["service"] for c in fake_ha.ws_service_calls] + assert services == [ + "start", + "pause", + "stop", + "return_to_base", + "locate", + "clean_spot", + "set_fan_speed", + "send_command", + ] + assert all(c["domain"] == "vacuum" for c in fake_ha.ws_service_calls) + assert all( + c["service_data"]["entity_id"] == "vacuum.roborock" for c in fake_ha.ws_service_calls + ) + + fan = _find_call(fake_ha, "set_fan_speed") + assert fan["service_data"]["fan_speed"] == "turbo" + + cmd = _find_call(fake_ha, "send_command") + assert cmd["service_data"]["command"] == "set_zone" + assert cmd["service_data"]["params"] == {"zone": [1, 2, 3, 4]} + + +async def test_vacuum_send_command_without_params(client: HAClient, fake_ha: FakeHA) -> None: + """``send_command`` omits the ``params`` key when none are provided.""" + robo = client.vacuum("roborock") + robo._apply_state( + {"state": "cleaning", "attributes": {"supported_features": _VACUUM_FULL_FEATURES}} + ) + + await robo.send_command("find_dock") + cmd = _find_call(fake_ha, "send_command") + assert cmd["service_data"]["command"] == "find_dock" + assert "params" not in cmd["service_data"] + + +async def test_vacuum_unsupported_features_are_noops(client: HAClient, fake_ha: FakeHA) -> None: + """Actions degrade safely on vacuums that lack the relevant feature bits.""" + basic = client.vacuum("basic") + # No supported_features attribute at all. + basic._apply_state({"state": "docked", "attributes": {}}) + + assert basic.supports_start is False + assert basic.supports_pause is False + assert basic.supports_stop is False + assert basic.supports_return_home is False + assert basic.supports_locate is False + assert basic.supports_fan_speed is False + assert basic.supports_send_command is False + assert basic.supports_clean_spot is False + + await basic.start() + await basic.pause() + await basic.stop() + await basic.return_to_base() + await basic.locate() + await basic.clean_spot() + await basic.set_fan_speed("turbo") + await basic.send_command("find_dock", {"x": 1}) + + assert fake_ha.ws_service_calls == [] + + # Non-int supported_features is treated as unsupported. + basic._apply_state({"state": "docked", "attributes": {"supported_features": "32"}}) + assert basic.supports_fan_speed is False + + +async def test_vacuum_state_props() -> None: + """Vacuum state convenience properties reflect the underlying string state.""" + ha = HAClient.from_url("http://x", token="t", load_plugins=False) + try: + robo = ha.vacuum("roborock") + robo._apply_state( + { + "state": "cleaning", + "attributes": { + "battery_level": 72, + "fan_speed": "balanced", + "fan_speed_list": ["quiet", "balanced", "turbo"], + }, + } + ) + assert robo.is_cleaning + assert not robo.is_docked + assert robo.battery_level == 72 + assert robo.fan_speed == "balanced" + assert robo.fan_speed_list == ["quiet", "balanced", "turbo"] + + for state, prop in [ + ("docked", "is_docked"), + ("idle", "is_idle"), + ("paused", "is_paused"), + ("returning", "is_returning"), + ("error", "is_error"), + ]: + robo._apply_state({"state": state, "attributes": {}}) + assert getattr(robo, prop) + + # Missing / malformed attributes degrade to None / empty. + robo._apply_state({"state": "docked", "attributes": {}}) + assert robo.battery_level is None + assert robo.fan_speed is None + assert robo.fan_speed_list == [] + + # Non-string entries are filtered out of fan_speed_list. + robo._apply_state({"state": "docked", "attributes": {"fan_speed_list": ["a", 1, "b"]}}) + assert robo.fan_speed_list == ["a", "b"] + + # Non-list fan_speed_list returns empty list. + robo._apply_state({"state": "docked", "attributes": {"fan_speed_list": "not-a-list"}}) + assert robo.fan_speed_list == [] + finally: + await ha.close() + + +async def test_vacuum_listeners(client: HAClient, fake_ha: FakeHA) -> None: + """Vacuum listener decorators fire on the relevant transitions.""" + robo = client.vacuum("hall") + started: list[tuple[Any, Any]] = [] + docked: list[tuple[Any, Any]] = [] + errored: list[tuple[Any, Any]] = [] + battery: list[tuple[Any, Any]] = [] + fan: list[tuple[Any, Any]] = [] + + @robo.on_start + def _on_start(old: Any, new: Any) -> None: + started.append((old, new)) + + @robo.on_dock + def _on_dock(old: Any, new: Any) -> None: + docked.append((old, new)) + + @robo.on_error + def _on_error(old: Any, new: Any) -> None: + errored.append((old, new)) + + @robo.on_battery_change + def _on_battery(old: Any, new: Any) -> None: + battery.append((old, new)) + + @robo.on_fan_speed_change + def _on_fan(old: Any, new: Any) -> None: + fan.append((old, new)) + + await fake_ha.push_state_changed( + "vacuum.hall", + {"state": "cleaning", "attributes": {"battery_level": 75, "fan_speed": "turbo"}}, + {"state": "docked", "attributes": {"battery_level": 80, "fan_speed": "balanced"}}, + ) + await fake_ha.push_state_changed( + "vacuum.hall", + {"state": "docked", "attributes": {"battery_level": 90, "fan_speed": "balanced"}}, + {"state": "cleaning", "attributes": {"battery_level": 75, "fan_speed": "turbo"}}, + ) + await fake_ha.push_state_changed( + "vacuum.hall", + {"state": "error", "attributes": {}}, + {"state": "docked", "attributes": {}}, + ) + await asyncio.sleep(0.05) + + assert started == [("docked", "cleaning")] + assert docked == [("cleaning", "docked")] + assert errored == [("docked", "error")] + assert battery == [(80, 75), (75, 90)] + assert fan == [("balanced", "turbo"), ("turbo", "balanced")]