entity.py

Go to the documentation of this file.

"""Base class for iRobot devices."""
 
from __future__ import annotations
 
import asyncio
import logging
 
from homeassistant.components.vacuum import (
    ATTR_STATUS,
    STATE_CLEANING,
    STATE_DOCKED,
    STATE_ERROR,
    STATE_RETURNING,
    StateVacuumEntity,
    VacuumEntityFeature,
)
from homeassistant.const import ATTR_CONNECTIONS, STATE_IDLE, STATE_PAUSED
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
import homeassistant.util.dt as dt_util
from homeassistant.util.unit_system import METRIC_SYSTEM
 
from . import roomba_reported_state
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
ATTR_CLEANING_TIME = "cleaning_time"
ATTR_CLEANED_AREA = "cleaned_area"
ATTR_ERROR = "error"
ATTR_ERROR_CODE = "error_code"
ATTR_POSITION = "position"
ATTR_SOFTWARE_VERSION = "software_version"
 
# Commonly supported features
SUPPORT_IROBOT = (
    VacuumEntityFeature.BATTERY
    | VacuumEntityFeature.PAUSE
    | VacuumEntityFeature.RETURN_HOME
    | VacuumEntityFeature.SEND_COMMAND
    | VacuumEntityFeature.START
    | VacuumEntityFeature.STATE
    | VacuumEntityFeature.STOP
    | VacuumEntityFeature.LOCATE
)
 
STATE_MAP = {
    "": STATE_IDLE,
    "charge": STATE_DOCKED,
    "evac": STATE_RETURNING,  # Emptying at cleanbase
    "hmMidMsn": STATE_CLEANING,  # Recharging at the middle of a cycle
    "hmPostMsn": STATE_RETURNING,  # Cycle finished
    "hmUsrDock": STATE_RETURNING,
    "pause": STATE_PAUSED,
    "run": STATE_CLEANING,
    "stop": STATE_IDLE,
    "stuck": STATE_ERROR,
}
 
 
class IRobotEntity(Entity):
    """Base class for iRobot Entities."""
 
    _attr_should_poll = False
    _attr_has_entity_name = True
 
    def __init__(self, roomba, blid):
        """Initialize the iRobot handler."""
        self.vacuumvacuum = roomba
        self._blid_blid = blid
        self.vacuum_statevacuum_state = roomba_reported_state(roomba)
 
        self._attr_device_info_attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, self.robot_unique_idrobot_unique_id)},
            serial_number=self.vacuum_statevacuum_state.get("hwPartsRev", {}).get("navSerialNo"),
            manufacturer="iRobot",
            model=self.vacuum_statevacuum_state.get("sku"),
            name=str(self.vacuum_statevacuum_state.get("name")),
            sw_version=self.vacuum_statevacuum_state.get("softwareVer"),
            hw_version=self.vacuum_statevacuum_state.get("hardwareRev"),
        )
 
        if mac_address := self.vacuum_statevacuum_state.get("hwPartsRev", {}).get(
            "wlan0HwAddr", self.vacuum_statevacuum_state.get("mac")
        ):
            self._attr_device_info_attr_device_info[ATTR_CONNECTIONS] = {
                (dr.CONNECTION_NETWORK_MAC, mac_address)
            }
 
    @property
    def robot_unique_id(self):
        """Return the uniqueid of the vacuum cleaner."""
        return f"roomba_{self._blid}"
 
    @property
    def unique_id(self):
        """Return the uniqueid of the vacuum cleaner."""
        return self.robot_unique_idrobot_unique_id
 
    @property
    def battery_level(self):
        """Return the battery level of the vacuum cleaner."""
        return self.vacuum_statevacuum_state.get("batPct")
 
    @property
    def run_stats(self):
        """Return the run stats."""
        return self.vacuum_statevacuum_state.get("bbrun", {})
 
    @property
    def mission_stats(self):
        """Return the mission stats."""
        return self.vacuum_statevacuum_state.get("bbmssn", {})
 
    @property
    def battery_stats(self):
        """Return the battery stats."""
        return self.vacuum_statevacuum_state.get("bbchg3", {})
 
    @property
    def last_mission(self):
        """Return last mission start time."""
        if (
            ts := self.vacuum_statevacuum_state.get("cleanMissionStatus", {}).get("mssnStrtTm")
        ) is None or ts == 0:
            return None
        return dt_util.utc_from_timestamp(ts)
 
    @property
    def _robot_state(self):
        """Return the state of the vacuum cleaner."""
        clean_mission_status = self.vacuum_statevacuum_state.get("cleanMissionStatus", {})
        cycle = clean_mission_status.get("cycle")
        phase = clean_mission_status.get("phase")
        try:
            state = STATE_MAP[phase]
        except KeyError:
            return STATE_ERROR
        if cycle != "none" and state in (STATE_IDLE, STATE_DOCKED):
            state = STATE_PAUSED
        return state
 
    async def async_added_to_hass(self):
        """Register callback function."""
        self.vacuumvacuum.register_on_message_callback(self.on_messageon_message)
 
    def new_state_filter(self, new_state):
        """Filter out wifi state messages."""
        return len(new_state) > 1 or "signal" not in new_state
 
    def on_message(self, json_data):
        """Update state on message change."""
        state = json_data.get("state", {}).get("reported", {})
        if self.new_state_filternew_state_filter(state):
            self.schedule_update_ha_stateschedule_update_ha_state()
 
 
