home-automation-home-assistant/custom_components/deebot/vacuum.py
2021-03-21 18:41:41 +01:00

247 line
7.9 KiB
Python

"""Support for Deebot Vaccums."""
import base64
from typing import Optional, Dict, Any, Union, List
from deebotozmo import *
from homeassistant.util import slugify
from . import HUB as hub
CONF_COUNTRY = "country"
CONF_CONTINENT = "continent"
CONF_DEVICEID = "deviceid"
CONF_LIVEMAPPATH = "livemappath"
CONF_LIVEMAP = "live_map"
CONF_SHOWCOLORROOMS = "show_color_rooms"
DEEBOT_DEVICES = "deebot_devices"
from homeassistant.components.vacuum import (
PLATFORM_SCHEMA,
STATE_CLEANING,
STATE_DOCKED,
STATE_ERROR,
STATE_IDLE,
STATE_PAUSED,
STATE_RETURNING,
SUPPORT_BATTERY,
SUPPORT_FAN_SPEED,
SUPPORT_LOCATE,
SUPPORT_PAUSE,
SUPPORT_RETURN_HOME,
SUPPORT_SEND_COMMAND,
SUPPORT_START,
SUPPORT_STATE,
VacuumEntity,
)
_LOGGER = logging.getLogger(__name__)
SUPPORT_DEEBOT = (
SUPPORT_BATTERY
| SUPPORT_FAN_SPEED
| SUPPORT_LOCATE
| SUPPORT_PAUSE
| SUPPORT_RETURN_HOME
| SUPPORT_SEND_COMMAND
| SUPPORT_START
| SUPPORT_STATE
)
STATE_CODE_TO_STATE = {
'STATE_IDLE': STATE_IDLE,
'STATE_CLEANING': STATE_CLEANING,
'STATE_RETURNING': STATE_RETURNING,
'STATE_DOCKED': STATE_DOCKED,
'STATE_ERROR': STATE_ERROR,
'STATE_PAUSED': STATE_PAUSED,
}
ATTR_COMPONENT_PREFIX = "component_"
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Deebot vacuums."""
if DEEBOT_DEVICES not in hass.data:
hass.data[DEEBOT_DEVICES] = []
for vacbot in hub.vacbots:
vacuum = DeebotVacuum(hass, vacbot)
add_devices([vacuum])
class DeebotVacuum(VacuumEntity):
"""Deebot Vacuums"""
def __init__(self, hass, vacbot):
"""Initialize the Deebot Vacuum."""
self._hass = hass
self.device = vacbot
if self.device.vacuum.get("nick", None) is not None:
self._name = "{}".format(self.device.vacuum["nick"])
else:
# In case there is no nickname defined, use the device id
self._name = "{}".format(self.device.vacuum["did"])
self._fan_speed = None
self._live_map = None
self._live_map_path = hub.config.get(CONF_LIVEMAPPATH) + self._name + '_liveMap.png'
self.device.refresh_statuses()
_LOGGER.debug("Vacuum initialized: %s", self.name)
def on_fan_change(self, fan_speed):
self._fan_speed = fan_speed
@property
def should_poll(self) -> bool:
"""Return True if entity has to be polled for state."""
return True
@property
def unique_id(self) -> str:
"""Return an unique ID."""
return self.device.vacuum.get("did", None)
@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_DEEBOT
@property
def state(self):
"""Return the state of the vacuum cleaner."""
if self.device.vacuum_status is not None and self.device.is_available == True:
return STATE_CODE_TO_STATE[self.device.vacuum_status]
@property
def available(self) -> bool:
"""Return True if entity is available."""
return self.device.is_available
async def async_return_to_base(self, **kwargs):
"""Set the vacuum cleaner to return to the dock."""
await self.hass.async_add_executor_job(self.device.Charge)
@property
def battery_level(self):
"""Return the battery level of the vacuum cleaner."""
if self.device.battery_status is not None:
return self.device.battery_status
return super().battery_level
@property
def fan_speed(self):
"""Return the fan speed of the vacuum cleaner."""
return self.device.fan_speed
async def async_set_fan_speed(self, fan_speed, **kwargs):
await self.hass.async_add_executor_job(self.device.SetFanSpeed, fan_speed)
@property
def fan_speed_list(self):
"""Get the list of available fan speed steps of the vacuum cleaner."""
return [FAN_SPEED_QUIET, FAN_SPEED_NORMAL, FAN_SPEED_MAX, FAN_SPEED_MAXPLUS]
async def async_pause(self):
"""Pause the vacuum cleaner."""
await self.hass.async_add_executor_job(self.device.CleanPause)
async def async_start(self):
"""Start the vacuum cleaner."""
await self.hass.async_add_executor_job(self.device.CleanResume)
async def async_locate(self, **kwargs):
"""Locate the vacuum cleaner."""
await self.hass.async_add_executor_job(self.device.PlaySound)
async def async_send_command(self, command, params=None, **kwargs):
"""Send a command to a vacuum cleaner."""
_LOGGER.debug("async_send_command %s (%s), %s", command, params, kwargs)
if command == 'spot_area':
await self.hass.async_add_executor_job(self.device.SpotArea, params['rooms'], params['cleanings'])
return
if command == 'custom_area':
await self.hass.async_add_executor_job(self.device.CustomArea, params['coordinates'], params['cleanings'])
return
if command == 'set_water':
await self.hass.async_add_executor_job(self.device.SetWaterLevel, params['amount'])
return
if command == 'relocate':
await self.hass.async_add_executor_job(self.device.Relocate)
return
if command == 'auto_clean':
self.hass.async_add_executor_job(self.device.Clean, params['type'])
return
if command == 'refresh_components':
await self.hass.async_add_executor_job(self.device.refresh_components)
return
if command == 'refresh_statuses':
await self.hass.async_add_executor_job(self.device.refresh_statuses)
return
if command == 'refresh_live_map':
await self.hass.async_add_executor_job(self.device.refresh_liveMap)
return
if command == 'save_live_map':
if(self._live_map != self.device.live_map):
self._live_map = self.device.live_map
with open(params['path'], "wb") as fh:
fh.write(base64.decodebytes(self.device.live_map))
await self.hass.async_add_executor_job(self.device.exc_command, command, params)
async def async_update(self):
"""Fetch state from the device."""
await self.hass.async_add_executor_job(self.device.request_all_statuses)
try:
if(self._live_map != self.device.live_map):
self._live_map = self.device.live_map
with open(self._live_map_path, "wb") as fh:
fh.write(base64.decodebytes(self.device.live_map))
except KeyError:
_LOGGER.warning("Can't access local folder: %s", self._live_map_path)
@property
def device_state_attributes(self) -> Optional[Dict[str, Any]]:
"""Return device specific state attributes.
Implemented by platform classes. Convention for attribute names
is lowercase snake_case.
"""
data: Dict[str, Union[int, List[int]]] = {}
# Needed for custom vacuum-card (https://github.com/denysdovhan/vacuum-card)
# Should find a better way without breaking everyone rooms script
data['status'] = STATE_CODE_TO_STATE[self.device.vacuum_status]
if self.device.getSavedRooms() is not None:
for r in self.device.getSavedRooms():
# convert room name to snake_case to meet the convention
room_name = "room_" + slugify(r["subtype"])
room_values = data.get(room_name)
if room_values is None:
data[room_name] = r["id"]
elif isinstance(room_values, list):
room_values.append(r["id"])
else:
# Convert from int to list
data[room_name] = [room_values, r["id"]]
return data