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

Initial working version

Browse Source
This commit is contained in:
Michael Graf 2021-03-20 16:11:04 +01:00
parent 771f91e362
commit fb1f2869f2
12 changed files with 1090 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

115
custom_components/vorwerk/__init__.py Normal file
View File

@ -0,0 +1,115 @@
"""Support for botvac connected Vorwerk vacuum cleaners."""
import asyncio
from datetime import timedelta
import logging
from pybotvac.exceptions import NeatoException
from pybotvac.robot import Robot
from pybotvac.vorwerk import Vorwerk
import voluptuous as vol
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.util import Throttle
from .const import (
VORWERK_DOMAIN,
VORWERK_PLATFORMS,
VORWERK_ROBOT_ENDPOINT,
VORWERK_ROBOT_NAME,
VORWERK_ROBOT_SECRET,
VORWERK_ROBOT_SERIAL,
VORWERK_ROBOT_TRAITS,
VORWERK_ROBOTS,
)
_LOGGER = logging.getLogger(__name__)
VORWERK_SCHEMA = vol.Schema(
vol.All(
{
vol.Required(VORWERK_ROBOT_NAME): cv.string,
vol.Required(VORWERK_ROBOT_SERIAL): cv.string,
vol.Required(VORWERK_ROBOT_SECRET): cv.string,
vol.Optional(
VORWERK_ROBOT_ENDPOINT, default="https://nucleo.ksecosys.com:4443"
): cv.string,
}
)
)
CONFIG_SCHEMA = vol.Schema(
{VORWERK_DOMAIN: vol.Schema(vol.All(cv.ensure_list, [VORWERK_SCHEMA]))},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
"""Set up the Vorwerk component."""
hass.data[VORWERK_DOMAIN] = {}
if VORWERK_DOMAIN in config:
hass.async_create_task(
hass.config_entries.flow.async_init(
VORWERK_DOMAIN,
context={"source": SOURCE_IMPORT},
data=config[VORWERK_DOMAIN],
)
)
return True
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Set up config entry."""
@Throttle(timedelta(minutes=1))
def create_robot(config):
return Robot(
serial=config[VORWERK_ROBOT_SERIAL],
secret=config[VORWERK_ROBOT_SECRET],
traits=config.get(VORWERK_ROBOT_TRAITS, []),
vendor=Vorwerk(),
name=config[VORWERK_ROBOT_NAME],
endpoint=config[VORWERK_ROBOT_ENDPOINT],
)
try:
robots = await asyncio.gather(
*(
hass.async_add_executor_job(create_robot, robot_conf)
for robot_conf in entry.data[VORWERK_ROBOTS]
),
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
)
raise ConfigEntryNotReady from ex
for component in VORWERK_PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
"""Unload config entry."""
unload_ok: bool = all(
await asyncio.gather(
*(
hass.config_entries.async_forward_entry_unload(entry, component)
for component in VORWERK_PLATFORMS
)
)
)
if unload_ok:
hass.data[VORWERK_DOMAIN].pop(entry.entry_id)
return unload_ok

18
custom_components/vorwerk/authsession.py Normal file
View File

@ -0,0 +1,18 @@
"""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

118
custom_components/vorwerk/config_flow.py Normal file
View File

@ -0,0 +1,118 @@
"""Config flow to configure Vorwerk integration."""
import logging
from typing import Any, Dict, Optional
from pybotvac.exceptions import NeatoException
from requests.models import HTTPError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_CODE, CONF_EMAIL, CONF_TOKEN
from . import authsession
# pylint: disable=unused-import
from .const import (
VORWERK_DOMAIN,
VORWERK_ROBOT_ENDPOINT,
VORWERK_ROBOT_NAME,
VORWERK_ROBOT_SECRET,
VORWERK_ROBOT_SERIAL,
VORWERK_ROBOT_TRAITS,
VORWERK_ROBOTS,
)
DOCS_URL = "https://www.home-assistant.io/integrations/vorwerk"
_LOGGER = logging.getLogger(__name__)
class VorwerkConfigFlow(config_entries.ConfigFlow, domain=VORWERK_DOMAIN):
"""Vorwerk integration config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
def __init__(self):
"""Initialize the config flow."""
self._email: Optional[str] = None
self._session = authsession.VorwerkSession()
async def async_step_user(self, user_input=None):
"""Step when user initializes a integration."""
if user_input is not None:
self._email = user_input.get(CONF_EMAIL)
if self._email:
await self.async_set_unique_id(self._email)
self._abort_if_unique_id_configured()
return await self.async_step_code()
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_EMAIL): str,
}
),
description_placeholders={"docs_url": DOCS_URL},
)
async def async_step_code(self, user_input: Dict[str, Any] = {}) -> Dict[str, Any]:
"""Step when user enters OTP Code from email."""
assert self._email is not None # typing
errors = {}
code = user_input.get(CONF_CODE) if user_input else None
if code:
try:
robots = await self.async_get_robots(self._email, code)
return self.async_create_entry(
title=self._email,
data={
CONF_EMAIL: self._email,
CONF_TOKEN: self._session.token,
VORWERK_ROBOTS: robots,
},
)
except (HTTPError, NeatoException):
errors["base"] = "invalid_auth"
self._session.send_email_otp(self._email)
return self.async_show_form(
step_id="code",
data_schema=vol.Schema(
{
vol.Required(CONF_CODE): str,
}
),
description_placeholders={"docs_url": DOCS_URL},
errors=errors,
)
async def async_step_import(self, user_input: Dict[str, Any]) -> Dict[str, Any]:
"""Import a config flow from configuration."""
unique_id = "from configuration"
data = {VORWERK_ROBOTS: user_input}
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured(data)
_LOGGER.info("Creating new Vorwerk robot config entry")
return self.async_create_entry(
title="from configuration",
data=data,
)
async def async_get_robots(self, email: str, code: str):
"""Fetch the robot list from vorwerk."""
self._session.fetch_token_passwordless(email, code)
return [
{
VORWERK_ROBOT_NAME: robot["name"],
VORWERK_ROBOT_SERIAL: robot["serial"],
VORWERK_ROBOT_SECRET: robot["secret_key"],
VORWERK_ROBOT_TRAITS: robot["traits"],
VORWERK_ROBOT_ENDPOINT: robot["nucleo_url"],
}
for robot in self._session.get("users/me/robots").json()
]

164
custom_components/vorwerk/const.py Normal file
View File

@ -0,0 +1,164 @@
"""Constants for Vorwerk integration."""
VORWERK_DOMAIN = "vorwerk"
VORWERK_ROBOTS = "robots"
VORWERK_ROBOT_NAME = "name"
VORWERK_ROBOT_SERIAL = "serial"
VORWERK_ROBOT_SECRET = "secret"
VORWERK_ROBOT_TRAITS = "traits"
VORWERK_ROBOT_ENDPOINT = "endpoint"
VORWERK_PLATFORMS = ["vacuum", "switch", "sensor"]
SCAN_INTERVAL_MINUTES = 1
MODE = {1: "Eco", 2: "Turbo"}
ACTION = {
0: "Invalid",
1: "House Cleaning",
2: "Spot Cleaning",
3: "Manual Cleaning",
4: "Docking",
5: "User Menu Active",
6: "Suspended Cleaning",
7: "Updating",
8: "Copying logs",
9: "Recovering Location",
10: "IEC test",
11: "Map cleaning",
12: "Exploring map (creating a persistent map)",
13: "Acquiring Persistent Map IDs",
14: "Creating & Uploading Map",
15: "Suspended Exploration",
}
ERRORS = {
"ui_error_battery_battundervoltlithiumsafety": "Replace battery",
"ui_error_battery_critical": "Replace battery",
"ui_error_battery_invalidsensor": "Replace battery",
"ui_error_battery_lithiumadapterfailure": "Replace battery",
"ui_error_battery_mismatch": "Replace battery",
"ui_error_battery_nothermistor": "Replace battery",
"ui_error_battery_overtemp": "Replace battery",
"ui_error_battery_overvolt": "Replace battery",
"ui_error_battery_undercurrent": "Replace battery",
"ui_error_battery_undertemp": "Replace battery",
"ui_error_battery_undervolt": "Replace battery",
"ui_error_battery_unplugged": "Replace battery",
"ui_error_brush_stuck": "Brush stuck",
"ui_error_brush_overloaded": "Brush overloaded",
"ui_error_bumper_stuck": "Bumper stuck",
"ui_error_check_battery_switch": "Check battery",
"ui_error_corrupt_scb": "Call customer service corrupt board",
"ui_error_deck_debris": "Deck debris",
"ui_error_dflt_app": "Check MyKobold app",
"ui_error_disconnect_chrg_cable": "Disconnected charge cable",
"ui_error_disconnect_usb_cable": "Disconnected USB cable",
"ui_error_dust_bin_missing": "Dust bin missing",
"ui_error_dust_bin_full": "Dust bin full",
"ui_error_dust_bin_emptied": "Dust bin emptied",
"ui_error_hardware_failure": "Hardware failure",
"ui_error_ldrop_stuck": "Clear my path",
"ui_error_lds_jammed": "Clear my path",
"ui_error_lds_bad_packets": "Check MyKobold app",
"ui_error_lds_disconnected": "Check MyKobold app",
"ui_error_lds_missed_packets": "Check MyKobold app",
"ui_error_lwheel_stuck": "Clear my path",
"ui_error_navigation_backdrop_frontbump": "Clear my path",
"ui_error_navigation_backdrop_leftbump": "Clear my path",
"ui_error_navigation_backdrop_wheelextended": "Clear my path",
"ui_error_navigation_noprogress": "Clear my path",
"ui_error_navigation_origin_unclean": "Clear my path",
"ui_error_navigation_pathproblems": "Cannot return to base",
"ui_error_navigation_pinkycommsfail": "Clear my path",
"ui_error_navigation_falling": "Clear my path",
"ui_error_navigation_noexitstogo": "Clear my path",
"ui_error_navigation_nomotioncommands": "Clear my path",
"ui_error_navigation_rightdrop_leftbump": "Clear my path",
"ui_error_navigation_undockingfailed": "Clear my path",
"ui_error_picked_up": "Picked up",
"ui_error_qa_fail": "Check MyKobold app",
"ui_error_rdrop_stuck": "Clear my path",
"ui_error_reconnect_failed": "Reconnect failed",
"ui_error_rwheel_stuck": "Clear my path",
"ui_error_stuck": "Stuck!",
"ui_error_unable_to_return_to_base": "Unable to return to base",
"ui_error_unable_to_see": "Clean vacuum sensors",
"ui_error_vacuum_slip": "Clear my path",
"ui_error_vacuum_stuck": "Clear my path",
"ui_error_warning": "Error check app",
"batt_base_connect_fail": "Battery failed to connect to base",
"batt_base_no_power": "Battery base has no power",
"batt_low": "Battery low",
"batt_on_base": "Battery on base",
"clean_tilt_on_start": "Clean the tilt on start",
"dustbin_full": "Dust bin full",
"dustbin_missing": "Dust bin missing",
"gen_picked_up": "Picked up",
"hw_fail": "Hardware failure",
"hw_tof_sensor_sensor": "Hardware sensor disconnected",
"lds_bad_packets": "Bad packets",
"lds_deck_debris": "Debris on deck",
"lds_disconnected": "Disconnected",
"lds_jammed": "Jammed",
"lds_missed_packets": "Missed packets",
"maint_brush_stuck": "Brush stuck",
"maint_brush_overload": "Brush overloaded",
"maint_bumper_stuck": "Bumper stuck",
"maint_customer_support_qa": "Contact customer support",
"maint_vacuum_stuck": "Vacuum is stuck",
"maint_vacuum_slip": "Vacuum is stuck",
"maint_left_drop_stuck": "Vacuum is stuck",
"maint_left_wheel_stuck": "Vacuum is stuck",
"maint_right_drop_stuck": "Vacuum is stuck",
"maint_right_wheel_stuck": "Vacuum is stuck",
"not_on_charge_base": "Not on the charge base",
"nav_robot_falling": "Clear my path",
"nav_no_path": "Clear my path",
"nav_path_problem": "Clear my path",
"nav_backdrop_frontbump": "Clear my path",
"nav_backdrop_leftbump": "Clear my path",
"nav_backdrop_wheelextended": "Clear my path",
"nav_mag_sensor": "Clear my path",
"nav_no_exit": "Clear my path",
"nav_no_movement": "Clear my path",
"nav_rightdrop_leftbump": "Clear my path",
"nav_undocking_failed": "Clear my path",
}
ALERTS = {
"ui_alert_dust_bin_full": "Please empty dust bin",
"ui_alert_recovering_location": "Returning to start",
"ui_alert_battery_chargebasecommerr": "Battery error",
"ui_alert_busy_charging": "Busy charging",
"ui_alert_charging_base": "Base charging",
"ui_alert_charging_power": "Charging power",
"ui_alert_connect_chrg_cable": "Connect charge cable",
"ui_alert_info_thank_you": "Thank you",
"ui_alert_invalid": "Invalid check app",
"ui_alert_old_error": "Old error",
"ui_alert_swupdate_fail": "Update failed",
"dustbin_full": "Please empty dust bin",
"maint_brush_change": "Change the brush",
"maint_filter_change": "Change the filter",
"clean_completed_to_start": "Cleaning completed",
"nav_floorplan_not_created": "No floorplan found",
"nav_floorplan_load_fail": "Failed to load floorplan",
"nav_floorplan_localization_fail": "Failed to load floorplan",
"clean_incomplete_to_start": "Cleaning incomplete",
"log_upload_failed": "Logs failed to upload",
}
ATTR_CLEAN_START = "clean_start"
ATTR_CLEAN_STOP = "clean_stop"
ATTR_CLEAN_AREA = "clean_area"
ATTR_CLEAN_BATTERY_START = "battery_level_at_clean_start"
ATTR_CLEAN_BATTERY_END = "battery_level_at_clean_end"
ATTR_CLEAN_SUSP_COUNT = "clean_suspension_count"
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"

15
custom_components/vorwerk/manifest.json Normal file
View File

@ -0,0 +1,15 @@
{
"domain": "vorwerk",
"name": "Vorwerk Kobold",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/vorwerk",
"requirements": [
"pybotvac==0.0.20"
],
"codeowners": [
"@trunneml"
],
"dependencies": [
"http"
]
}

92
custom_components/vorwerk/sensor.py Normal file
View File

@ -0,0 +1,92 @@
"""Support for Vorwerk sensors."""
from datetime import timedelta
import logging
from pybotvac.exceptions import NeatoRobotException
from homeassistant.components.sensor import DEVICE_CLASS_BATTERY
from homeassistant.const import PERCENTAGE
from homeassistant.helpers.entity import Entity
from .const import SCAN_INTERVAL_MINUTES, VORWERK_DOMAIN, VORWERK_ROBOTS
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
BATTERY = "Battery"
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up the Vorwerk sensor using config entry."""
_LOGGER.debug("Adding sensors for vorwerk robots")
async_add_entities(
[
VorwerkSensor(robot)
for robot in hass.data[VORWERK_DOMAIN][entry.entry_id][VORWERK_ROBOTS]
],
True,
)
class VorwerkSensor(Entity):
"""Vorwerk sensor."""
def __init__(self, robot):
"""Initialize Vorwerk sensor."""
self.robot = robot
self._available = False
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):
"""Return the name of this sensor."""
return self._robot_name
@property
def unique_id(self):
"""Return unique ID."""
return self._robot_serial
@property
def device_class(self):
"""Return the device class."""
return DEVICE_CLASS_BATTERY
@property
def available(self):
"""Return availability."""
return self._available
@property
def state(self):
"""Return the state."""
return self._state["details"]["charge"]
@property
def unit_of_measurement(self):
"""Return unit of measurement."""
return PERCENTAGE
@property
def device_info(self):
"""Device info for robot."""
return {"identifiers": {(VORWERK_DOMAIN, self._robot_serial)}}

