SFS/homeassistant-vorwerk
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use DataUpdateCoordinator

Browse Source
This commit is contained in:
Michael Graf 2021-04-25 20:30:12 +02:00
parent bba9a3c1eb
commit d10af9824a
9 changed files with 398 additions and 234 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
__init__.py
View File

@ -12,11 +12,16 @@ 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 homeassistant.util import Throttle
from .api import VorwerkState
from .const import (
MIN_TIME_BETWEEN_UPDATES,
VORWERK_DOMAIN,
VORWERK_PLATFORMS,
VORWERK_ROBOT_API,
VORWERK_ROBOT_COORDINATOR,
VORWERK_ROBOT_ENDPOINT,
VORWERK_ROBOT_NAME,
VORWERK_ROBOT_SECRET,
@ -65,7 +70,45 @@ async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Set up config entry."""
robots = await _async_create_robots(hass, entry.data[VORWERK_ROBOTS])
robot_states = [ VorwerkState(robot) for robot in robots ]
hass.data[VORWERK_DOMAIN][entry.entry_id] = {
VORWERK_ROBOTS: [
{
VORWERK_ROBOT_API: r,
VORWERK_ROBOT_COORDINATOR: _create_coordinator(hass, r),
}
for r in robot_states
]
}
for component in VORWERK_PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
def _create_coordinator(
hass: HomeAssistantType, robot_state: VorwerkState
) -> DataUpdateCoordinator:
async def async_update_data():
"""Fetch data from API endpoint."""
await hass.async_add_executor_job(robot_state.update)
return DataUpdateCoordinator(
hass,
_LOGGER,
name=robot_state.robot.name,
update_method=async_update_data,
update_interval=MIN_TIME_BETWEEN_UPDATES,
)
async def _async_create_robots(hass, robot_confs):
@Throttle(timedelta(minutes=1))
def create_robot(config):
return Robot(
@ -77,27 +120,19 @@ async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool
endpoint=config[VORWERK_ROBOT_ENDPOINT],
)
robots = []
try:
robots = await asyncio.gather(
*(
hass.async_add_executor_job(create_robot, robot_conf)
for robot_conf in entry.data[VORWERK_ROBOTS]
for robot_conf in robot_confs
),
return_exceptions=False,
)
hass.data[VORWERK_DOMAIN][entry.entry_id] = {VORWERK_ROBOTS: robots}
except NeatoException as ex:
_LOGGER.warning(
"Failed to connect to robot %s: %s", entry.data[VORWERK_ROBOT_NAME], ex
)
_LOGGER.error("Failed to connect to robots: %s", ex)
raise ConfigEntryNotReady from ex
for component in VORWERK_PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
return robots
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:

214
api.py Normal file
View File

@ -0,0 +1,214 @@
"""Auth sessions for pybotvac."""
from functools import wraps
import logging
from typing import Optional
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
def when_available(f):
"""Prevent calling the method and return None when not available."""
@wraps(f)
def wrapper(self, *args, **kw):
if self.available:
return f(self, *args, **kw)
else:
return None
return wrapper
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 = {}
self.robot_info = {}
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
self._available = True
@property
@when_available
def docked(self) -> Optional[bool]:
"""Vacuum is docked."""
return (
self.robot_state["state"] == ROBOT_STATE_IDLE
and self.robot_state["details"]["isDocked"]
)
@property
@when_available
def charging(self) -> Optional[bool]:
"""Vacuum is charging."""
return (
self.robot_state.get("state") == ROBOT_STATE_IDLE
and self.robot_state["details"]["isCharging"]
)
@property
@when_available
def state(self) -> Optional[str]:
"""Return Home Assistant vacuum state."""
robot_state = self.robot_state.get("state")
if self.charging or self.docked:
return STATE_DOCKED
elif robot_state == ROBOT_STATE_IDLE:
return STATE_IDLE
elif robot_state == ROBOT_STATE_BUSY:
if robot_state["action"] != ROBOT_ACTION_DOCKING:
return STATE_RETURNING
else:
return STATE_CLEANING
elif robot_state == ROBOT_STATE_PAUSE:
return STATE_PAUSED
elif robot_state == ROBOT_STATE_ERROR:
return STATE_ERROR
return None
@property
@when_available
def alert(self) -> Optional[str]:
"""Return vacuum alert message."""
if "alert" in self.robot_state:
return ALERTS.get(self.robot_state["alert"], self.robot_state["alert"])
return None
@property
@when_available
def status(self) -> Optional[str]:
"""Return vacuum status message."""
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
@when_available
def battery_level(self):
"""Return the battery level of the vacuum cleaner."""
return self.robot_state["details"]["charge"]
@property
def device_info(self):
"""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
@when_available
def scheduleEnabled(self) -> Optional[bool]:
"""Return True when schedule is enabled."""
return bool(self.robot_state["details"]["isScheduleEnabled"])

18
authsession.py
View File

@ -1,18 +0,0 @@
"""Auth sessions for pybotvac."""
import pybotvac
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

15
config_flow.py
View File

@ -9,7 +9,7 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_CODE, CONF_EMAIL, CONF_TOKEN
from . import authsession
from . import api
# pylint: disable=unused-import
from .const import (
@ -36,7 +36,7 @@ class VorwerkConfigFlow(config_entries.ConfigFlow, domain=VORWERK_DOMAIN):
def __init__(self):
"""Initialize the config flow."""
self._email: Optional[str] = None
self._session = authsession.VorwerkSession()
self._session = api.VorwerkSession()
async def async_step_user(self, user_input=None):
"""Step when user initializes a integration."""
@ -67,7 +67,9 @@ class VorwerkConfigFlow(config_entries.ConfigFlow, domain=VORWERK_DOMAIN):
code = user_input.get(CONF_CODE) if user_input else None
if code:
try:
robots = await self.async_get_robots(self._email, code)
robots = await self.hass.async_add_executor_job(
self._get_robots, self._email, code
)
return self.async_create_entry(
title=self._email,
data={
@ -79,7 +81,10 @@ class VorwerkConfigFlow(config_entries.ConfigFlow, domain=VORWERK_DOMAIN):
except (HTTPError, NeatoException):
errors["base"] = "invalid_auth"
self._session.send_email_otp(self._email)
await self.hass.async_add_executor_job(
self._session.send_email_otp, self._email
)
return self.async_show_form(
step_id="code",
data_schema=vol.Schema(
@ -105,7 +110,7 @@ class VorwerkConfigFlow(config_entries.ConfigFlow, domain=VORWERK_DOMAIN):
data=data,
)
async def async_get_robots(self, email: str, code: str):
def _get_robots(self, email: str, code: str):
"""Fetch the robot list from vorwerk."""
self._session.fetch_token_passwordless(email, code)
return [

21
const.py
View File

@ -1,8 +1,11 @@
"""Constants for Vorwerk integration."""
from datetime import timedelta
VORWERK_DOMAIN = "vorwerk"
VORWERK_ROBOTS = "robots"
VORWERK_ROBOT_API = "robot_api"
VORWERK_ROBOT_COORDINATOR = "robot_coordinator"
VORWERK_ROBOT_NAME = "name"
VORWERK_ROBOT_SERIAL = "serial"
@ -12,7 +15,7 @@ VORWERK_ROBOT_ENDPOINT = "endpoint"
VORWERK_PLATFORMS = ["vacuum", "switch", "sensor"]
SCAN_INTERVAL_MINUTES = 1
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=1)
MODE = {1: "Eco", 2: "Turbo"}
@ -162,3 +165,19 @@ ATTR_CLEAN_SUSP_TIME = "clean_suspension_time"
ATTR_CLEAN_PAUSE_TIME = "clean_pause_time"
ATTR_CLEAN_ERROR_TIME = "clean_error_time"
ATTR_LAUNCHED_FROM = "launched_from"
ATTR_NAVIGATION = "navigation"
ATTR_CATEGORY = "category"
ATTR_ZONE = "zone"
ROBOT_STATE_INVALID = 0
ROBOT_STATE_IDLE = 1
ROBOT_STATE_BUSY = 2
ROBOT_STATE_PAUSE = 3
ROBOT_STATE_ERROR = 4
ROBOT_ACTION_HOUSE_CLEANING = 1
ROBOT_ACTION_SPOT_CLEANING = 2
ROBOT_ACTION_MANUAL_CLEANING = 3
ROBOT_ACTION_DOCKING = 4

3
manifest.json
View File

@ -11,5 +11,6 @@
],
"dependencies": [
"http"
]
],
"iot_class": "cloud_polling"
}

52
sensor.py
View File

@ -1,18 +1,26 @@
"""Support for Vorwerk sensors."""
from datetime import timedelta
import logging
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from homeassistant.components.sensor import DEVICE_CLASS_BATTERY
from homeassistant.const import PERCENTAGE
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import SCAN_INTERVAL_MINUTES, VORWERK_DOMAIN, VORWERK_ROBOTS
from .api import VorwerkState
from .const import (
VORWERK_DOMAIN,
VORWERK_ROBOT_API,
VORWERK_ROBOT_COORDINATOR,
VORWERK_ROBOTS,
)
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
BATTERY = "Battery"
@ -22,39 +30,25 @@ async def async_setup_entry(hass, entry, async_add_entities):
_LOGGER.debug("Adding sensors for vorwerk robots")
async_add_entities(
[
VorwerkSensor(robot)
VorwerkSensor(robot[VORWERK_ROBOT_API], robot[VORWERK_ROBOT_COORDINATOR])
for robot in hass.data[VORWERK_DOMAIN][entry.entry_id][VORWERK_ROBOTS]
],
True,
)
class VorwerkSensor(Entity):
class VorwerkSensor(CoordinatorEntity, Entity):
"""Vorwerk sensor."""
def __init__(self, robot):
def __init__(
self, robot_state: VorwerkState, coordinator: DataUpdateCoordinator
) -> None:
"""Initialize Vorwerk sensor."""
self.robot = robot
self._available = False
super().__init__(coordinator)
self.robot: Robot = robot_state.robot
self._state: VorwerkState = robot_state
self._robot_name = f"{self.robot.name} {BATTERY}"
self._robot_serial = self.robot.serial
self._state = None
def update(self):
"""Update Vorwerk Sensor."""
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._available:
_LOGGER.error(
"Vorwerk sensor connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._available = False
return
self._available = True
_LOGGER.debug("self._state=%s", self._state)
@property
def name(self):
@ -74,12 +68,12 @@ class VorwerkSensor(Entity):
@property
def available(self):
"""Return availability."""
return self._available
return self._state.available
@property
def state(self):
"""Return the state."""
return self._state["details"]["charge"]
return self._state.battery_level
@property
def unit_of_measurement(self):
@ -89,4 +83,4 @@ class VorwerkSensor(Entity):
@property
def device_info(self):
"""Device info for robot."""
return {"identifiers": {(VORWERK_DOMAIN, self._robot_serial)}}
return self._state.device_info

101
switch.py
View File

@ -1,31 +1,36 @@
"""Support for Vorwerk Connected Vacuums switches."""
from datetime import timedelta
import logging
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.helpers.entity import ToggleEntity
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import SCAN_INTERVAL_MINUTES, VORWERK_DOMAIN, VORWERK_ROBOTS
from .api import VorwerkState
from .const import (
VORWERK_DOMAIN,
VORWERK_ROBOT_API,
VORWERK_ROBOT_COORDINATOR,
VORWERK_ROBOTS,
)
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
SWITCH_TYPE_SCHEDULE = "schedule"
SWITCH_TYPES = {SWITCH_TYPE_SCHEDULE: ["Schedule"]}
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Vorwerk switch with config entry."""
_LOGGER.debug("Adding switches for vorwerk (%s)", entry.title)
dev = [
VorwerkConnectedSwitch(robot, switch_type)
VorwerkScheduleSwitch(
robot[VORWERK_ROBOT_API], robot[VORWERK_ROBOT_COORDINATOR]
)
for robot in hass.data[VORWERK_DOMAIN][entry.entry_id][VORWERK_ROBOTS]
for switch_type in SWITCH_TYPES
]
if not dev:
@ -34,46 +39,19 @@ async def async_setup_entry(hass, entry, async_add_entities):
async_add_entities(dev, True)
class VorwerkConnectedSwitch(ToggleEntity):
"""Vorwerk Connected Switches."""
class VorwerkScheduleSwitch(CoordinatorEntity, ToggleEntity):
"""Vorwerk Schedule Switches."""
def __init__(self, robot, switch_type):
"""Initialize the Vorwerk Connected switches."""
self.type = switch_type
self.robot = robot
self._available = False
self._robot_name = f"{self.robot.name} {SWITCH_TYPES[self.type][0]}"
self._state = None
self._schedule_state = None
self._clean_state = None
def __init__(
self, robot_state: VorwerkState, coordinator: DataUpdateCoordinator
) -> None:
"""Initialize the Vorwerk Schedule switch."""
super().__init__(coordinator)
self.robot: Robot = robot_state.robot
self._robot_name = f"{self.robot.name} Schedule"
self._state: VorwerkState = robot_state
self._robot_serial = self.robot.serial
def update(self):
"""Update the states of Vorwerk switches."""
_LOGGER.debug("Running Vorwerk switch update for '%s'", self.entity_id)
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._available: # Print only once when available
_LOGGER.error(
"Vorwerk switch connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._available = False
return
self._available = True
_LOGGER.debug("self._state=%s", self._state)
if self.type == SWITCH_TYPE_SCHEDULE:
_LOGGER.debug("State: %s", self._state)
if self._state["details"]["isScheduleEnabled"]:
self._schedule_state = STATE_ON
else:
self._schedule_state = STATE_OFF
_LOGGER.debug(
"Schedule state for '%s': %s", self.entity_id, self._schedule_state
)
@property
def name(self):
"""Return the name of the switch."""
@ -82,7 +60,7 @@ class VorwerkConnectedSwitch(ToggleEntity):
@property
def available(self):
"""Return True if entity is available."""
return self._available
return self._state.available
@property
def unique_id(self):
@ -92,19 +70,21 @@ class VorwerkConnectedSwitch(ToggleEntity):
@property
def is_on(self):
"""Return true if switch is on."""
if self.type == SWITCH_TYPE_SCHEDULE:
if self._schedule_state == STATE_ON:
return True
return False
if self._state.available:
if self._state.scheduleEnabled:
return STATE_ON
else:
return STATE_OFF
@property
def device_info(self):
"""Device info for robot."""
return {"identifiers": {(VORWERK_DOMAIN, self._robot_serial)}}
return self._state.device_info
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs):
"""Turn the switch on."""
if self.type == SWITCH_TYPE_SCHEDULE:
def turn_on():
try:
self.robot.enable_schedule()
except NeatoRobotException as ex:
@ -112,12 +92,19 @@ class VorwerkConnectedSwitch(ToggleEntity):
"Vorwerk switch connection error '%s': %s", self.entity_id, ex
)
def turn_off(self, **kwargs):
await self.hass.async_add_executor_job(turn_on)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs):
"""Turn the switch off."""
if self.type == SWITCH_TYPE_SCHEDULE:
def turn_off():
try:
self.robot.disable_schedule()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk switch connection error '%s': %s", self.entity_id, ex
)
await self.hass.async_add_executor_job(turn_off)
await self.coordinator.async_request_refresh()

149
vacuum.py
View File

@ -1,18 +1,16 @@
"""Support for Neato Connected Vacuums."""
from datetime import timedelta
import logging
from typing import Any
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
import voluptuous as vol
from homeassistant.components.vacuum import (
ATTR_STATUS,
STATE_CLEANING,
STATE_DOCKED,
STATE_ERROR,
STATE_IDLE,
STATE_PAUSED,
STATE_RETURNING,
SUPPORT_BATTERY,
SUPPORT_CLEAN_SPOT,
SUPPORT_LOCATE,
@ -23,12 +21,17 @@ from homeassistant.components.vacuum import (
SUPPORT_STOP,
StateVacuumEntity,
)
from homeassistant.const import ATTR_MODE
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .api import VorwerkState
from .const import (
ACTION,
ALERTS,
ATTR_CATEGORY,
ATTR_CLEAN_AREA,
ATTR_CLEAN_BATTERY_END,
ATTR_CLEAN_BATTERY_START,
@ -39,16 +42,16 @@ from .const import (
ATTR_CLEAN_SUSP_COUNT,
ATTR_CLEAN_SUSP_TIME,
ATTR_LAUNCHED_FROM,
ERRORS,
MODE,
SCAN_INTERVAL_MINUTES,
ATTR_NAVIGATION,
ATTR_ZONE,
VORWERK_DOMAIN,
VORWERK_ROBOT_API,
VORWERK_ROBOT_COORDINATOR,
VORWERK_ROBOTS,
)
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
SUPPORT_VORWERK = (
SUPPORT_BATTERY
@ -62,18 +65,15 @@ SUPPORT_VORWERK = (
)
ATTR_NAVIGATION = "navigation"
ATTR_CATEGORY = "category"
ATTR_ZONE = "zone"
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Vorwerk vacuum with config entry."""
_LOGGER.debug("Adding vorwerk vacuums")
async_add_entities(
[
VorwerkConnectedVacuum(robot)
VorwerkConnectedVacuum(
robot[VORWERK_ROBOT_API], robot[VORWERK_ROBOT_COORDINATOR]
)
for robot in hass.data[VORWERK_DOMAIN][entry.entry_id][VORWERK_ROBOTS]
],
True,
@ -94,19 +94,23 @@ async def async_setup_entry(hass, entry, async_add_entities):
)
class VorwerkConnectedVacuum(StateVacuumEntity):
class VorwerkConnectedVacuum(CoordinatorEntity, StateVacuumEntity):
"""Representation of a Vorwerk Connected Vacuum."""
def __init__(self, robot):
def __init__(
self, robot_state: VorwerkState, coordinator: DataUpdateCoordinator[Any]
) -> None:
"""Initialize the Vorwerk Connected Vacuum."""
self.robot = robot
self._available = False
super().__init__(coordinator)
self.robot: Robot = robot_state.robot
self._state: VorwerkState = robot_state
self._name = f"{self.robot.name}"
self._robot_has_map = False
self._robot_serial = self.robot.serial
self._status_state = None
# Variables form neato impl
# We keep it here for later implementations
self._clean_state = None
self._state = None
self._clean_time_start = None
self._clean_time_stop = None
self._clean_area = None
@ -117,76 +121,7 @@ class VorwerkConnectedVacuum(StateVacuumEntity):
self._clean_pause_time = None
self._clean_error_time = None
self._launched_from = None
self._battery_level = None
self._robot_boundaries = []
self._robot_stats = None
def update(self):
"""Update the states of Vorwerk Vacuums."""
_LOGGER.debug("Running Vorwerk Vacuums update for '%s'", self.entity_id)
try:
if self._robot_stats is None:
self._robot_stats = self.robot.get_general_info().json().get("data")
except NeatoRobotException:
_LOGGER.warning("Couldn't fetch robot information of %s", self.entity_id)
try:
self._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.entity_id, ex
)
self._state = None
self._available = False
return
self._available = True
_LOGGER.debug("self._state=%s", self._state)
if "alert" in self._state:
robot_alert = ALERTS.get(self._state["alert"])
else:
robot_alert = None
if self._state["state"] == 1:
if self._state["details"]["isCharging"]:
self._clean_state = STATE_DOCKED
self._status_state = "Charging"
elif (
self._state["details"]["isDocked"]
and not self._state["details"]["isCharging"]
):
self._clean_state = STATE_DOCKED
self._status_state = "Docked"
else:
self._clean_state = STATE_IDLE
self._status_state = "Stopped"
if robot_alert is not None:
self._status_state = robot_alert
elif self._state["state"] == 2:
if robot_alert is None:
self._clean_state = STATE_CLEANING
self._status_state = (
f"{MODE.get(self._state['cleaning']['mode'])} "
f"{ACTION.get(self._state['action'])}"
)
if (
"boundary" in self._state["cleaning"]
and "name" in self._state["cleaning"]["boundary"]
):
self._status_state += (
f" {self._state['cleaning']['boundary']['name']}"
)
else:
self._status_state = robot_alert
elif self._state["state"] == 3:
self._clean_state = STATE_PAUSED
self._status_state = "Paused"
elif self._state["state"] == 4:
self._clean_state = STATE_ERROR
self._status_state = ERRORS.get(self._state["error"])
self._battery_level = self._state["details"]["charge"]
@property
def name(self):
@ -201,12 +136,12 @@ class VorwerkConnectedVacuum(StateVacuumEntity):
@property
def battery_level(self):
"""Return the battery level of the vacuum cleaner."""
return self._battery_level
return self._state.battery_level
@property
def available(self):
"""Return if the robot is available."""
return self._available
return self._state.available
@property
def icon(self):
@ -216,7 +151,7 @@ class VorwerkConnectedVacuum(StateVacuumEntity):
@property
def state(self):
"""Return the status of the vacuum cleaner."""
return self._clean_state
return self._state.state if self._state else None
@property
def unique_id(self):
@ -228,8 +163,8 @@ class VorwerkConnectedVacuum(StateVacuumEntity):
"""Return the state attributes of the vacuum cleaner."""
data = {}
if self._status_state is not None:
data[ATTR_STATUS] = self._status_state
if self._state.status is not None:
data[ATTR_STATUS] = self._state.status
if self._clean_time_start is not None:
data[ATTR_CLEAN_START] = self._clean_time_start
if self._clean_time_stop is not None:
@ -256,22 +191,16 @@ class VorwerkConnectedVacuum(StateVacuumEntity):
@property
def device_info(self):
"""Device info for robot."""
info = {
"identifiers": {(VORWERK_DOMAIN, self._robot_serial)},
"name": self._name,
}
if self._robot_stats:
info["manufacturer"] = self._robot_stats["battery"]["vendor"]
info["model"] = self._robot_stats["model"]
info["sw_version"] = self._robot_stats["firmware"]
return info
return self._state.device_info
def start(self):
"""Start cleaning or resume cleaning."""
if not self._state:
return
try:
if self._state["state"] == 1:
if self._state.state == STATE_IDLE:
self.robot.start_cleaning()
elif self._state["state"] == 3:
elif self._state.state == STATE_PAUSED:
self.robot.resume_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
@ -290,9 +219,8 @@ class VorwerkConnectedVacuum(StateVacuumEntity):
def return_to_base(self, **kwargs):
"""Set the vacuum cleaner to return to the dock."""
try:
if self._clean_state == STATE_CLEANING:
if self._state.state == STATE_CLEANING:
self.robot.pause_cleaning()
self._clean_state = STATE_RETURNING
self.robot.send_to_base()
except NeatoRobotException as ex:
_LOGGER.error(
@ -339,7 +267,6 @@ class VorwerkConnectedVacuum(StateVacuumEntity):
)
return
self._clean_state = STATE_CLEANING
try:
self.robot.start_cleaning(mode, navigation, category, boundary_id)
except NeatoRobotException as ex: