Revert "Remove neato integration (#154902)"

.strict-typing

@@ -361,6 +361,7 @@ homeassistant.components.myuplink.*
 homeassistant.components.nam.*
 homeassistant.components.nanoleaf.*
 homeassistant.components.nasweb.*
+homeassistant.components.neato.*
 homeassistant.components.nest.*
 homeassistant.components.netatmo.*
 homeassistant.components.network.*

homeassistant/components/neato/__init__.py

@@ -0,0 +1,76 @@
+"""Support for Neato botvac connected vacuum cleaners."""
+
+import logging
+
+import aiohttp
+from pybotvac import Account
+from pybotvac.exceptions import NeatoException
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_TOKEN, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers import config_entry_oauth2_flow
+
+from . import api
+from .const import NEATO_DOMAIN, NEATO_LOGIN
+from .hub import NeatoHub
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS = [
+    Platform.BUTTON,
+    Platform.CAMERA,
+    Platform.SENSOR,
+    Platform.SWITCH,
+    Platform.VACUUM,
+]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up config entry."""
+    hass.data.setdefault(NEATO_DOMAIN, {})
+    if CONF_TOKEN not in entry.data:
+        raise ConfigEntryAuthFailed
+
+    implementation = (
+        await config_entry_oauth2_flow.async_get_config_entry_implementation(
+            hass, entry
+        )
+    )
+
+    session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
+    try:
+        await session.async_ensure_token_valid()
+    except aiohttp.ClientResponseError as ex:
+        _LOGGER.debug("API error: %s (%s)", ex.code, ex.message)
+        if ex.code in (401, 403):
+            raise ConfigEntryAuthFailed("Token not valid, trigger renewal") from ex
+        raise ConfigEntryNotReady from ex
+
+    neato_session = api.ConfigEntryAuth(hass, entry, implementation)
+    hass.data[NEATO_DOMAIN][entry.entry_id] = neato_session
+    hub = NeatoHub(hass, Account(neato_session))
+
+    await hub.async_update_entry_unique_id(entry)
+
+    try:
+        await hass.async_add_executor_job(hub.update_robots)
+    except NeatoException as ex:
+        _LOGGER.debug("Failed to connect to Neato API")
+        raise ConfigEntryNotReady from ex
+
+    hass.data[NEATO_LOGIN] = hub
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        hass.data[NEATO_DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/neato/api.py

@@ -0,0 +1,58 @@
+"""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
+
+from homeassistant import config_entries, core
+from homeassistant.components.application_credentials import AuthImplementation
+from homeassistant.helpers import config_entry_oauth2_flow
+
+
+class ConfigEntryAuth(pybotvac.OAuthSession):  # type: ignore[misc]
+    """Provide Neato Botvac authentication tied to an OAuth2 based config entry."""
+
+    def __init__(
+        self,
+        hass: core.HomeAssistant,
+        config_entry: config_entries.ConfigEntry,
+        implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
+    ) -> None:
+        """Initialize Neato Botvac Auth."""
+        self.hass = hass
+        self.session = config_entry_oauth2_flow.OAuth2Session(
+            hass, config_entry, implementation
+        )
+        super().__init__(self.session.token, vendor=pybotvac.Neato())
+
+    def refresh_tokens(self) -> str:
+        """Refresh and return new Neato Botvac tokens."""
+        run_coroutine_threadsafe(
+            self.session.async_ensure_token_valid(), self.hass.loop
+        ).result()
+
+        return self.session.token["access_token"]  # type: ignore[no-any-return]
+
+
+class NeatoImplementation(AuthImplementation):
+    """Neato implementation of LocalOAuth2Implementation.
+
+    We need this class because we have to add client_secret
+    and scope to the authorization request.
+    """
+
+    @property
+    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}
+
+    async def async_generate_authorize_url(self, flow_id: str) -> str:
+        """Generate a url for the user to authorize.
+
+        We must make sure that the plus signs are not encoded.
+        """
+        url = await super().async_generate_authorize_url(flow_id)
+        return f"{url}&scope=public_profile+control_robots+maps"

homeassistant/components/neato/application_credentials.py