18
custom_components/vorwerk/services.yaml Normal file
View File

@ -0,0 +1,18 @@
custom_cleaning:
description: Zone Cleaning service call specific to Vorwerk Kobolds.
fields:
entity_id:
description: Name of the vacuum entity. [Required]
example: "vacuum.mein_vr"
mode:
description: "Set the cleaning mode: 1 for eco and 2 for turbo. Defaults to turbo if not set."
example: 2
navigation:
description: "Set the navigation mode: 1 for normal, 2 for extra care, 3 for deep. Defaults to normal if not set."
example: 1
category:
description: "Whether to use a persistent map or not for cleaning (i.e. No go lines): 2 for no map, 4 for map. Default to using map if not set (and fallback to no map if no map is found)."
example: 2
zone:
description: Only supported on the VR300. Name of the zone to clean. Defaults to no zone i.e. complete house cleanup.
example: "Kitchen"

22
custom_components/vorwerk/strings.json Normal file
View File

@ -0,0 +1,22 @@
{
"config": {
"step": {
"user": {
"title": "Vorwerk Account Info",
"data": {
"email": "[%key:common::config_flow::data::email%]",
"code": "Code"
},
"description": "To recieve an authentication code, enter the email address of your vorwerk account.\n\nSee [Vorwerk documentation]({docs_url})."
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
}
},
"title": "Vorwerk Botvac"
}

123
custom_components/vorwerk/switch.py Normal file
View File

@ -0,0 +1,123 @@
"""Support for Vorwerk Connected Vacuums switches."""
from datetime import timedelta
import logging
from pybotvac.exceptions import NeatoRobotException
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.helpers.entity import ToggleEntity
from .const import SCAN_INTERVAL_MINUTES, VORWERK_DOMAIN, 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)
for robot in hass.data[VORWERK_DOMAIN][entry.entry_id][VORWERK_ROBOTS]
for switch_type in SWITCH_TYPES
]
if not dev:
return
async_add_entities(dev, True)
class VorwerkConnectedSwitch(ToggleEntity):
"""Vorwerk Connected 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
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."""
return self._robot_name
@property
def available(self):
"""Return True if entity is available."""
return self._available
@property
def unique_id(self):
"""Return a unique ID."""
return self._robot_serial
@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
@property
def device_info(self):
"""Device info for robot."""
return {"identifiers": {(VORWERK_DOMAIN, self._robot_serial)}}
def turn_on(self, **kwargs):
"""Turn the switch on."""
if self.type == SWITCH_TYPE_SCHEDULE:
try:
self.robot.enable_schedule()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk switch connection error '%s': %s", self.entity_id, ex
)
def turn_off(self, **kwargs):
"""Turn the switch off."""
if self.type == SWITCH_TYPE_SCHEDULE:
try:
self.robot.disable_schedule()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk switch connection error '%s': %s", self.entity_id, ex
)

