Strict typing for Neato (#53633)
* Strict typing * Rebase * Tweak import * Cleanup * Rebase + typing hub * Flake8 * Update homeassistant/components/neato/config_flow.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update homeassistant/components/neato/vacuum.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update homeassistant/components/neato/camera.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Address review comments * Black * Update homeassistant/components/neato/config_flow.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Specific dict definition * Annotations Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Simone Chemelli
GitHub
parent
3da61b77a9
commit
18a0fcf931
10 changed files with 213 additions and 133 deletions
|
|
@ -63,6 +63,7 @@ homeassistant.components.mailbox.*
|
|||
homeassistant.components.media_player.*
|
||||
homeassistant.components.mysensors.*
|
||||
homeassistant.components.nam.*
|
||||
homeassistant.components.neato.*
|
||||
homeassistant.components.nest.*
|
||||
homeassistant.components.netatmo.*
|
||||
homeassistant.components.network.*
|
||||
|
|
|
|||
|
|
@ -1,5 +1,8 @@
|
|||
"""API for Neato Botvac bound to Home Assistant OAuth."""
|
||||
from __future__ import annotations
|
||||
|
||||
from asyncio import run_coroutine_threadsafe
|
||||
from typing import Any
|
||||
|
||||
import pybotvac
|
||||
|
||||
|
|
@ -7,7 +10,7 @@ from homeassistant import config_entries, core
|
|||
from homeassistant.helpers import config_entry_oauth2_flow
|
||||
|
||||
|
||||
class ConfigEntryAuth(pybotvac.OAuthSession):
|
||||
class ConfigEntryAuth(pybotvac.OAuthSession): # type: ignore[misc]
|
||||
"""Provide Neato Botvac authentication tied to an OAuth2 based config entry."""
|
||||
|
||||
def __init__(
|
||||
|
|
@ -29,7 +32,7 @@ class ConfigEntryAuth(pybotvac.OAuthSession):
|
|||
self.session.async_ensure_token_valid(), self.hass.loop
|
||||
).result()
|
||||
|
||||
return self.session.token["access_token"]
|
||||
return self.session.token["access_token"] # type: ignore[no-any-return]
|
||||
|
||||
|
||||
class NeatoImplementation(config_entry_oauth2_flow.LocalOAuth2Implementation):
|
||||
|
|
@ -39,7 +42,7 @@ class NeatoImplementation(config_entry_oauth2_flow.LocalOAuth2Implementation):
|
|||
"""
|
||||
|
||||
@property
|
||||
def extra_authorize_data(self) -> dict:
|
||||
def extra_authorize_data(self) -> dict[str, Any]:
|
||||
"""Extra data that needs to be appended to the authorize url."""
|
||||
return {"client_secret": self.client_secret}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,20 @@
|
|||
"""Support for loading picture from Neato."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pybotvac.exceptions import NeatoRobotException
|
||||
from pybotvac.robot import Robot
|
||||
from urllib3.response import HTTPResponse
|
||||
|
||||
from homeassistant.components.camera import Camera
|
||||
from homeassistant.components.neato import NeatoHub
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from .const import (
|
||||
NEATO_DOMAIN,
|
||||
|
|
@ -20,11 +30,13 @@ SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
|
|||
ATTR_GENERATED_AT = "generated_at"
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up Neato camera with config entry."""
|
||||
dev = []
|
||||
neato = hass.data.get(NEATO_LOGIN)
|
||||
mapdata = hass.data.get(NEATO_MAP_DATA)
|
||||
neato: NeatoHub = hass.data[NEATO_LOGIN]
|
||||
mapdata: dict[str, Any] | None = hass.data.get(NEATO_MAP_DATA)
|
||||
for robot in hass.data[NEATO_ROBOTS]:
|
||||
if "maps" in robot.traits:
|
||||
dev.append(NeatoCleaningMap(neato, robot, mapdata))
|
||||
|
|
@ -39,7 +51,9 @@ async def async_setup_entry(hass, entry, async_add_entities):
|
|||
class NeatoCleaningMap(Camera):
|
||||
"""Neato cleaning map for last clean."""
|
||||
|
||||
def __init__(self, neato, robot, mapdata):
|
||||
def __init__(
|
||||
self, neato: NeatoHub, robot: Robot, mapdata: dict[str, Any] | None
|
||||
) -> None:
|
||||
"""Initialize Neato cleaning map."""
|
||||
super().__init__()
|
||||
self.robot = robot
|
||||
|
|
@ -47,24 +61,18 @@ class NeatoCleaningMap(Camera):
|
|||
self._mapdata = mapdata
|
||||
self._available = neato is not None
|
||||
self._robot_name = f"{self.robot.name} Cleaning Map"
|
||||
self._robot_serial = self.robot.serial
|
||||
self._generated_at = None
|
||||
self._image_url = None
|
||||
self._image = None
|
||||
self._robot_serial: str = self.robot.serial
|
||||
self._generated_at: str | None = None
|
||||
self._image_url: str | None = None
|
||||
self._image: bytes | None = None
|
||||
|
||||
def camera_image(self):
|
||||
def camera_image(self) -> bytes | None:
|
||||
"""Return image response."""
|
||||
self.update()
|
||||
return self._image
|
||||
|
||||
def update(self):
|
||||
def update(self) -> None:
|
||||
"""Check the contents of the map list."""
|
||||
if self.neato is None:
|
||||
_LOGGER.error("Error while updating '%s'", self.entity_id)
|
||||
self._image = None
|
||||
self._image_url = None
|
||||
self._available = False
|
||||
return
|
||||
|
||||
_LOGGER.debug("Running camera update for '%s'", self.entity_id)
|
||||
try:
|
||||
|
|
@ -80,7 +88,8 @@ class NeatoCleaningMap(Camera):
|
|||
return
|
||||
|
||||
image_url = None
|
||||
map_data = self._mapdata[self._robot_serial]["maps"][0]
|
||||
if self._mapdata:
|
||||
map_data: dict[str, Any] = self._mapdata[self._robot_serial]["maps"][0]
|
||||
image_url = map_data["url"]
|
||||
if image_url == self._image_url:
|
||||
_LOGGER.debug(
|
||||
|
|
@ -89,7 +98,7 @@ class NeatoCleaningMap(Camera):
|
|||
return
|
||||
|
||||
try:
|
||||
image = self.neato.download_map(image_url)
|
||||
image: HTTPResponse = self.neato.download_map(image_url)
|
||||
except NeatoRobotException as ex:
|
||||
if self._available: # Print only once when available
|
||||
_LOGGER.error(
|
||||
|
|
@ -102,33 +111,33 @@ class NeatoCleaningMap(Camera):
|
|||
|
||||
self._image = image.read()
|
||||
self._image_url = image_url
|
||||
self._generated_at = map_data["generated_at"]
|
||||
self._generated_at = map_data.get("generated_at")
|
||||
self._available = True
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of this camera."""
|
||||
return self._robot_name
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
def unique_id(self) -> str:
|
||||
"""Return unique ID."""
|
||||
return self._robot_serial
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
def available(self) -> bool:
|
||||
"""Return if the robot is available."""
|
||||
return self._available
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Device info for neato robot."""
|
||||
return {"identifiers": {(NEATO_DOMAIN, self._robot_serial)}}
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
"""Return the state attributes of the vacuum cleaner."""
|
||||
data = {}
|
||||
data: dict[str, Any] = {}
|
||||
|
||||
if self._generated_at is not None:
|
||||
data[ATTR_GENERATED_AT] = self._generated_at
|
||||
|
|
|
|||
|
|
@ -2,10 +2,13 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from types import MappingProxyType
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import SOURCE_REAUTH
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers import config_entry_oauth2_flow
|
||||
|
||||
from .const import NEATO_DOMAIN
|
||||
|
|
@ -23,7 +26,9 @@ class OAuth2FlowHandler(
|
|||
"""Return logger."""
|
||||
return logging.getLogger(__name__)
|
||||
|
||||
async def async_step_user(self, user_input: dict | None = None) -> dict:
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Create an entry for the flow."""
|
||||
current_entries = self._async_current_entries()
|
||||
if self.source != SOURCE_REAUTH and current_entries:
|
||||
|
|
@ -32,11 +37,13 @@ class OAuth2FlowHandler(
|
|||
|
||||
return await super().async_step_user(user_input=user_input)
|
||||
|
||||
async def async_step_reauth(self, data) -> dict:
|
||||
async def async_step_reauth(self, data: MappingProxyType[str, Any]) -> FlowResult:
|
||||
"""Perform reauth upon migration of old entries."""
|
||||
return await self.async_step_reauth_confirm()
|
||||
|
||||
async def async_step_reauth_confirm(self, user_input: dict | None = None) -> dict:
|
||||
async def async_step_reauth_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Confirm reauth upon migration of old entries."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
|
|
@ -44,7 +51,7 @@ class OAuth2FlowHandler(
|
|||
)
|
||||
return await self.async_step_user()
|
||||
|
||||
async def async_oauth_create_entry(self, data: dict) -> dict:
|
||||
async def async_oauth_create_entry(self, data: dict[str, Any]) -> FlowResult:
|
||||
"""Create an entry for the flow. Update an entry if one already exist."""
|
||||
current_entries = self._async_current_entries()
|
||||
if self.source == SOURCE_REAUTH and current_entries:
|
||||
|
|
|
|||
|
|
@ -3,7 +3,9 @@ from datetime import timedelta
|
|||
import logging
|
||||
|
||||
from pybotvac import Account
|
||||
from urllib3.response import HTTPResponse
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.util import Throttle
|
||||
|
||||
|
|
@ -21,23 +23,23 @@ class NeatoHub:
|
|||
self.my_neato: Account = neato
|
||||
|
||||
@Throttle(timedelta(minutes=1))
|
||||
def update_robots(self):
|
||||
def update_robots(self) -> None:
|
||||
"""Update the robot states."""
|
||||
_LOGGER.debug("Running HUB.update_robots %s", self._hass.data.get(NEATO_ROBOTS))
|
||||
self._hass.data[NEATO_ROBOTS] = self.my_neato.robots
|
||||
self._hass.data[NEATO_PERSISTENT_MAPS] = self.my_neato.persistent_maps
|
||||
self._hass.data[NEATO_MAP_DATA] = self.my_neato.maps
|
||||
|
||||
def download_map(self, url):
|
||||
def download_map(self, url: str) -> HTTPResponse:
|
||||
"""Download a new map image."""
|
||||
map_image_data = self.my_neato.get_map_image(url)
|
||||
return map_image_data
|
||||
|
||||
async def async_update_entry_unique_id(self, entry) -> str:
|
||||
async def async_update_entry_unique_id(self, entry: ConfigEntry) -> str:
|
||||
"""Update entry for unique_id."""
|
||||
|
||||
await self._hass.async_add_executor_job(self.my_neato.refresh_userdata)
|
||||
unique_id = self.my_neato.unique_id
|
||||
unique_id: str = self.my_neato.unique_id
|
||||
|
||||
if entry.unique_id == unique_id:
|
||||
return unique_id
|
||||
|
|
|
|||
|
|
@ -1,11 +1,20 @@
|
|||
"""Support for Neato sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pybotvac.exceptions import NeatoRobotException
|
||||
from pybotvac.robot import Robot
|
||||
|
||||
from homeassistant.components.neato import NeatoHub
|
||||
from homeassistant.components.sensor import DEVICE_CLASS_BATTERY, SensorEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import PERCENTAGE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from .const import NEATO_DOMAIN, NEATO_LOGIN, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
|
||||
|
||||
|
|
@ -16,10 +25,12 @@ SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
|
|||
BATTERY = "Battery"
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up the Neato sensor using config entry."""
|
||||
dev = []
|
||||
neato = hass.data.get(NEATO_LOGIN)
|
||||
neato: NeatoHub = hass.data[NEATO_LOGIN]
|
||||
for robot in hass.data[NEATO_ROBOTS]:
|
||||
dev.append(NeatoSensor(neato, robot))
|
||||
|
||||
|
|
@ -33,15 +44,15 @@ async def async_setup_entry(hass, entry, async_add_entities):
|
|||
class NeatoSensor(SensorEntity):
|
||||
"""Neato sensor."""
|
||||
|
||||
def __init__(self, neato, robot):
|
||||
def __init__(self, neato: NeatoHub, robot: Robot) -> None:
|
||||
"""Initialize Neato sensor."""
|
||||
self.robot = robot
|
||||
self._available = False
|
||||
self._robot_name = f"{self.robot.name} {BATTERY}"
|
||||
self._robot_serial = self.robot.serial
|
||||
self._state = None
|
||||
self._available: bool = False
|
||||
self._robot_name: str = f"{self.robot.name} {BATTERY}"
|
||||
self._robot_serial: str = self.robot.serial
|
||||
self._state: dict[str, Any] | None = None
|
||||
|
||||
def update(self):
|
||||
def update(self) -> None:
|
||||
"""Update Neato Sensor."""
|
||||
try:
|
||||
self._state = self.robot.state
|
||||
|
|
@ -58,36 +69,38 @@ class NeatoSensor(SensorEntity):
|
|||
_LOGGER.debug("self._state=%s", self._state)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of this sensor."""
|
||||
return self._robot_name
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
def unique_id(self) -> str:
|
||||
"""Return unique ID."""
|
||||
return self._robot_serial
|
||||
|
||||
@property
|
||||
def device_class(self):
|
||||
def device_class(self) -> str:
|
||||
"""Return the device class."""
|
||||
return DEVICE_CLASS_BATTERY
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
def available(self) -> bool:
|
||||
"""Return availability."""
|
||||
return self._available
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
def state(self) -> str | None:
|
||||
"""Return the state."""
|
||||
return self._state["details"]["charge"] if self._state else None
|
||||
if self._state is not None:
|
||||
return str(self._state["details"]["charge"])
|
||||
return None
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
def unit_of_measurement(self) -> str:
|
||||
"""Return unit of measurement."""
|
||||
return PERCENTAGE
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Device info for neato robot."""
|
||||
return {"identifiers": {(NEATO_DOMAIN, self._robot_serial)}}
|
||||
|
|
|
|||
|
|
@ -1,11 +1,19 @@
|
|||
"""Support for Neato Connected Vacuums switches."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pybotvac.exceptions import NeatoRobotException
|
||||
from pybotvac.robot import Robot
|
||||
|
||||
from homeassistant.components.neato import NeatoHub
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import STATE_OFF, STATE_ON
|
||||
from homeassistant.helpers.entity import ToggleEntity
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity import DeviceInfo, ToggleEntity
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from .const import NEATO_DOMAIN, NEATO_LOGIN, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
|
||||
|
||||
|
|
@ -18,10 +26,13 @@ SWITCH_TYPE_SCHEDULE = "schedule"
|
|||
SWITCH_TYPES = {SWITCH_TYPE_SCHEDULE: ["Schedule"]}
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up Neato switch with config entry."""
|
||||
dev = []
|
||||
neato = hass.data.get(NEATO_LOGIN)
|
||||
neato: NeatoHub = hass.data[NEATO_LOGIN]
|
||||
|
||||
for robot in hass.data[NEATO_ROBOTS]:
|
||||
for type_name in SWITCH_TYPES:
|
||||
dev.append(NeatoConnectedSwitch(neato, robot, type_name))
|
||||
|
|
@ -36,18 +47,18 @@ async def async_setup_entry(hass, entry, async_add_entities):
|
|||
class NeatoConnectedSwitch(ToggleEntity):
|
||||
"""Neato Connected Switches."""
|
||||
|
||||
def __init__(self, neato, robot, switch_type):
|
||||
def __init__(self, neato: NeatoHub, robot: Robot, switch_type: str) -> None:
|
||||
"""Initialize the Neato 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._state: dict[str, Any] | None = None
|
||||
self._schedule_state: str | None = None
|
||||
self._clean_state = None
|
||||
self._robot_serial = self.robot.serial
|
||||
self._robot_serial: str = self.robot.serial
|
||||
|
||||
def update(self):
|
||||
def update(self) -> None:
|
||||
"""Update the states of Neato switches."""
|
||||
_LOGGER.debug("Running Neato switch update for '%s'", self.entity_id)
|
||||
try:
|
||||
|
|
@ -65,7 +76,7 @@ class NeatoConnectedSwitch(ToggleEntity):
|
|||
_LOGGER.debug("self._state=%s", self._state)
|
||||
if self.type == SWITCH_TYPE_SCHEDULE:
|
||||
_LOGGER.debug("State: %s", self._state)
|
||||
if self._state["details"]["isScheduleEnabled"]:
|
||||
if self._state is not None and self._state["details"]["isScheduleEnabled"]:
|
||||
self._schedule_state = STATE_ON
|
||||
else:
|
||||
self._schedule_state = STATE_OFF
|
||||
|
|
@ -74,34 +85,33 @@ class NeatoConnectedSwitch(ToggleEntity):
|
|||
)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of the switch."""
|
||||
return self._robot_name
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
def available(self) -> bool:
|
||||
"""Return True if entity is available."""
|
||||
return self._available
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self._robot_serial
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if switch is on."""
|
||||
if self.type == SWITCH_TYPE_SCHEDULE:
|
||||
if self._schedule_state == STATE_ON:
|
||||
return True
|
||||
return False
|
||||
return bool(
|
||||
self.type == SWITCH_TYPE_SCHEDULE and self._schedule_state == STATE_ON
|
||||
)
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Device info for neato robot."""
|
||||
return {"identifiers": {(NEATO_DOMAIN, self._robot_serial)}}
|
||||
|
||||
def turn_on(self, **kwargs):
|
||||
def turn_on(self, **kwargs: Any) -> None:
|
||||
"""Turn the switch on."""
|
||||
if self.type == SWITCH_TYPE_SCHEDULE:
|
||||
try:
|
||||
|
|
@ -111,7 +121,7 @@ class NeatoConnectedSwitch(ToggleEntity):
|
|||
"Neato switch connection error '%s': %s", self.entity_id, ex
|
||||
)
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
def turn_off(self, **kwargs: Any) -> None:
|
||||
"""Turn the switch off."""
|
||||
if self.type == SWITCH_TYPE_SCHEDULE:
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -1,7 +1,11 @@
|
|||
"""Support for Neato Connected Vacuums."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pybotvac import Robot
|
||||
from pybotvac.exceptions import NeatoRobotException
|
||||
import voluptuous as vol
|
||||
|
||||
|
|
@ -24,9 +28,14 @@ from homeassistant.components.vacuum import (
|
|||
SUPPORT_STOP,
|
||||
StateVacuumEntity,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ATTR_MODE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import config_validation as cv, entity_platform
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import NeatoHub
|
||||
from .const import (
|
||||
ACTION,
|
||||
ALERTS,
|
||||
|
|
@ -72,12 +81,14 @@ ATTR_CATEGORY = "category"
|
|||
ATTR_ZONE = "zone"
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up Neato vacuum with config entry."""
|
||||
dev = []
|
||||
neato = hass.data.get(NEATO_LOGIN)
|
||||
mapdata = hass.data.get(NEATO_MAP_DATA)
|
||||
persistent_maps = hass.data.get(NEATO_PERSISTENT_MAPS)
|
||||
neato: NeatoHub = hass.data[NEATO_LOGIN]
|
||||
mapdata: dict[str, Any] | None = hass.data.get(NEATO_MAP_DATA)
|
||||
persistent_maps: dict[str, Any] | None = hass.data.get(NEATO_PERSISTENT_MAPS)
|
||||
for robot in hass.data[NEATO_ROBOTS]:
|
||||
dev.append(NeatoConnectedVacuum(neato, robot, mapdata, persistent_maps))
|
||||
|
||||
|
|
@ -105,33 +116,39 @@ async def async_setup_entry(hass, entry, async_add_entities):
|
|||
class NeatoConnectedVacuum(StateVacuumEntity):
|
||||
"""Representation of a Neato Connected Vacuum."""
|
||||
|
||||
def __init__(self, neato, robot, mapdata, persistent_maps):
|
||||
def __init__(
|
||||
self,
|
||||
neato: NeatoHub,
|
||||
robot: Robot,
|
||||
mapdata: dict[str, Any] | None,
|
||||
persistent_maps: dict[str, Any] | None,
|
||||
) -> None:
|
||||
"""Initialize the Neato Connected Vacuum."""
|
||||
self.robot = robot
|
||||
self._available = neato is not None
|
||||
self._available: bool = neato is not None
|
||||
self._mapdata = mapdata
|
||||
self._name = f"{self.robot.name}"
|
||||
self._robot_has_map = self.robot.has_persistent_maps
|
||||
self._name: str = f"{self.robot.name}"
|
||||
self._robot_has_map: bool = self.robot.has_persistent_maps
|
||||
self._robot_maps = persistent_maps
|
||||
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
|
||||
self._robot_serial: str = self.robot.serial
|
||||
self._status_state: str | None = None
|
||||
self._clean_state: str | None = None
|
||||
self._state: dict[str, Any] | None = None
|
||||
self._clean_time_start: str | None = None
|
||||
self._clean_time_stop: str | None = None
|
||||
self._clean_area: float | None = None
|
||||
self._clean_battery_start: int | None = None
|
||||
self._clean_battery_end: int | None = None
|
||||
self._clean_susp_charge_count: int | None = None
|
||||
self._clean_susp_time: int | None = None
|
||||
self._clean_pause_time: int | None = None
|
||||
self._clean_error_time: int | None = None
|
||||
self._launched_from: str | None = None
|
||||
self._battery_level: int | None = None
|
||||
self._robot_boundaries: list = []
|
||||
self._robot_stats: dict[str, Any] | None = None
|
||||
|
||||
def update(self):
|
||||
def update(self) -> None:
|
||||
"""Update the states of Neato Vacuums."""
|
||||
_LOGGER.debug("Running Neato Vacuums update for '%s'", self.entity_id)
|
||||
try:
|
||||
|
|
@ -151,6 +168,8 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
self._available = False
|
||||
return
|
||||
|
||||
if self._state is None:
|
||||
return
|
||||
self._available = True
|
||||
_LOGGER.debug("self._state=%s", self._state)
|
||||
if "alert" in self._state:
|
||||
|
|
@ -198,10 +217,12 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
|
||||
self._battery_level = self._state["details"]["charge"]
|
||||
|
||||
if not self._mapdata.get(self._robot_serial, {}).get("maps", []):
|
||||
if self._mapdata is None or not self._mapdata.get(self._robot_serial, {}).get(
|
||||
"maps", []
|
||||
):
|
||||
return
|
||||
|
||||
mapdata = self._mapdata[self._robot_serial]["maps"][0]
|
||||
mapdata: dict[str, Any] = self._mapdata[self._robot_serial]["maps"][0]
|
||||
self._clean_time_start = mapdata["start_at"]
|
||||
self._clean_time_stop = mapdata["end_at"]
|
||||
self._clean_area = mapdata["cleaned_area"]
|
||||
|
|
@ -215,10 +236,11 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
|
||||
if (
|
||||
self._robot_has_map
|
||||
and self._state
|
||||
and self._state["availableServices"]["maps"] != "basic-1"
|
||||
and self._robot_maps[self._robot_serial]
|
||||
and self._robot_maps
|
||||
):
|
||||
allmaps = self._robot_maps[self._robot_serial]
|
||||
allmaps: dict = self._robot_maps[self._robot_serial]
|
||||
_LOGGER.debug(
|
||||
"Found the following maps for '%s': %s", self.entity_id, allmaps
|
||||
)
|
||||
|
|
@ -249,44 +271,44 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of the device."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def supported_features(self):
|
||||
def supported_features(self) -> int:
|
||||
"""Flag vacuum cleaner robot features that are supported."""
|
||||
return SUPPORT_NEATO
|
||||
|
||||
@property
|
||||
def battery_level(self):
|
||||
def battery_level(self) -> int | None:
|
||||
"""Return the battery level of the vacuum cleaner."""
|
||||
return self._battery_level
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
def available(self) -> bool:
|
||||
"""Return if the robot is available."""
|
||||
return self._available
|
||||
|
||||
@property
|
||||
def icon(self):
|
||||
def icon(self) -> str:
|
||||
"""Return neato specific icon."""
|
||||
return "mdi:robot-vacuum-variant"
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
def state(self) -> str | None:
|
||||
"""Return the status of the vacuum cleaner."""
|
||||
return self._clean_state
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self._robot_serial
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self):
|
||||
def extra_state_attributes(self) -> dict[str, Any]:
|
||||
"""Return the state attributes of the vacuum cleaner."""
|
||||
data = {}
|
||||
data: dict[str, Any] = {}
|
||||
|
||||
if self._status_state is not None:
|
||||
data[ATTR_STATUS] = self._status_state
|
||||
|
|
@ -314,28 +336,32 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
return data
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Device info for neato robot."""
|
||||
info = {"identifiers": {(NEATO_DOMAIN, self._robot_serial)}, "name": self._name}
|
||||
info: DeviceInfo = {
|
||||
"identifiers": {(NEATO_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):
|
||||
def start(self) -> None:
|
||||
"""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(
|
||||
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
|
||||
)
|
||||
if self._state:
|
||||
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(
|
||||
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
|
||||
)
|
||||
|
||||
def pause(self):
|
||||
def pause(self) -> None:
|
||||
"""Pause the vacuum."""
|
||||
try:
|
||||
self.robot.pause_cleaning()
|
||||
|
|
@ -344,7 +370,7 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
|
||||
)
|
||||
|
||||
def return_to_base(self, **kwargs):
|
||||
def return_to_base(self, **kwargs: Any) -> None:
|
||||
"""Set the vacuum cleaner to return to the dock."""
|
||||
try:
|
||||
if self._clean_state == STATE_CLEANING:
|
||||
|
|
@ -356,7 +382,7 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
|
||||
)
|
||||
|
||||
def stop(self, **kwargs):
|
||||
def stop(self, **kwargs: Any) -> None:
|
||||
"""Stop the vacuum cleaner."""
|
||||
try:
|
||||
self.robot.stop_cleaning()
|
||||
|
|
@ -365,7 +391,7 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
|
||||
)
|
||||
|
||||
def locate(self, **kwargs):
|
||||
def locate(self, **kwargs: Any) -> None:
|
||||
"""Locate the robot by making it emit a sound."""
|
||||
try:
|
||||
self.robot.locate()
|
||||
|
|
@ -374,7 +400,7 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
|
||||
)
|
||||
|
||||
def clean_spot(self, **kwargs):
|
||||
def clean_spot(self, **kwargs: Any) -> None:
|
||||
"""Run a spot cleaning starting from the base."""
|
||||
try:
|
||||
self.robot.start_spot_cleaning()
|
||||
|
|
@ -383,7 +409,9 @@ class NeatoConnectedVacuum(StateVacuumEntity):
|
|||
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
|
||||
)
|
||||
|
||||
def neato_custom_cleaning(self, mode, navigation, category, zone=None):
|
||||
def neato_custom_cleaning(
|
||||
self, mode: str, navigation: str, category: str, zone: str | None = None
|
||||
) -> None:
|
||||
"""Zone cleaning service call."""
|
||||
boundary_id = None
|
||||
if zone is not None:
|
||||
|
|
|
|||
14
mypy.ini
14
mypy.ini
|
|
@ -704,6 +704,17 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.neato.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.nest.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
|
@ -1515,9 +1526,6 @@ ignore_errors = true
|
|||
[mypy-homeassistant.components.mullvad.*]
|
||||
ignore_errors = true
|
||||
|
||||
[mypy-homeassistant.components.neato.*]
|
||||
ignore_errors = true
|
||||
|
||||
[mypy-homeassistant.components.ness_alarm.*]
|
||||
ignore_errors = true
|
||||
|
||||
|
|
|
|||
|
|
@ -101,7 +101,6 @@ IGNORED_MODULES: Final[list[str]] = [
|
|||
"homeassistant.components.mobile_app.*",
|
||||
"homeassistant.components.motion_blinds.*",
|
||||
"homeassistant.components.mullvad.*",
|
||||
"homeassistant.components.neato.*",
|
||||
"homeassistant.components.ness_alarm.*",
|
||||
"homeassistant.components.nest.legacy.*",
|
||||
"homeassistant.components.netio.*",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue