Add solarman integration

homeassistant/components/indevolt/__init__.py

@@ -0,0 +1,32 @@
+"""Home Assistant integration for INDEVOLT devices."""
+
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+
+from .const import PLATFORMS
+from .coordinator import IndevoltDeviceUpdateCoordinator
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Indevolt from a config entry."""
+    coordinator = IndevoltDeviceUpdateCoordinator(hass, entry)
+
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = 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."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+    if unload_ok and entry.runtime_data:
+        coordinator = entry.runtime_data
+        await coordinator.async_shutdown()
+        entry.runtime_data = None
+
+    return unload_ok

homeassistant/components/indevolt/config_flow.py

@@ -0,0 +1,222 @@
+"""Config flow for Indevolt integration."""
+
+import logging
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
+
+from .const import DEFAULT_PORT, DEFAULT_SCAN_INTERVAL, DOMAIN, SUPPORTED_MODELS
+from .indevolt import Indevolt
+from .utils import get_device_gen
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class IndevoltConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Indevolt."""
+
+    VERSION = 1
+
+    host = ""
+    port: int
+    model = ""
+    device_sn = ""
+    fw_version = ""
+
+    async def async_step_user(self, user_input=None) -> ConfigFlowResult:
+        """Handle the initial step via user interface."""
+        errors = {}
+        if user_input is not None:
+            self.host = user_input["host"]
+            self.port = user_input.get("port", DEFAULT_PORT)
+            self.model = user_input["model"]
+
+            if get_device_gen(self.model) == 1:
+                self.fw_version = "V1.3.0A_R006.072_M4848_00000039"
+            else:
+                self.fw_version = "V1.3.09_R00D.012_M4801_00000015"
+
+            try:
+                self.device_sn = await self.get_device_sn(self.host, self.port)
+
+            except TimeoutError:
+                errors["base"] = "timeout"
+            except ConnectionError:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unknown error occurred while verifying device")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(f"{self.model}_{self.device_sn}")
+                self._abort_if_unique_id_configured()
+
+                if not errors:
+                    return self.async_create_entry(
+                        title=f"{self.model} ({self.host})",
+                        data={
+                            "host": self.host,
+                            "port": self.port,
+                            "scan_interval": user_input.get(
+                                "scan_interval", DEFAULT_SCAN_INTERVAL
+                            ),
+                            "sn": self.device_sn,
+                            "fw_version": self.fw_version,
+                            "model": self.model,
+                        },
+                    )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required("host"): str,
+                    vol.Optional("port", default=DEFAULT_PORT): int,
+                    vol.Optional("scan_interval", default=DEFAULT_SCAN_INTERVAL): int,
+                    vol.Required("model"): vol.In(SUPPORTED_MODELS),
+                }
+            ),
+            errors=errors,
+        )
+
+    async def async_step_reconfigure(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle reconfiguration of the integration."""
+        errors: dict[str, str] = {}
+        reconfigure_entry = self._get_reconfigure_entry()
+
+        if user_input is not None:
+            self.host = user_input["host"]
+            self.port = user_input["port"]
+            self.model = user_input["model"]
+
+            if get_device_gen(self.model) == 1:
+                self.fw_version = "V1.3.0A_R006.072_M4848_00000039"
+            else:
+                self.fw_version = "V1.3.09_R00D.012_M4801_00000015"
+
+            try:
+                self.device_sn = await self.get_device_sn(self.host, self.port)
+
+            except TimeoutError:
+                errors["base"] = "timeout"
+            except ConnectionError:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unknown error occurred while verifying device")
+                errors["base"] = "unknown"
+            else:
+                await self.async_set_unique_id(f"{self.model}_{self.device_sn}")
+                self._abort_if_unique_id_mismatch(reason="another_device")
+
+                return self.async_update_reload_and_abort(
+                    self._get_reconfigure_entry(),
+                    data_updates={
+                        "host": self.host,
+                        "port": self.port,
+                        "scan_interval": user_input.get(
+                            "scan_interval", DEFAULT_SCAN_INTERVAL
+                        ),
+                        "sn": self.device_sn,
+                        "fw_version": self.fw_version,
+                        "model": self.model,
+                    },
+                )
+        return self.async_show_form(
+            step_id="reconfigure",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(
+                        "host", default=reconfigure_entry.data.get("host")
+                    ): str,
+                    vol.Optional(
+                        "port",
+                        default=reconfigure_entry.data.get("port"),
+                    ): int,
+                    vol.Optional(
+                        "scan_interval",
+                        default=reconfigure_entry.data.get("scan_interval"),
+                    ): int,
+                    vol.Required(
+                        "model",
+                        default=reconfigure_entry.data.get("model"),
+                    ): vol.In(SUPPORTED_MODELS),
+                }
+            ),
+            description_placeholders={
+                "title": reconfigure_entry.title,
+            },
+            errors=errors,
+        )
+
+    async def async_step_zeroconf(
+        self, discovery_info: ZeroconfServiceInfo
+    ) -> ConfigFlowResult:
+        """Handle zeroconf discovery."""
+        self.host = discovery_info.host
+        self.model = discovery_info.properties["product_type"]
+        self.device_sn = discovery_info.properties["serial"]
+        self.fw_version = discovery_info.properties["fw_version"]
+
+        await self.async_set_unique_id(f"{self.model}_{self.device_sn}")
+        self._abort_if_unique_id_configured(updates={"host": discovery_info.host})
+
+        return await self.async_step_discovery_confirm()
+
+    async def async_step_discovery_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Confirm discovery."""
+        assert self.host
+        assert self.model
+        assert self.device_sn
+
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                await self.get_device_sn(self.host, DEFAULT_PORT)
+
+            except TimeoutError:
+                errors["base"] = "timeout"
+            except ConnectionError:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unknown error occurred while verifying device")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title=f"{self.model} ({self.host})",
+                    data={
+                        "host": self.host,
+                        "port": DEFAULT_PORT,
+                        "scan_interval": DEFAULT_SCAN_INTERVAL,
+                        "sn": self.device_sn,
+                        "fw_version": self.fw_version,
+                        "model": self.model,
+                    },
+                )
+
+        self._set_confirm_only()
+
+        name = f"{self.model} ({self.device_sn})"
+        self.context["title_placeholders"] = {"name": name}
+
+        return self.async_show_form(
+            step_id="discovery_confirm",
+            description_placeholders={
+                "model": self.model,
+                "sn": self.device_sn,
+                "host": self.host,
+            },
+            errors=errors,
+        )
+
+    async def get_device_sn(self, host: str, port: int) -> str:
+        """Get serial number from the device."""
+        device = Indevolt(async_get_clientsession(self.hass), host, port)
+        data = await device.fetch_data(0)
+        return data.get("0", "unknown")