29
custom_components/vorwerk/translations/de.json Normal file
View File

@ -0,0 +1,29 @@
{
"config": {
"step": {
"user": {
"title": "Vorwerk-Kontoinformationen",
"data": {
"email": "E-Mailaddresse"
},
"description": "Um einen Authentifizierungscode per E-Mail zu erhalten, gib die E-Mailadresse deines Vorwerk-Accounts ein.\n\nSiehe [Vorwerk-Dokumentation]({docs_url})."
},
"code": {
"title": "Vorwerk Account Info",
"data": {
"code": "Code"
},
"description": "Gib den per E-Mail erhaltenen Code ein.\n\nSee [Vorwerk documentation]({docs_url})."
}
},
},
"error": {
"invalid_auth": "Ung\u00fcltige Authentifizierung"
},
"abort": {
"invalid_auth": "Ung\u00fcltige Authentifizierung",
"already_configured": "Bereits konfiguriert"
}
},
"title": "Vorwerk Kobold"
}

28
custom_components/vorwerk/translations/en.json Normal file
View File

@ -0,0 +1,28 @@
{
"config": {
"step": {
"user": {
"title": "Vorwerk Account Info",
"data": {
"email": "Email"
},
"description": "To recieve an authentication code, enter the email address of your vorwerk account.\n\nSee [Vorwerk documentation]({docs_url})."
},
"code": {
"title": "Vorwerk Account Info",
"data": {
"code": "Code"
},
"description": "Enter the code you received by email.\n\nSee [Vorwerk documentation]({docs_url})."
}
},
"error": {
"invalid_auth": "Invalid authentication"
},
"abort": {
"already_configured": "Account is already configured",
"invalid_auth": "Invalid authentication"
}
},
"title": "Vorwerk Kobold"
}

