Add romy vacuum integration (home-assistant#93750)

Co-authored-by: Erik Montnemery <[email protected]>
Co-authored-by: Robert Resch <[email protected]>
Co-authored-by: Allen Porter <[email protected]>

.coveragerc

@@ -1103,6 +1103,9 @@ omit =
     homeassistant/components/ripple/sensor.py
     homeassistant/components/roborock/coordinator.py
     homeassistant/components/rocketchat/notify.py
+    homeassistant/components/romy/__init__.py
+    homeassistant/components/romy/coordinator.py
+    homeassistant/components/romy/vacuum.py
     homeassistant/components/roomba/__init__.py
     homeassistant/components/roomba/binary_sensor.py
     homeassistant/components/roomba/braava.py

.strict-typing

@@ -361,6 +361,7 @@ homeassistant.components.rhasspy.*
 homeassistant.components.ridwell.*
 homeassistant.components.rituals_perfume_genie.*
 homeassistant.components.roku.*
+homeassistant.components.romy.*
 homeassistant.components.rpi_power.*
 homeassistant.components.rss_feed_template.*
 homeassistant.components.rtsp_to_webrtc.*

CODEOWNERS

@@ -1122,6 +1122,8 @@ build.json @home-assistant/supervisor
 /tests/components/roborock/ @humbertogontijo @Lash-L
 /homeassistant/components/roku/ @ctalkington
 /tests/components/roku/ @ctalkington
+/homeassistant/components/romy/ @xeniter
+/tests/components/romy/ @xeniter
 /homeassistant/components/roomba/ @pschmitt @cyr-ius @shenxn @Xitee1
 /tests/components/roomba/ @pschmitt @cyr-ius @shenxn @Xitee1
 /homeassistant/components/roon/ @pavoni

homeassistant/components/romy/__init__.py

@@ -0,0 +1,42 @@
+"""ROMY Integration."""
+
+import romy
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN, LOGGER, PLATFORMS
+from .coordinator import RomyVacuumCoordinator
+
+
+async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
+    """Initialize the ROMY platform via config entry."""
+
+    new_romy = await romy.create_romy(
+        config_entry.data[CONF_HOST], config_entry.data.get(CONF_PASSWORD, "")
+    )
+
+    coordinator = RomyVacuumCoordinator(hass, new_romy)
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[config_entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
+
+    config_entry.async_on_unload(config_entry.add_update_listener(update_listener))
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Handle removal of an entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+    return unload_ok
+
+
+async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+    """Handle options update."""
+    LOGGER.debug("update_listener")
+    await hass.config_entries.async_reload(config_entry.entry_id)

homeassistant/components/romy/config_flow.py

@@ -0,0 +1,148 @@
+"""Config flow for ROMY integration."""
+from __future__ import annotations
+
+import romy
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components import zeroconf
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.data_entry_flow import FlowResult
+import homeassistant.helpers.config_validation as cv
+
+from .const import DOMAIN, LOGGER
+
+
+class RomyConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle config flow for ROMY."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Handle a config flow for ROMY."""
+        self.host: str = ""
+        self.password: str = ""
+        self.robot_name_given_by_user: str = ""
+
+    async def async_step_user(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Handle the user step."""
+        errors: dict[str, str] = {}
+
+        if user_input:
+            self.host = user_input[CONF_HOST]
+
+            new_romy = await romy.create_romy(self.host, "")
+
+            if not new_romy.is_initialized:
+                errors[CONF_HOST] = "cannot_connect"
+            else:
+                await self.async_set_unique_id(new_romy.unique_id)
+                self._abort_if_unique_id_configured()
+
+                self.robot_name_given_by_user = new_romy.user_name
+
+                if not new_romy.is_unlocked:
+                    return await self.async_step_password()
+                return await self._async_step_finish_config()
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_HOST): cv.string,
+                },
+            ),
+            errors=errors,
+        )
+
+    async def async_step_password(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Unlock the robots local http interface with password."""
+        errors: dict[str, str] = {}
+
+        if user_input:
+            self.password = user_input[CONF_PASSWORD]
+            new_romy = await romy.create_romy(self.host, self.password)
+
+            if not new_romy.is_initialized:
+                errors[CONF_PASSWORD] = "cannot_connect"
+            elif not new_romy.is_unlocked:
+                errors[CONF_PASSWORD] = "invalid_auth"
+
+            if not errors:
+                return await self._async_step_finish_config()
+
+        return self.async_show_form(
+            step_id="password",
+            data_schema=vol.Schema(
+                {vol.Required(CONF_PASSWORD): vol.All(cv.string, vol.Length(8))},
+            ),
+            errors=errors,
+        )
+
+    async def async_step_zeroconf(
+        self, discovery_info: zeroconf.ZeroconfServiceInfo
+    ) -> FlowResult:
+        """Handle zeroconf discovery."""
+
+        LOGGER.debug("Zeroconf discovery_info: %s", discovery_info)
+
+        # connect and gather information from your ROMY
+        self.host = discovery_info.host
+        LOGGER.debug("ZeroConf Host: %s", self.host)
+
+        new_discovered_romy = await romy.create_romy(self.host, "")
+
+        self.robot_name_given_by_user = new_discovered_romy.user_name
+        LOGGER.debug("ZeroConf Name: %s", self.robot_name_given_by_user)
+
+        # get unique id and stop discovery if robot is already added
+        unique_id = new_discovered_romy.unique_id
+        LOGGER.debug("ZeroConf Unique_id: %s", unique_id)
+        await self.async_set_unique_id(unique_id)
+        self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})
+
+        self.context.update(
+            {
+                "title_placeholders": {
+                    "name": f"{self.robot_name_given_by_user} ({self.host} / {unique_id})"
+                },
+                "configuration_url": f"http://{self.host}:{new_discovered_romy.port}",
+            }
+        )
+
+        # if robot got already unlocked with password add it directly
+        if not new_discovered_romy.is_initialized:
+            return self.async_abort(reason="cannot_connect")
+
+        if new_discovered_romy.is_unlocked:
+            return await self.async_step_zeroconf_confirm()
+
+        return await self.async_step_password()
+
+    async def async_step_zeroconf_confirm(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Handle a confirmation flow initiated by zeroconf."""
+        if user_input is None:
+            return self.async_show_form(
+                step_id="zeroconf_confirm",
+                description_placeholders={
+                    "name": self.robot_name_given_by_user,
+                    "host": self.host,
+                },
+            )
+        return await self._async_step_finish_config()
+
+    async def _async_step_finish_config(self) -> FlowResult:
+        """Finish the configuration setup."""
+        return self.async_create_entry(
+            title=self.robot_name_given_by_user,
+            data={
+                CONF_HOST: self.host,
+                CONF_PASSWORD: self.password,
+            },
+        )

homeassistant/components/romy/const.py

@@ -0,0 +1,11 @@
+"""Constants for the ROMY integration."""
+
+from datetime import timedelta
+import logging
+
+from homeassistant.const import Platform
+
+DOMAIN = "romy"
+PLATFORMS = [Platform.VACUUM]
+UPDATE_INTERVAL = timedelta(seconds=5)
+LOGGER = logging.getLogger(__package__)

homeassistant/components/romy/coordinator.py

@@ -0,0 +1,22 @@
+"""ROMY coordinator."""
+
+from romy import RomyRobot
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN, LOGGER, UPDATE_INTERVAL
+
+
+class RomyVacuumCoordinator(DataUpdateCoordinator[None]):
+    """ROMY Vacuum Coordinator."""
+
+    def __init__(self, hass: HomeAssistant, romy: RomyRobot) -> None:
+        """Initialize."""
+        super().__init__(hass, LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
+        self.hass = hass
+        self.romy = romy
+
+    async def _async_update_data(self) -> None:
+        """Update ROMY Vacuum Cleaner data."""
+        await self.romy.async_update()

homeassistant/components/romy/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "romy",
+  "name": "ROMY Vacuum Cleaner",
+  "codeowners": ["@xeniter"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/romy",
+  "iot_class": "local_polling",
+  "requirements": ["romy==0.0.7"],
+  "zeroconf": ["_aicu-http._tcp.local."]
+}

homeassistant/components/romy/strings.json

@@ -0,0 +1,51 @@
+{
+  "config": {
+    "flow_title": "{name}",
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
+    },
+    "step": {
+      "user": {
+        "data": {
+          "host": "[%key:common::config_flow::data::host%]"
+        }
+      },
+      "password": {
+        "title": "Password required",
+        "data": {
+          "password": "[%key:common::config_flow::data::password%]"
+        },
+        "data_description": {
+          "password": "(8 characters, see QR Code under the dustbin)."
+        }
+      },
+      "zeroconf_confirm": {
+        "description": "Do you want to add ROMY Vacuum Cleaner {name} to Home Assistant?"
+      }
+    }
+  },
+  "entity": {
+    "vacuum": {
+      "romy": {
+        "state_attributes": {
+          "fan_speed": {
+            "state": {
+              "default": "Default",
+              "normal": "Normal",
+              "silent": "Silent",
+              "intensive": "Intensive",
+              "super_silent": "Super silent",
+              "high": "High",
+              "auto": "Auto"
+            }
+          }
+        }
+      }
+    }
+  }
+}