diff --git a/__init__.py b/__init__.py index 6576415..edd32d2 100644 --- a/__init__.py +++ b/__init__.py @@ -1,21 +1,40 @@ """Support for botvac connected Vorwerk vacuum cleaners.""" +from __future__ import annotations + import asyncio import logging +from typing import Any -from pybotvac.exceptions import NeatoException +from pybotvac.exceptions import NeatoException, NeatoRobotException from pybotvac.robot import Robot from pybotvac.vorwerk import Vorwerk import voluptuous as vol +from homeassistant.components.vacuum import ( + STATE_CLEANING, + STATE_DOCKED, + STATE_ERROR, + STATE_IDLE, + STATE_PAUSED, + STATE_RETURNING, +) from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry from homeassistant.exceptions import ConfigEntryNotReady import homeassistant.helpers.config_validation as cv from homeassistant.helpers.typing import ConfigType, HomeAssistantType from homeassistant.helpers.update_coordinator import DataUpdateCoordinator -from .api import VorwerkState from .const import ( + ACTION, + ALERTS, + ERRORS, MIN_TIME_BETWEEN_UPDATES, + MODE, + ROBOT_ACTION_DOCKING, + ROBOT_STATE_BUSY, + ROBOT_STATE_ERROR, + ROBOT_STATE_IDLE, + ROBOT_STATE_PAUSE, VORWERK_DOMAIN, VORWERK_PLATFORMS, VORWERK_ROBOT_API, @@ -145,3 +164,164 @@ async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> boo if unload_ok: hass.data[VORWERK_DOMAIN].pop(entry.entry_id) return unload_ok + + +class VorwerkState: + """Class to convert robot_state dict to more useful object.""" + + def __init__(self, robot: Robot) -> None: + """Initialize new vorwerk vacuum state.""" + self.robot = robot + self.robot_state: dict[Any, Any] = {} + self.robot_info: dict[Any, Any] = {} + + @property + def available(self) -> bool: + """Return true when robot state is available.""" + return bool(self.robot_state) + + def update(self): + """Update robot state and robot info.""" + _LOGGER.debug("Running Vorwerk Vacuums update for '%s'", self.robot.name) + self._update_robot_info() + self._update_state() + + def _update_state(self): + try: + if self.robot_info is None: + self.robot_info = self.robot.get_general_info().json().get("data") + except NeatoRobotException: + _LOGGER.warning("Couldn't fetch robot information of %s", self.robot.name) + + def _update_robot_info(self): + try: + self.robot_state = self.robot.state + except NeatoRobotException as ex: + if self.available: # print only once when available + _LOGGER.error( + "Vorwerk vacuum connection error for '%s': %s", self.robot.name, ex + ) + self.robot_state = {} + return + + @property + def docked(self) -> bool | None: + """Vacuum is docked.""" + if not self.available: + return None + return ( + self.robot_state["state"] == ROBOT_STATE_IDLE + and self.robot_state["details"]["isDocked"] + ) + + @property + def charging(self) -> bool | None: + """Vacuum is charging.""" + if not self.available: + return None + return ( + self.robot_state.get("state") == ROBOT_STATE_IDLE + and self.robot_state["details"]["isCharging"] + ) + + @property + def state(self) -> str | None: + """Return Home Assistant vacuum state.""" + if not self.available: + return None + robot_state = self.robot_state.get("state") + state = None + if self.charging or self.docked: + state = STATE_DOCKED + elif robot_state == ROBOT_STATE_IDLE: + state = STATE_IDLE + elif robot_state == ROBOT_STATE_BUSY: + if robot_state["action"] != ROBOT_ACTION_DOCKING: + state = STATE_RETURNING + else: + state = STATE_CLEANING + elif robot_state == ROBOT_STATE_PAUSE: + state = STATE_PAUSED + elif robot_state == ROBOT_STATE_ERROR: + state = STATE_ERROR + return state + + @property + def alert(self) -> str | None: + """Return vacuum alert message.""" + if not self.available: + return None + if "alert" in self.robot_state: + return ALERTS.get(self.robot_state["alert"], self.robot_state["alert"]) + return None + + @property + def status(self) -> str | None: + """Return vacuum status message.""" + if not self.available: + return None + + status = None + if self.state == STATE_ERROR: + status = self._error_status() + elif self.alert: + status = self.alert + elif self.state == STATE_DOCKED: + if self.charging: + status = "Charging" + if self.docked: + status = "Docked" + elif self.state == STATE_IDLE: + status = "Stopped" + elif self.state == STATE_CLEANING: + status = self._cleaning_status() + elif self.state == STATE_PAUSED: + status = "Paused" + + return status + + def _error_status(self): + """Return error status.""" + robot_state = self.robot_state.get("state") + return ERRORS.get(robot_state["error"], robot_state["error"]) + + def _cleaning_status(self): + """Return cleaning status.""" + robot_state = self.robot_state.get("state") + status_items = [ + MODE.get(robot_state["cleaning"]["mode"]), + ACTION.get(robot_state["action"]), + ] + if ( + "boundary" in robot_state["cleaning"] + and "name" in robot_state["cleaning"]["boundary"] + ): + status_items.append(robot_state["cleaning"]["boundary"]["name"]) + return " ".join(s for s in status_items if s) + + @property + def battery_level(self) -> str | None: + """Return the battery level of the vacuum cleaner.""" + if not self.available: + return None + return self.robot_state["details"]["charge"] + + @property + def device_info(self) -> dict[str, str]: + """Device info for robot.""" + info = { + "identifiers": {(VORWERK_DOMAIN, self.robot.serial)}, + "name": self.robot.name, + } + if self.robot_info: + info["manufacturer"] = self.robot_info["battery"]["vendor"] + info["model"] = self.robot_info["model"] + info["sw_version"] = self.robot_info["firmware"] + return info + + @property + def schedule_enabled(self): + """Return True when schedule is enabled.""" + if not self.available: + return None + return bool(self.robot_state["details"]["isScheduleEnabled"]) diff --git a/api.py b/api.py deleted file mode 100644 index d674beb..0000000 --- a/api.py +++ /dev/null @@ -1,209 +0,0 @@ -"""Auth sessions for pybotvac.""" -from __future__ import annotations - -import logging -from typing import Any - -import pybotvac -from pybotvac.exceptions import NeatoRobotException - -from homeassistant.components.vacuum import ( - STATE_CLEANING, - STATE_DOCKED, - STATE_ERROR, - STATE_IDLE, - STATE_PAUSED, - STATE_RETURNING, -) - -from .const import ( - ACTION, - ALERTS, - ERRORS, - MODE, - ROBOT_ACTION_DOCKING, - ROBOT_STATE_BUSY, - ROBOT_STATE_ERROR, - ROBOT_STATE_IDLE, - ROBOT_STATE_PAUSE, - VORWERK_DOMAIN, -) - -_LOGGER = logging.getLogger(__name__) - - -class VorwerkSession(pybotvac.PasswordlessSession): - """PasswordlessSession pybotvac session for Vorwerk cloud.""" - - # The client_id is the same for all users. - CLIENT_ID = "KY4YbVAvtgB7lp8vIbWQ7zLk3hssZlhR" - - def __init__(self): - """Initialize Vorwerk cloud session.""" - super().__init__(client_id=VorwerkSession.CLIENT_ID, vendor=pybotvac.Vorwerk()) - - @property - def token(self): - """Return the token dict. Contains id_token, access_token and refresh_token.""" - return self._token - - -class VorwerkState: - """Class to convert robot_state dict to more useful object.""" - - def __init__(self, robot: pybotvac.Robot) -> None: - """Initialize new vorwerk vacuum state.""" - self.robot = robot - self.robot_state: dict[Any, Any] = {} - self.robot_info: dict[Any, Any] = {} - - @property - def available(self) -> bool: - """Return true when robot state is available.""" - return bool(self.robot_state) - - def update(self): - """Update robot state and robot info.""" - _LOGGER.debug("Running Vorwerk Vacuums update for '%s'", self.robot.name) - self._update_robot_info() - self._update_state() - - def _update_state(self): - try: - if self.robot_info is None: - self.robot_info = self.robot.get_general_info().json().get("data") - except NeatoRobotException: - _LOGGER.warning("Couldn't fetch robot information of %s", self.robot.name) - - def _update_robot_info(self): - try: - self.robot_state = self.robot.state - except NeatoRobotException as ex: - if self.available: # print only once when available - _LOGGER.error( - "Vorwerk vacuum connection error for '%s': %s", self.robot.name, ex - ) - self.robot_state = {} - return - - @property - def docked(self) -> bool | None: - """Vacuum is docked.""" - if not self.available: - return None - return ( - self.robot_state["state"] == ROBOT_STATE_IDLE - and self.robot_state["details"]["isDocked"] - ) - - @property - def charging(self) -> bool | None: - """Vacuum is charging.""" - if not self.available: - return None - return ( - self.robot_state.get("state") == ROBOT_STATE_IDLE - and self.robot_state["details"]["isCharging"] - ) - - @property - def state(self) -> str | None: - """Return Home Assistant vacuum state.""" - if not self.available: - return None - robot_state = self.robot_state.get("state") - state = None - if self.charging or self.docked: - state = STATE_DOCKED - elif robot_state == ROBOT_STATE_IDLE: - state = STATE_IDLE - elif robot_state == ROBOT_STATE_BUSY: - if robot_state["action"] != ROBOT_ACTION_DOCKING: - state = STATE_RETURNING - else: - state = STATE_CLEANING - elif robot_state == ROBOT_STATE_PAUSE: - state = STATE_PAUSED - elif robot_state == ROBOT_STATE_ERROR: - state = STATE_ERROR - return state - - @property - def alert(self) -> str | None: - """Return vacuum alert message.""" - if not self.available: - return None - if "alert" in self.robot_state: - return ALERTS.get(self.robot_state["alert"], self.robot_state["alert"]) - return None - - @property - def status(self) -> str | None: - """Return vacuum status message.""" - if not self.available: - return None - - status = None - if self.state == STATE_ERROR: - status = self._error_status() - elif self.alert: - status = self.alert - elif self.state == STATE_DOCKED: - if self.charging: - status = "Charging" - if self.docked: - status = "Docked" - elif self.state == STATE_IDLE: - status = "Stopped" - elif self.state == STATE_CLEANING: - status = self._cleaning_status() - elif self.state == STATE_PAUSED: - status = "Paused" - - return status - - def _error_status(self): - """Return error status.""" - robot_state = self.robot_state.get("state") - return ERRORS.get(robot_state["error"], robot_state["error"]) - - def _cleaning_status(self): - """Return cleaning status.""" - robot_state = self.robot_state.get("state") - status_items = [ - MODE.get(robot_state["cleaning"]["mode"]), - ACTION.get(robot_state["action"]), - ] - if ( - "boundary" in robot_state["cleaning"] - and "name" in robot_state["cleaning"]["boundary"] - ): - status_items.append(robot_state["cleaning"]["boundary"]["name"]) - return " ".join(s for s in status_items if s) - - @property - def battery_level(self) -> str | None: - """Return the battery level of the vacuum cleaner.""" - if not self.available: - return None - return self.robot_state["details"]["charge"] - - @property - def device_info(self) -> dict[str, str]: - """Device info for robot.""" - info = { - "identifiers": {(VORWERK_DOMAIN, self.robot.serial)}, - "name": self.robot.name, - } - if self.robot_info: - info["manufacturer"] = self.robot_info["battery"]["vendor"] - info["model"] = self.robot_info["model"] - info["sw_version"] = self.robot_info["firmware"] - return info - - @property - def schedule_enabled(self): - """Return True when schedule is enabled.""" - if not self.available: - return None - return bool(self.robot_state["details"]["isScheduleEnabled"]) diff --git a/config_flow.py b/config_flow.py index b7288a9..9306dfb 100644 --- a/config_flow.py +++ b/config_flow.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging from typing import Any +import pybotvac from pybotvac.exceptions import NeatoException from requests.models import HTTPError import voluptuous as vol @@ -11,10 +12,10 @@ import voluptuous as vol from homeassistant import config_entries from homeassistant.const import CONF_CODE, CONF_EMAIL, CONF_TOKEN -from . import api # pylint: disable=unused-import from .const import ( + VORWERK_CLIENT_ID, VORWERK_DOMAIN, VORWERK_ROBOT_ENDPOINT, VORWERK_ROBOT_NAME, @@ -36,7 +37,7 @@ class VorwerkConfigFlow(config_entries.ConfigFlow, domain=VORWERK_DOMAIN): def __init__(self): """Initialize the config flow.""" self._email: str | None = None - self._session = api.VorwerkSession() + self._session = VorwerkSession() async def async_step_user(self, user_input=None): """Step when user initializes a integration.""" @@ -121,3 +122,16 @@ class VorwerkConfigFlow(config_entries.ConfigFlow, domain=VORWERK_DOMAIN): } for robot in self._session.get("users/me/robots").json() ] + + +class VorwerkSession(pybotvac.PasswordlessSession): + """PasswordlessSession pybotvac session for Vorwerk cloud.""" + + def __init__(self): + """Initialize Vorwerk cloud session.""" + super().__init__(client_id=VORWERK_CLIENT_ID, vendor=pybotvac.Vorwerk()) + + @property + def token(self): + """Return the token dict. Contains id_token, access_token and refresh_token.""" + return self._token diff --git a/const.py b/const.py index 403042b..d5c45bf 100644 --- a/const.py +++ b/const.py @@ -15,6 +15,9 @@ VORWERK_ROBOT_ENDPOINT = "endpoint" VORWERK_PLATFORMS = ["vacuum", "switch", "sensor"] +# The client_id is the same for all users. +VORWERK_CLIENT_ID = "KY4YbVAvtgB7lp8vIbWQ7zLk3hssZlhR" + MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1) MODE = {1: "Eco", 2: "Turbo"} diff --git a/sensor.py b/sensor.py index aa50dd1..060c88e 100644 --- a/sensor.py +++ b/sensor.py @@ -11,7 +11,7 @@ from homeassistant.helpers.update_coordinator import ( DataUpdateCoordinator, ) -from .api import VorwerkState +from . import VorwerkState from .const import ( VORWERK_DOMAIN, VORWERK_ROBOT_API, diff --git a/switch.py b/switch.py index 847bdce..292d503 100644 --- a/switch.py +++ b/switch.py @@ -11,7 +11,7 @@ from homeassistant.helpers.update_coordinator import ( DataUpdateCoordinator, ) -from .api import VorwerkState +from . import VorwerkState from .const import ( VORWERK_DOMAIN, VORWERK_ROBOT_API, diff --git a/vacuum.py b/vacuum.py index 00e9944..113c1f6 100644 --- a/vacuum.py +++ b/vacuum.py @@ -30,7 +30,7 @@ from homeassistant.helpers.update_coordinator import ( DataUpdateCoordinator, ) -from .api import VorwerkState +from . import VorwerkState from .const import ( ATTR_CATEGORY, ATTR_NAVIGATION,