402 lines
16 KiB
Python
402 lines
16 KiB
Python
"""Support for Xiaomi vacuums."""
|
|
import logging
|
|
import asyncio
|
|
from datetime import timedelta
|
|
|
|
from homeassistant.components.vacuum import ( # noqa: F401
|
|
DOMAIN as ENTITY_DOMAIN,
|
|
StateVacuumEntity,
|
|
VacuumEntityFeature, # v2022.5
|
|
)
|
|
from .core.const import VacuumActivity
|
|
|
|
from . import (
|
|
DOMAIN,
|
|
CONF_MODEL,
|
|
XIAOMI_CONFIG_SCHEMA as PLATFORM_SCHEMA, # noqa: F401
|
|
HassEntry,
|
|
MiotEntity,
|
|
MIOT_LOCAL_MODELS,
|
|
async_setup_config_entry,
|
|
bind_services_to_entries,
|
|
)
|
|
from .core.utils import DeviceException
|
|
from .core.miot_spec import (
|
|
MiotSpec,
|
|
MiotService,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
DATA_KEY = f'{ENTITY_DOMAIN}.{DOMAIN}'
|
|
SCAN_INTERVAL = timedelta(seconds=60)
|
|
|
|
SERVICE_TO_METHOD = {}
|
|
|
|
|
|
async def async_setup_entry(hass, config_entry, async_add_entities):
|
|
HassEntry.init(hass, config_entry).new_adder(ENTITY_DOMAIN, async_add_entities)
|
|
await async_setup_config_entry(hass, config_entry, async_setup_platform, async_add_entities, ENTITY_DOMAIN)
|
|
|
|
|
|
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
|
|
hass.data.setdefault(DATA_KEY, {})
|
|
hass.data[DOMAIN]['add_entities'][ENTITY_DOMAIN] = async_add_entities
|
|
config['hass'] = hass
|
|
model = str(config.get(CONF_MODEL) or '')
|
|
spec = hass.data[DOMAIN]['miot_specs'].get(model)
|
|
entities = []
|
|
if isinstance(spec, MiotSpec):
|
|
for srv in spec.get_services(ENTITY_DOMAIN, 'mopping_machine'):
|
|
if not srv.get_property('status'):
|
|
continue
|
|
if model in MIOT_LOCAL_MODELS:
|
|
entities.append(MiotVacuumEntity(config, srv))
|
|
elif 'roborock.' in model or 'rockrobo.' in model:
|
|
entities.append(MiotRoborockVacuumEntity(config, srv))
|
|
elif 'viomi.' in model:
|
|
entities.append(MiotViomiVacuumEntity(config, srv))
|
|
else:
|
|
entities.append(MiotVacuumEntity(config, srv))
|
|
for entity in entities:
|
|
hass.data[DOMAIN]['entities'][entity.unique_id] = entity
|
|
async_add_entities(entities, update_before_add=True)
|
|
bind_services_to_entries(hass, SERVICE_TO_METHOD)
|
|
|
|
|
|
class MiotVacuumEntity(MiotEntity, StateVacuumEntity):
|
|
_attr_activity = None
|
|
|
|
def __init__(self, config: dict, miot_service: MiotService):
|
|
super().__init__(miot_service, config=config, logger=_LOGGER)
|
|
|
|
self._prop_power = miot_service.get_property('on', 'power')
|
|
self._prop_status = miot_service.get_property('status')
|
|
self._act_start = miot_service.get_action('start_sweep', 'start_mop')
|
|
self._act_pause = miot_service.get_action('pause_sweeping', 'pause')
|
|
self._act_stop = miot_service.get_action('stop_sweeping')
|
|
self._act_locate = miot_service.get_action('find_device', 'position')
|
|
self._prop_mode = miot_service.get_property('mode', 'clean_mode')
|
|
self._prop_fan = self._prop_mode
|
|
for srv in [*miot_service.spec.get_services('sweep', 'clean'), miot_service]:
|
|
if prop := srv.get_property('fan_level', 'speed_level', 'suction_state', 'fan_mode', 'mode'):
|
|
self._prop_fan = prop
|
|
break
|
|
self._prop_battery = miot_service.get_property('battery_level')
|
|
self._srv_battery = miot_service.spec.get_service('battery')
|
|
if self._srv_battery:
|
|
self._prop_battery = self._srv_battery.get_property('battery_level')
|
|
self._srv_audio = miot_service.spec.get_service('audio', 'voice')
|
|
if self._srv_audio and not self._act_locate:
|
|
self._act_locate = self._srv_audio.get_action('find_device', 'position')
|
|
self._act_charge = None
|
|
for srv in [*miot_service.spec.get_services('battery', 'go_charging'), miot_service]:
|
|
act = srv.get_action('start_charge', 'start_charging')
|
|
if act:
|
|
self._act_charge = act
|
|
break
|
|
|
|
if self._prop_power:
|
|
self._supported_features |= VacuumEntityFeature.TURN_ON
|
|
self._supported_features |= VacuumEntityFeature.TURN_OFF
|
|
if self._act_start:
|
|
self._supported_features |= VacuumEntityFeature.START
|
|
if self._act_pause:
|
|
self._supported_features |= VacuumEntityFeature.PAUSE
|
|
if self._act_stop:
|
|
self._supported_features |= VacuumEntityFeature.STOP
|
|
if self._act_charge:
|
|
self._supported_features |= VacuumEntityFeature.RETURN_HOME
|
|
if self._prop_fan:
|
|
self._supported_features |= VacuumEntityFeature.FAN_SPEED
|
|
if self._prop_status:
|
|
self._supported_features |= VacuumEntityFeature.STATUS
|
|
self._supported_features |= VacuumEntityFeature.STATE
|
|
if self._act_locate:
|
|
self._supported_features |= VacuumEntityFeature.LOCATE
|
|
|
|
async def async_update(self):
|
|
await super().async_update()
|
|
if not self._available:
|
|
return
|
|
if self._prop_status:
|
|
self._prop_status.description_to_dict(self._state_attrs)
|
|
val = self._prop_status.from_device(self.device)
|
|
if val is None:
|
|
pass
|
|
elif val in self._prop_status.list_search(
|
|
'Cleaning', 'Sweeping', 'Mopping', 'Sweeping And Mopping', 'Washing', 'Go Washing',
|
|
'Part Sweeping', 'Zone Sweeping', 'Select Sweeping', 'Spot Sweeping', 'Goto Target',
|
|
'Starting', 'Working', 'Busy', 'DustCollecting'
|
|
):
|
|
self._attr_activity = VacuumActivity.CLEANING
|
|
elif val in self._prop_status.list_search('Idle', 'Sleep'):
|
|
self._attr_activity = VacuumActivity.IDLE
|
|
elif val in self._prop_status.list_search('Charging', 'Charging Completed', 'Fullcharge', 'Charge Done', 'Drying'):
|
|
self._attr_activity = VacuumActivity.DOCKED
|
|
elif val in self._prop_status.list_search('Go Charging'):
|
|
self._attr_activity = VacuumActivity.RETURNING
|
|
elif val in self._prop_status.list_search('Paused'):
|
|
self._attr_activity = VacuumActivity.PAUSED
|
|
elif val in self._prop_status.list_search('Error', 'Charging Problem'):
|
|
self._attr_activity = VacuumActivity.ERROR
|
|
else:
|
|
self._attr_activity = VacuumActivity.IDLE
|
|
|
|
async def async_turn_on(self, **kwargs):
|
|
if self._prop_power:
|
|
await self.async_set_property(self._prop_power, True)
|
|
return await self.async_start()
|
|
|
|
async def async_turn_off(self, **kwargs):
|
|
return await self.async_stop()
|
|
|
|
async def async_start(self):
|
|
if self._act_start:
|
|
return await self.async_call_action(self._act_start)
|
|
return False
|
|
|
|
async def async_stop(self, **kwargs):
|
|
if self._act_stop:
|
|
return await self.async_call_action(self._act_stop)
|
|
return False
|
|
|
|
async def async_pause(self):
|
|
if self._act_pause:
|
|
return await self.async_call_action(self._act_pause)
|
|
return await self.async_stop()
|
|
|
|
async def async_return_to_base(self, **kwargs):
|
|
if self._act_charge:
|
|
return await self.async_call_action(self._act_charge)
|
|
return self.stop()
|
|
|
|
async def async_locate(self, **kwargs):
|
|
if self._act_locate:
|
|
return await self.async_call_action(self._act_locate)
|
|
return False
|
|
|
|
def clean_spot(self, **kwargs):
|
|
raise NotImplementedError()
|
|
|
|
@property
|
|
def fan_speed(self):
|
|
if self._prop_fan:
|
|
val = self._prop_fan.from_device(self.device)
|
|
try:
|
|
val = int(val)
|
|
except (TypeError, ValueError):
|
|
val = None
|
|
if val is not None:
|
|
return self._prop_fan.list_description(val)
|
|
return None
|
|
|
|
@property
|
|
def fan_speed_list(self):
|
|
if self._prop_fan:
|
|
return self._prop_fan.list_description(None) or []
|
|
return None
|
|
|
|
async def async_set_fan_speed(self, fan_speed, **kwargs):
|
|
if self._prop_fan:
|
|
val = self._prop_fan.list_value(fan_speed)
|
|
return self.async_set_property(self._prop_fan, val)
|
|
return False
|
|
|
|
async def async_send_command(self, command, params=None, **kwargs):
|
|
"""Send a command to a vacuum cleaner.
|
|
This method must be run in the event loop.
|
|
"""
|
|
return await self.async_miio_command(command, params)
|
|
|
|
|
|
class MiotRoborockVacuumEntity(MiotVacuumEntity):
|
|
def __init__(self, config: dict, miot_service: MiotService):
|
|
super().__init__(config, miot_service)
|
|
self._supported_features |= VacuumEntityFeature.PAUSE
|
|
self._supported_features |= VacuumEntityFeature.LOCATE
|
|
self._supported_features |= VacuumEntityFeature.SEND_COMMAND
|
|
|
|
async def async_added_to_hass(self):
|
|
await super().async_added_to_hass()
|
|
rooms = await self.get_room_mapping() or []
|
|
|
|
if add_buttons := self.device.entry.adders.get('button'):
|
|
from .button import ButtonSubEntity
|
|
for r in rooms:
|
|
if len(r) < 3:
|
|
continue
|
|
rid = r[0]
|
|
sub = f'segment_{rid}'
|
|
self._subs[sub] = ButtonSubEntity(self, sub, option={
|
|
'name': f'{self.device_name} {r[2]}',
|
|
'async_press_action': self.async_start_clean_segment,
|
|
'press_kwargs': {'segment': rid},
|
|
'state_attrs': {'room_id': r[1]},
|
|
})
|
|
add_buttons([self._subs[sub]], update_before_add=False)
|
|
self.logger.info('Room buttons: %s', [rooms, add_buttons])
|
|
|
|
async def async_update(self):
|
|
await super().async_update()
|
|
if not self._available:
|
|
return
|
|
if self._miio2miot:
|
|
self._state_attrs['props'] = self._miio2miot.miio_props_values
|
|
props = self.miio_props
|
|
adt = {}
|
|
if 'clean_area' in props:
|
|
adt['clean_area'] = round(props['clean_area'] / 1000000, 1)
|
|
if 'clean_time' in props:
|
|
adt['clean_time'] = round(props['clean_time'] / 60, 1)
|
|
if adt:
|
|
await self.async_update_attrs(adt)
|
|
self.device.dispatch(self.device.decode_attrs({'props': props}))
|
|
|
|
async def get_room_mapping(self):
|
|
if not self.miot_device:
|
|
return None
|
|
try:
|
|
rooms = await self.miot_device.async_send('get_room_mapping')
|
|
if rooms and rooms != 'unknown_method':
|
|
homes = await self.xiaomi_cloud.async_get_homerooms() if self.xiaomi_cloud else []
|
|
cloud_rooms = {}
|
|
for home in homes:
|
|
for room in home.get('roomlist', []):
|
|
cloud_rooms[room['id']] = room
|
|
for r in rooms:
|
|
room = cloud_rooms.get(r[1])
|
|
name = room['name'] if room else r[0]
|
|
if len(r) < 3:
|
|
r.append(name)
|
|
else:
|
|
r[2] = name
|
|
self._state_attrs['room_mapping'] = rooms
|
|
self.logger.info('Vacuum rooms: %s', rooms)
|
|
return rooms
|
|
self.logger.info('Vacuum rooms: %s', rooms)
|
|
except (DeviceException, Exception):
|
|
pass
|
|
return None
|
|
|
|
@property
|
|
def miio_props(self):
|
|
return self._state_attrs.get('props') or {}
|
|
|
|
async def async_pause(self):
|
|
"""Pause the cleaning task."""
|
|
if not self._act_pause:
|
|
return await self.async_miio_command('app_pause')
|
|
return await super().async_pause()
|
|
|
|
async def async_return_to_base(self, **kwargs):
|
|
if self.model in ['rockrobo.vacuum.v1']:
|
|
await self.async_stop()
|
|
return await super().async_return_to_base()
|
|
|
|
async def async_clean_spot(self, **kwargs):
|
|
"""Perform a spot clean-up."""
|
|
if self._miio2miot:
|
|
return await self.async_miio_command('app_spot')
|
|
return await super().async_clean_spot()
|
|
|
|
async def async_locate(self, **kwargs):
|
|
"""Locate the vacuum cleaner."""
|
|
if not self._act_locate:
|
|
return await self.async_miio_command('find_me', [''])
|
|
return await super().async_locate()
|
|
|
|
async def async_send_command(self, command, params=None, **kwargs):
|
|
"""Send a command to a vacuum cleaner."""
|
|
dvc = self.miot_device
|
|
if not dvc:
|
|
raise NotImplementedError()
|
|
return await self.async_miio_command(command, params)
|
|
|
|
async def async_start_clean_segment(self, segment, repeat=1, **kwargs):
|
|
segments = []
|
|
for r in self._state_attrs.get('room_mapping', []):
|
|
if segment in r:
|
|
segments.append(r[0])
|
|
break
|
|
if not segments:
|
|
await self.async_return_to_base()
|
|
return False
|
|
if self.state == VacuumActivity.CLEANING:
|
|
await self.async_pause()
|
|
await asyncio.sleep(1)
|
|
if self.model in ['roborock.vacuum.m1s']:
|
|
return await self.async_miio_command('app_segment_clean', segments)
|
|
return await self.async_miio_command('app_segment_clean', [{'segments': segments, 'repeat': repeat}])
|
|
|
|
|
|
class MiotViomiVacuumEntity(MiotVacuumEntity):
|
|
def __init__(self, config: dict, miot_service: MiotService):
|
|
super().__init__(config, miot_service)
|
|
self._supported_features |= VacuumEntityFeature.LOCATE
|
|
self._supported_features |= VacuumEntityFeature.SEND_COMMAND
|
|
self._miio_props = [
|
|
'run_state', 'mode', 'err_state', 'battary_life', 'box_type', 'mop_type', 's_time', 's_area',
|
|
'suction_grade', 'water_grade', 'remember_map', 'has_map', 'is_mop', 'has_newmap',
|
|
]
|
|
|
|
async def async_added_to_hass(self):
|
|
await super().async_added_to_hass()
|
|
if self._miio2miot:
|
|
self._miio2miot.extend_miio_props(self._miio_props)
|
|
|
|
async def async_update(self):
|
|
await super().async_update()
|
|
if not self._available:
|
|
return
|
|
props = self.device.props or {}
|
|
adt = {}
|
|
if 'miio.s_area' in props:
|
|
adt['clean_area'] = props['miio.s_area']
|
|
if 'miio.s_time' in props:
|
|
adt['clean_time'] = props['miio.s_time']
|
|
if adt:
|
|
await self.async_update_attrs(adt)
|
|
self.device.dispatch(self.device.decode_attrs(adt))
|
|
|
|
async def async_locate(self, **kwargs):
|
|
"""Locate the vacuum cleaner."""
|
|
if not self._act_locate:
|
|
return await self.async_miio_command('set_resetpos', [1])
|
|
return await super().async_locate()
|
|
|
|
async def async_send_command(self, command, params=None, **kwargs):
|
|
"""Send a command to a vacuum cleaner."""
|
|
dvc = self.miot_device
|
|
if not dvc:
|
|
raise NotImplementedError()
|
|
_LOGGER.debug('%s: Send command: %s %s', self.name_model, command, params)
|
|
if command == 'app_zoned_clean':
|
|
# params: [[x1, y2, x2, y1, repeats]]
|
|
rpt = 1
|
|
lst = []
|
|
for z in params or []:
|
|
rpt = z.pop(-1)
|
|
lst.append(z)
|
|
return await self.async_clean_zones(lst, rpt)
|
|
elif command == 'app_goto_target':
|
|
return await self.async_clean_point(params)
|
|
return await self.async_miio_command(command, params)
|
|
|
|
async def async_clean_zones(self, zones, repeats=1):
|
|
result = []
|
|
i = 0
|
|
for z in zones:
|
|
x1, y2, x2, y1 = z
|
|
res = '_'.join(str(x) for x in [i, 0, x1, y1, x1, y2, x2, y2, x2, y1])
|
|
for _ in range(repeats):
|
|
result.append(res)
|
|
i += 1
|
|
result = [i, *result]
|
|
await self.async_miio_command('set_uploadmap', [1])
|
|
await self.async_miio_command('set_zone', result)
|
|
return await self.async_miio_command('set_mode', [3, 1])
|
|
|
|
async def async_clean_point(self, point):
|
|
await self.async_miio_command('set_uploadmap', [0])
|
|
return await self.async_miio_command('set_pointclean', [1, *point])
|