348
custom_components/vorwerk/vacuum.py Normal file
View File

@ -0,0 +1,348 @@
"""Support for Neato Connected Vacuums."""
from datetime import timedelta
import logging
from pybotvac.exceptions import NeatoRobotException
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,
SUPPORT_PAUSE,
SUPPORT_RETURN_HOME,
SUPPORT_START,
SUPPORT_STATE,
SUPPORT_STOP,
StateVacuumEntity,
)
from homeassistant.const import ATTR_MODE
from homeassistant.helpers import config_validation as cv, entity_platform
from .const import (
ACTION,
ALERTS,
ATTR_CLEAN_AREA,
ATTR_CLEAN_BATTERY_END,
ATTR_CLEAN_BATTERY_START,
ATTR_CLEAN_ERROR_TIME,
ATTR_CLEAN_PAUSE_TIME,
ATTR_CLEAN_START,
ATTR_CLEAN_STOP,
ATTR_CLEAN_SUSP_COUNT,
ATTR_CLEAN_SUSP_TIME,
ATTR_LAUNCHED_FROM,
ERRORS,
MODE,
SCAN_INTERVAL_MINUTES,
VORWERK_DOMAIN,
VORWERK_ROBOTS,
)
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
SUPPORT_VORWERK = (
SUPPORT_BATTERY
| SUPPORT_PAUSE
| SUPPORT_RETURN_HOME
| SUPPORT_STOP
| SUPPORT_START
| SUPPORT_CLEAN_SPOT
| SUPPORT_STATE
| SUPPORT_LOCATE
)
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)
for robot in hass.data[VORWERK_DOMAIN][entry.entry_id][VORWERK_ROBOTS]
],
True,
)
platform = entity_platform.current_platform.get()
assert platform is not None
platform.async_register_entity_service(
"custom_cleaning",
{
vol.Optional(ATTR_MODE, default=2): cv.positive_int,
vol.Optional(ATTR_NAVIGATION, default=1): cv.positive_int,
vol.Optional(ATTR_CATEGORY, default=4): cv.positive_int,
vol.Optional(ATTR_ZONE): cv.string,
},
"vorwerk_custom_cleaning",
)
class VorwerkConnectedVacuum(StateVacuumEntity):
"""Representation of a Vorwerk Connected Vacuum."""
def __init__(self, robot):
"""Initialize the Vorwerk Connected Vacuum."""
self.robot = robot
self._available = False
self._name = f"{self.robot.name}"
self._robot_has_map = False
self._robot_serial = self.robot.serial
self._status_state = None
self._clean_state = None
self._state = None
self._clean_time_start = None
self._clean_time_stop = None
self._clean_area = None
self._clean_battery_start = None
self._clean_battery_end = None
self._clean_susp_charge_count = None
self._clean_susp_time = None
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):
"""Return the name of the device."""
return self._name
@property
def supported_features(self):
"""Flag vacuum cleaner robot features that are supported."""
return SUPPORT_VORWERK
@property
def battery_level(self):
"""Return the battery level of the vacuum cleaner."""
return self._battery_level
@property
def available(self):
"""Return if the robot is available."""
return self._available
@property
def icon(self):
"""Return specific icon."""
return "mdi:robot-vacuum-variant"
@property
def state(self):
"""Return the status of the vacuum cleaner."""
return self._clean_state
@property
def unique_id(self):
"""Return a unique ID."""
return self._robot_serial
@property
def device_state_attributes(self):
"""Return the state attributes of the vacuum cleaner."""
data = {}
if self._status_state is not None:
data[ATTR_STATUS] = self._status_state
if self._clean_time_start is not None:
data[ATTR_CLEAN_START] = self._clean_time_start
if self._clean_time_stop is not None:
data[ATTR_CLEAN_STOP] = self._clean_time_stop
if self._clean_area is not None:
data[ATTR_CLEAN_AREA] = self._clean_area
if self._clean_susp_charge_count is not None:
data[ATTR_CLEAN_SUSP_COUNT] = self._clean_susp_charge_count
if self._clean_susp_time is not None:
data[ATTR_CLEAN_SUSP_TIME] = self._clean_susp_time
if self._clean_pause_time is not None:
data[ATTR_CLEAN_PAUSE_TIME] = self._clean_pause_time
if self._clean_error_time is not None:
data[ATTR_CLEAN_ERROR_TIME] = self._clean_error_time
if self._clean_battery_start is not None:
data[ATTR_CLEAN_BATTERY_START] = self._clean_battery_start
if self._clean_battery_end is not None:
data[ATTR_CLEAN_BATTERY_END] = self._clean_battery_end
if self._launched_from is not None:
data[ATTR_LAUNCHED_FROM] = self._launched_from
return data
@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
def start(self):
"""Start cleaning or resume cleaning."""
try:
if self._state["state"] == 1:
self.robot.start_cleaning()
elif self._state["state"] == 3:
self.robot.resume_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk vacuum connection error for '%s': %s", self.entity_id, ex
)
def pause(self):
"""Pause the vacuum."""
try:
self.robot.pause_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk vacuum connection error for '%s': %s", self.entity_id, ex
)
def return_to_base(self, **kwargs):
"""Set the vacuum cleaner to return to the dock."""
try:
if self._clean_state == STATE_CLEANING:
self.robot.pause_cleaning()
self._clean_state = STATE_RETURNING
self.robot.send_to_base()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk vacuum connection error for '%s': %s", self.entity_id, ex
)
def stop(self, **kwargs):
"""Stop the vacuum cleaner."""
try:
self.robot.stop_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk vacuum connection error for '%s': %s", self.entity_id, ex
)
def locate(self, **kwargs):
"""Locate the robot by making it emit a sound."""
try:
self.robot.locate()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk vacuum connection error for '%s': %s", self.entity_id, ex
)
def clean_spot(self, **kwargs):
"""Run a spot cleaning starting from the base."""
try:
self.robot.start_spot_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk vacuum connection error for '%s': %s", self.entity_id, ex
)
def vorwerk_custom_cleaning(self, mode, navigation, category, zone=None):
"""Zone cleaning service call."""
boundary_id = None
if zone is not None:
for boundary in self._robot_boundaries:
if zone in boundary["name"]:
boundary_id = boundary["id"]
if boundary_id is None:
_LOGGER.error(
"Zone '%s' was not found for the robot '%s'", zone, self.entity_id
)
return
self._clean_state = STATE_CLEANING
try:
self.robot.start_cleaning(mode, navigation, category, boundary_id)
except NeatoRobotException as ex:
_LOGGER.error(
"Vorwerk vacuum connection error for '%s': %s", self.entity_id, ex
)