Implement TechnoVE integration

.strict-typing

@@ -344,6 +344,7 @@ homeassistant.components.tailwind.*
 homeassistant.components.tami4.*
 homeassistant.components.tautulli.*
 homeassistant.components.tcp.*
+homeassistant.components.technove.*
 homeassistant.components.tedee.*
 homeassistant.components.text.*
 homeassistant.components.threshold.*

CODEOWNERS

@@ -1316,6 +1316,8 @@ build.json @home-assistant/supervisor
 /tests/components/tasmota/ @emontnemery
 /homeassistant/components/tautulli/ @ludeeus @tkdrob
 /tests/components/tautulli/ @ludeeus @tkdrob
+/homeassistant/components/technove/ @Moustachauve
+/tests/components/technove/ @Moustachauve
 /homeassistant/components/tedee/ @patrickhilker @zweckj
 /tests/components/tedee/ @patrickhilker @zweckj
 /homeassistant/components/tellduslive/ @fredrike

homeassistant/components/technove/__init__.py

@@ -0,0 +1,31 @@
+"""The TechnoVE integration."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import TechnoVEDataUpdateCoordinator
+
+PLATFORMS = Platform.SENSOR
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up TechnoVE from a config entry."""
+    coordinator = TechnoVEDataUpdateCoordinator(hass, entry=entry)
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/technove/config_flow.py

@@ -0,0 +1,102 @@
+"""Config flow for TechnoVE."""
+
+from typing import Any
+
+from technove import Station as TechnoVEStation, TechnoVE, TechnoVEConnectionError
+import voluptuous as vol
+
+from homeassistant.components import onboarding, zeroconf
+from homeassistant.config_entries import ConfigFlow
+from homeassistant.const import CONF_IP_ADDRESS, CONF_MAC
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+
+class TechnoVEConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for TechnoVE."""
+
+    VERSION = 1
+    discovered_host: str
+    discovered_station: TechnoVEStation
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle a flow initiated by the user."""
+        errors = {}
+        if user_input is not None:
+            try:
+                station = await self._async_get_station(user_input[CONF_IP_ADDRESS])
+            except TechnoVEConnectionError:
+                errors["base"] = "cannot_connect"
+            else:
+                await self.async_set_unique_id(station.info.mac_address)
+                self._abort_if_unique_id_configured(
+                    updates={CONF_IP_ADDRESS: user_input[CONF_IP_ADDRESS]}
+                )
+                return self.async_create_entry(
+                    title=station.info.name,
+                    data={
+                        CONF_IP_ADDRESS: user_input[CONF_IP_ADDRESS],
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema({vol.Required(CONF_IP_ADDRESS): str}),
+            errors=errors or {},
+        )
+
+    async def async_step_zeroconf(
+        self, discovery_info: zeroconf.ZeroconfServiceInfo
+    ) -> FlowResult:
+        """Handle zeroconf discovery."""
+        # Abort quick if the mac address is provided by discovery info
+        if mac := discovery_info.properties.get(CONF_MAC):
+            await self.async_set_unique_id(mac)
+            self._abort_if_unique_id_configured(
+                updates={CONF_IP_ADDRESS: discovery_info.host}
+            )
+
+        self.discovered_host = discovery_info.host
+        try:
+            self.discovered_station = await self._async_get_station(discovery_info.host)
+        except TechnoVEConnectionError:
+            return self.async_abort(reason="cannot_connect")
+
+        await self.async_set_unique_id(self.discovered_station.info.mac_address)
+        self._abort_if_unique_id_configured(
+            updates={CONF_IP_ADDRESS: discovery_info.host}
+        )
+
+        self.context.update(
+            {
+                "title_placeholders": {"name": self.discovered_station.info.name},
+            }
+        )
+        return await self.async_step_zeroconf_confirm()
+
+    async def async_step_zeroconf_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle a flow initiated by zeroconf."""
+        if user_input is not None or not onboarding.async_is_onboarded(self.hass):
+            return self.async_create_entry(
+                title=self.discovered_station.info.name,
+                data={
+                    CONF_IP_ADDRESS: self.discovered_host,
+                },
+            )
+
+        return self.async_show_form(
+            step_id="zeroconf_confirm",
+            description_placeholders={"name": self.discovered_station.info.name},
+        )
+
+    async def _async_get_station(self, host: str) -> TechnoVEStation:
+        """Get information from a TechnoVE station."""
+        session = async_get_clientsession(self.hass)
+        api = TechnoVE(host, session=session)
+        return await api.update()

homeassistant/components/technove/const.py

@@ -0,0 +1,8 @@
+"""Constants for the TechnoVE integration."""
+from datetime import timedelta
+import logging
+
+DOMAIN = "technove"
+
+LOGGER = logging.getLogger(__package__)
+SCAN_INTERVAL = timedelta(seconds=5)

homeassistant/components/technove/coordinator.py

@@ -0,0 +1,46 @@
+"""DataUpdateCoordinator for TechnoVE."""
+from __future__ import annotations
+
+from technove import Station as TechnoVEStation, TechnoVE, TechnoVEError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_IP_ADDRESS
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN, LOGGER, SCAN_INTERVAL
+
+
+class TechnoVEDataUpdateCoordinator(DataUpdateCoordinator[TechnoVEStation]):
+    """Class to manage fetching TechnoVE data from single endpoint."""
+
+    config_entry: ConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        *,
+        entry: ConfigEntry,
+    ) -> None:
+        """Initialize global TechnoVE data updater."""
+        self.technove = TechnoVE(
+            entry.data[CONF_IP_ADDRESS], session=async_get_clientsession(hass)
+        )
+        self.unsub: CALLBACK_TYPE | None = None
+
+        super().__init__(
+            hass,
+            LOGGER,
+            name=DOMAIN,
+            update_interval=SCAN_INTERVAL,
+        )
+
+    async def _async_update_data(self) -> TechnoVEStation:
+        """Fetch data from TechnoVE."""
+        try:
+            station = await self.technove.update()
+        except TechnoVEError as error:
+            raise UpdateFailed(f"Invalid response from API: {error}") from error
+
+        return station

homeassistant/components/technove/diagnostics.py

@@ -0,0 +1,20 @@
+"""Diagnostics support for TechnoVE."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import TechnoVEDataUpdateCoordinator
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, entry: ConfigEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry."""
+    coordinator: TechnoVEDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+
+    data: dict[str, Any] = {"info": coordinator.data.info.__dict__}
+    return data

homeassistant/components/technove/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "technove",
+  "name": "TechnoVE",
+  "codeowners": ["@Moustachauve"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/technove",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "requirements": ["python-technove==1.0.0"],
+  "zeroconf": ["_technove-stations._tcp.local."]
+}

homeassistant/components/technove/models.py

@@ -0,0 +1,26 @@
+"""Models for TechnoVE."""
+from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import TechnoVEDataUpdateCoordinator
+
+
+class TechnoVEEntity(CoordinatorEntity[TechnoVEDataUpdateCoordinator]):
+    """Defines a base TechnoVE entity."""
+
+    _attr_has_entity_name = True
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return information about this TechnoVE station."""
+        return DeviceInfo(
+            connections={
+                (CONNECTION_NETWORK_MAC, self.coordinator.data.info.mac_address)
+            },
+            identifiers={(DOMAIN, self.coordinator.data.info.mac_address)},
+            name=self.coordinator.data.info.name,
+            manufacturer="TechnoVE",
+            model=f"TechnoVE i{self.coordinator.data.info.max_station_current}",
+            sw_version=self.coordinator.data.info.version,
+        )