class IRobotVacuum(IRobotEntity, StateVacuumEntity):  # pylint: disable=hass-enforce-class-module
    """Base class for iRobot robots."""
 
    _attr_name = None
    _attr_supported_features = SUPPORT_IROBOT
    _attr_available = True  # Always available, otherwise setup will fail
 
    def __init__(self, roomba, blid):
        """Initialize the iRobot handler."""
        super().__init__(roomba, blid)
        self._cap_position_cap_position = self.vacuum_statevacuum_state.get("cap", {}).get("pose") == 1
 
    @property
    def state(self):
        """Return the state of the vacuum cleaner."""
        return self._robot_state_robot_state
 
    @property
    def extra_state_attributes(self):
        """Return the state attributes of the device."""
        state = self.vacuum_statevacuum_state
 
        # Roomba software version
        software_version = state.get("softwareVer")
 
        # Set properties that are to appear in the GUI
        state_attrs = {ATTR_SOFTWARE_VERSION: software_version}
 
        # Set legacy status to avoid break changes
        state_attrs[ATTR_STATUS] = self.vacuumvacuum.current_state
 
        # Only add cleaning time and cleaned area attrs when the vacuum is
        # currently on
        if self.statestatestatestatestatestate == STATE_CLEANING:
            # Get clean mission status
            (
                state_attrs[ATTR_CLEANING_TIME],
                state_attrs[ATTR_CLEANED_AREA],
            ) = self.get_cleaning_statusget_cleaning_status(state)
 
        # Error
        if self.vacuumvacuum.error_code != 0:
            state_attrs[ATTR_ERROR] = self.vacuumvacuum.error_message
            state_attrs[ATTR_ERROR_CODE] = self.vacuumvacuum.error_code
 
        # Not all Roombas expose position data
        # https://github.com/koalazak/dorita980/issues/48
        if self._cap_position_cap_position:
            pos_state = state.get("pose", {})
            position = None
            pos_x = pos_state.get("point", {}).get("x")
            pos_y = pos_state.get("point", {}).get("y")
            theta = pos_state.get("theta")
            if all(item is not None for item in (pos_x, pos_y, theta)):
                position = f"({pos_x}, {pos_y}, {theta})"
            state_attrs[ATTR_POSITION] = position
 
        return state_attrs
 
    def get_cleaning_status(self, state) -> tuple[int, int]:
        """Return the cleaning time and cleaned area from the device."""
        if not (mission_state := state.get("cleanMissionStatus")):
            return (0, 0)
 
        if cleaning_time := mission_state.get("mssnM", 0):
            pass
        elif start_time := mission_state.get("mssnStrtTm"):
            now = dt_util.as_timestamp(dt_util.utcnow())
            if now > start_time:
                cleaning_time = (now - start_time) // 60
 
        if cleaned_area := mission_state.get("sqft", 0):  # Imperial
            # Convert to m2 if the unit_system is set to metric
            if self.hasshass.config.units is METRIC_SYSTEM:
                cleaned_area = round(cleaned_area * 0.0929)
 
        return (cleaning_time, cleaned_area)
 
    def on_message(self, json_data):
        """Update state on message change."""
        state = json_data.get("state", {}).get("reported", {})
        if self.new_state_filternew_state_filter(state):
            _LOGGER.debug("Got new state from the vacuum: %s", json_data)
            self.schedule_update_ha_stateschedule_update_ha_state()
 
    async def async_start(self):
        """Start or resume the cleaning task."""
        if self.statestatestatestatestatestate == STATE_PAUSED:
            await self.hasshass.async_add_executor_job(self.vacuumvacuum.send_command, "resume")
        else:
            await self.hasshass.async_add_executor_job(self.vacuumvacuum.send_command, "start")
 
    async def async_stop(self, **kwargs):
        """Stop the vacuum cleaner."""
        await self.hasshass.async_add_executor_job(self.vacuumvacuum.send_command, "stop")
 
    async def async_pause(self):
        """Pause the cleaning cycle."""
        await self.hasshass.async_add_executor_job(self.vacuumvacuum.send_command, "pause")
 
    async def async_return_to_base(self, **kwargs):
        """Set the vacuum cleaner to return to the dock."""
        if self.statestatestatestatestatestate == STATE_CLEANING:
            await self.async_pauseasync_pauseasync_pause()
            for _ in range(10):
                if self.statestatestatestatestatestate == STATE_PAUSED:
                    break
                await asyncio.sleep(1)
        await self.hasshass.async_add_executor_job(self.vacuumvacuum.send_command, "dock")
 
    async def async_locate(self, **kwargs):
        """Located vacuum."""
        await self.hasshass.async_add_executor_job(self.vacuumvacuum.send_command, "find")
 
    async def async_send_command(self, command, params=None, **kwargs):
        """Send raw command."""
        _LOGGER.debug("async_send_command %s (%s), %s", command, params, kwargs)
        await self.hasshass.async_add_executor_job(
            self.vacuumvacuum.send_command, command, params
        )