@@ -0,0 +1,28 @@
+"""Application credentials platform for neato."""
+
+from pybotvac import Neato
+
+from homeassistant.components.application_credentials import (
+    AuthorizationServer,
+    ClientCredential,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_oauth2_flow
+
+from . import api
+
+
+async def async_get_auth_implementation(
+    hass: HomeAssistant, auth_domain: str, credential: ClientCredential
+) -> config_entry_oauth2_flow.AbstractOAuth2Implementation:
+    """Return auth implementation for a custom auth implementation."""
+    vendor = Neato()
+    return api.NeatoImplementation(
+        hass,
+        auth_domain,
+        credential,
+        AuthorizationServer(
+            authorize_url=vendor.auth_endpoint,
+            token_url=vendor.token_endpoint,
+        ),
+    )

homeassistant/components/neato/button.py

@@ -0,0 +1,44 @@
+"""Support for Neato buttons."""
+
+from __future__ import annotations
+
+from pybotvac import Robot
+
+from homeassistant.components.button import ButtonEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import EntityCategory
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import NEATO_ROBOTS
+from .entity import NeatoEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Neato button from config entry."""
+    entities = [NeatoDismissAlertButton(robot) for robot in hass.data[NEATO_ROBOTS]]
+
+    async_add_entities(entities, True)
+
+
+class NeatoDismissAlertButton(NeatoEntity, ButtonEntity):
+    """Representation of a dismiss_alert button entity."""
+
+    _attr_translation_key = "dismiss_alert"
+    _attr_entity_category = EntityCategory.CONFIG
+
+    def __init__(
+        self,
+        robot: Robot,
+    ) -> None:
+        """Initialize a dismiss_alert Neato button entity."""
+        super().__init__(robot)
+        self._attr_unique_id = f"{robot.serial}_dismiss_alert"
+
+    async def async_press(self) -> None:
+        """Press the button."""
+        await self.hass.async_add_executor_job(self.robot.dismiss_current_alert)

homeassistant/components/neato/camera.py

@@ -0,0 +1,130 @@
+"""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.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import NEATO_LOGIN, NEATO_MAP_DATA, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
+from .entity import NeatoEntity
+from .hub import NeatoHub
+
+_LOGGER = logging.getLogger(__name__)
+
+SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
+ATTR_GENERATED_AT = "generated_at"
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Neato camera with config entry."""
+    neato: NeatoHub = hass.data[NEATO_LOGIN]
+    mapdata: dict[str, Any] | None = hass.data.get(NEATO_MAP_DATA)
+    dev = [
+        NeatoCleaningMap(neato, robot, mapdata)
+        for robot in hass.data[NEATO_ROBOTS]
+        if "maps" in robot.traits
+    ]
+
+    if not dev:
+        return
+
+    _LOGGER.debug("Adding robots for cleaning maps %s", dev)
+    async_add_entities(dev, True)
+
+
+class NeatoCleaningMap(NeatoEntity, Camera):
+    """Neato cleaning map for last clean."""
+
+    _attr_translation_key = "cleaning_map"
+
+    def __init__(
+        self, neato: NeatoHub, robot: Robot, mapdata: dict[str, Any] | None
+    ) -> None:
+        """Initialize Neato cleaning map."""
+        super().__init__(robot)
+        Camera.__init__(self)
+        self.neato = neato
+        self._mapdata = mapdata
+        self._available = neato is not None
+        self._robot_serial: str = self.robot.serial
+        self._attr_unique_id = self.robot.serial
+        self._generated_at: str | None = None
+        self._image_url: str | None = None
+        self._image: bytes | None = None
+
+    def camera_image(
+        self, width: int | None = None, height: int | None = None
+    ) -> bytes | None:
+        """Return image response."""
+        self.update()
+        return self._image
+
+    def update(self) -> None:
+        """Check the contents of the map list."""
+
+        _LOGGER.debug("Running camera update for '%s'", self.entity_id)
+        try:
+            self.neato.update_robots()
+        except NeatoRobotException as ex:
+            if self._available:  # Print only once when available
+                _LOGGER.error(
+                    "Neato camera connection error for '%s': %s", self.entity_id, ex
+                )
+            self._image = None
+            self._image_url = None
+            self._available = False
+            return
+
+        if self._mapdata:
+            map_data: dict[str, Any] = self._mapdata[self._robot_serial]["maps"][0]
+        if (image_url := map_data["url"]) == self._image_url:
+            _LOGGER.debug(
+                "The map image_url for '%s' is the same as old", self.entity_id
+            )
+            return
+
+        try:
+            image: HTTPResponse = self.neato.download_map(image_url)
+        except NeatoRobotException as ex:
+            if self._available:  # Print only once when available
+                _LOGGER.error(
+                    "Neato camera connection error for '%s': %s", self.entity_id, ex
+                )
+            self._image = None
+            self._image_url = None
+            self._available = False
+            return
+
+        self._image = image.read()
+        self._image_url = image_url
+        self._generated_at = map_data.get("generated_at")
+        self._available = True
+
+    @property
+    def available(self) -> bool:
+        """Return if the robot is available."""
+        return self._available
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Return the state attributes of the vacuum cleaner."""
+        data: dict[str, Any] = {}
+
+        if self._generated_at is not None:
+            data[ATTR_GENERATED_AT] = self._generated_at
+
+        return data