homeassistant/components/indevolt/const.py

@@ -0,0 +1,10 @@
+"""Constants for the indevolt integration."""
+
+from homeassistant.const import Platform
+
+DOMAIN = "indevolt"
+DEFAULT_PORT = 8080
+DEFAULT_SCAN_INTERVAL = 30
+PLATFORMS = [Platform.SENSOR]
+
+SUPPORTED_MODELS = ["BK1600/BK1600Ultra", "SolidFlex/PowerFlex2000"]

homeassistant/components/indevolt/coordinator.py

@@ -0,0 +1,64 @@
+"""Coordinator for indevolt integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
+from .indevolt import KEYS_GEN1, KEYS_GEN2, Indevolt
+from .utils import get_device_gen
+
+_LOGGER = logging.getLogger(__name__)
+
+type IndevoltConfigEntry = ConfigEntry[IndevoltDeviceUpdateCoordinator]
+
+
+class IndevoltDeviceUpdateCoordinator(DataUpdateCoordinator):
+    """Coordinator for managing Indevolt device data updates and control operations."""
+
+    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+        """Initialize the Indevolt device coordinator."""
+        super().__init__(
+            hass,
+            logger=_LOGGER,
+            config_entry=config_entry,
+            name=DOMAIN,
+            update_interval=timedelta(
+                seconds=config_entry.data.get("scan_interval", DEFAULT_SCAN_INTERVAL)
+            ),
+        )
+
+        # Initialize the API client for communicating with the Indevolt device.
+        self.api = Indevolt(
+            async_get_clientsession(hass),
+            config_entry.data["host"],
+            config_entry.data["port"],
+        )
+
+    async def _async_update_data(self):
+        """Fetch and update device data.
+
+        This is automatically called by the DataUpdateCoordinator framework
+        according to the defined update_interval.
+        """
+        try:
+            data = {}
+            if get_device_gen(self.config_entry.data["model"]) == 1:
+                data = await self.api.fetch_all_data(KEYS_GEN1)
+            else:
+                data = await self.api.fetch_all_data(KEYS_GEN2)
+            self.data = data
+
+        except ConnectionError as e:
+            raise UpdateFailed(
+                translation_domain=DOMAIN,
+                translation_key="update_failed",
+            ) from e
+        else:
+            return data

homeassistant/components/indevolt/diagnostics.py

@@ -0,0 +1,27 @@
+"""Diagnostics support for Indevolt."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.diagnostics import async_redact_data
+from homeassistant.core import HomeAssistant
+
+from .coordinator import IndevoltConfigEntry
+
+TO_REDACT = {"sn"}
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, entry: IndevoltConfigEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry."""
+    data = entry.runtime_data.data
+
+    return async_redact_data(
+        {
+            "entry": async_redact_data(entry.data, TO_REDACT),
+            "data": data,
+        },
+        TO_REDACT,
+    )

homeassistant/components/indevolt/entity.py

@@ -0,0 +1,34 @@
+"""Base entity for the Indevolt integration."""
+
+from __future__ import annotations
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import IndevoltDeviceUpdateCoordinator
+
+
+class IndevoltEntity(CoordinatorEntity[IndevoltDeviceUpdateCoordinator]):
+    """Defines an Indevolt entity."""
+
+    _attr_has_entity_name = True
+
+    def __init__(self, coordinator: IndevoltDeviceUpdateCoordinator) -> None:
+        """Initialize the Indevolt entity."""
+        super().__init__(coordinator)
+
+        entry = coordinator.config_entry
+        if entry is not None:
+            sn = entry.data.get("sn", "unknown")
+            model = entry.data.get("model", "unknown")
+            sw_version = entry.data.get("fw_version", "unknown")
+
+            self._attr_device_info = DeviceInfo(
+                identifiers={(DOMAIN, sn)},
+                name=f"Indevolt {model}",
+                manufacturer="INDEVOLT",
+                sw_version=sw_version,
+                model=model,
+                serial_number=sn,
+            )