Add Lunatone gateway integration (home-assistant#149182)

Co-authored-by: Abílio Costa <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

.strict-typing

@@ -326,6 +326,7 @@ homeassistant.components.london_underground.*
 homeassistant.components.lookin.*
 homeassistant.components.lovelace.*
 homeassistant.components.luftdaten.*
+homeassistant.components.lunatone.*
 homeassistant.components.madvr.*
 homeassistant.components.manual.*
 homeassistant.components.mastodon.*

CODEOWNERS

@@ -0,0 +1,64 @@
+"""The Lunatone integration."""
+
+from typing import Final
+
+from lunatone_rest_api_client import Auth, Devices, Info
+
+from homeassistant.const import CONF_URL, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryError
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+from .coordinator import (
+    LunatoneConfigEntry,
+    LunatoneData,
+    LunatoneDevicesDataUpdateCoordinator,
+    LunatoneInfoDataUpdateCoordinator,
+)
+
+PLATFORMS: Final[list[Platform]] = [Platform.LIGHT]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: LunatoneConfigEntry) -> bool:
+    """Set up Lunatone from a config entry."""
+    auth_api = Auth(async_get_clientsession(hass), entry.data[CONF_URL])
+    info_api = Info(auth_api)
+    devices_api = Devices(auth_api)
+
+    coordinator_info = LunatoneInfoDataUpdateCoordinator(hass, entry, info_api)
+    await coordinator_info.async_config_entry_first_refresh()
+
+    if info_api.serial_number is None:
+        raise ConfigEntryError(
+            translation_domain=DOMAIN, translation_key="missing_device_info"
+        )
+
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, str(info_api.serial_number))},
+        name=info_api.name,
+        manufacturer="Lunatone",
+        sw_version=info_api.version,
+        hw_version=info_api.data.device.pcb,
+        configuration_url=entry.data[CONF_URL],
+        serial_number=str(info_api.serial_number),
+        model_id=(
+            f"{info_api.data.device.article_number}{info_api.data.device.article_info}"
+        ),
+    )
+
+    coordinator_devices = LunatoneDevicesDataUpdateCoordinator(hass, entry, devices_api)
+    await coordinator_devices.async_config_entry_first_refresh()
+
+    entry.runtime_data = LunatoneData(coordinator_info, coordinator_devices)
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: LunatoneConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/lunatone/__init__.py

@@ -0,0 +1,83 @@
+"""Config flow for Lunatone."""
+
+from typing import Any, Final
+
+import aiohttp
+from lunatone_rest_api_client import Auth, Info
+import voluptuous as vol
+
+from homeassistant.config_entries import (
+    SOURCE_RECONFIGURE,
+    ConfigFlow,
+    ConfigFlowResult,
+)
+from homeassistant.const import CONF_URL
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import DOMAIN
+
+DATA_SCHEMA: Final[vol.Schema] = vol.Schema(
+    {vol.Required(CONF_URL, default="http://"): cv.string},
+)
+
+
+def compose_title(name: str | None, serial_number: int) -> str:
+    """Compose a title string from a given name and serial number."""
+    return f"{name or 'DALI Gateway'} {serial_number}"
+
+
+class LunatoneConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Lunatone config flow."""
+
+    VERSION = 1
+    MINOR_VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initialized by the user."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            url = user_input[CONF_URL]
+            data = {CONF_URL: url}
+            self._async_abort_entries_match(data)
+            auth_api = Auth(
+                session=async_get_clientsession(self.hass),
+                base_url=url,
+            )
+            info_api = Info(auth_api)
+            try:
+                await info_api.async_update()
+            except aiohttp.InvalidUrlClientError:
+                errors["base"] = "invalid_url"
+            except aiohttp.ClientConnectionError:
+                errors["base"] = "cannot_connect"
+            else:
+                if info_api.data is None or info_api.serial_number is None:
+                    errors["base"] = "missing_device_info"
+                else:
+                    await self.async_set_unique_id(str(info_api.serial_number))
+                    if self.source == SOURCE_RECONFIGURE:
+                        self._abort_if_unique_id_mismatch()
+                        return self.async_update_reload_and_abort(
+                            self._get_reconfigure_entry(),
+                            data_updates=data,
+                            title=compose_title(info_api.name, info_api.serial_number),
+                        )
+                    self._abort_if_unique_id_configured()
+                    return self.async_create_entry(
+                        title=compose_title(info_api.name, info_api.serial_number),
+                        data={CONF_URL: url},
+                    )
+        return self.async_show_form(
+            step_id="user",
+            data_schema=DATA_SCHEMA,
+            errors=errors,
+        )
+
+    async def async_step_reconfigure(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a reconfiguration flow initialized by the user."""
+        return await self.async_step_user(user_input)

homeassistant/components/lunatone/config_flow.py

@@ -0,0 +1,5 @@
+"""Constants for the Lunatone integration."""
+
+from typing import Final
+
+DOMAIN: Final = "lunatone"

homeassistant/components/lunatone/const.py

@@ -0,0 +1,101 @@
+"""Coordinator for handling data fetching and updates."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import timedelta
+import logging
+
+import aiohttp
+from lunatone_rest_api_client import Device, Devices, Info
+from lunatone_rest_api_client.models import InfoData
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+DEFAULT_DEVICES_SCAN_INTERVAL = timedelta(seconds=10)
+
+
+@dataclass
+class LunatoneData:
+    """Data for Lunatone integration."""
+
+    coordinator_info: LunatoneInfoDataUpdateCoordinator
+    coordinator_devices: LunatoneDevicesDataUpdateCoordinator
+
+
+type LunatoneConfigEntry = ConfigEntry[LunatoneData]
+
+
+class LunatoneInfoDataUpdateCoordinator(DataUpdateCoordinator[InfoData]):
+    """Data update coordinator for Lunatone info."""
+
+    config_entry: LunatoneConfigEntry
+
+    def __init__(
+        self, hass: HomeAssistant, config_entry: LunatoneConfigEntry, info_api: Info
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=config_entry,
+            name=f"{DOMAIN}-info",
+            always_update=False,
+        )
+        self.info_api = info_api
+
+    async def _async_update_data(self) -> InfoData:
+        """Update info data."""
+        try:
+            await self.info_api.async_update()
+        except aiohttp.ClientConnectionError as ex:
+            raise UpdateFailed(
+                "Unable to retrieve info data from Lunatone REST API"
+            ) from ex
+
+        if self.info_api.data is None:
+            raise UpdateFailed("Did not receive info data from Lunatone REST API")
+        return self.info_api.data
+
+
+class LunatoneDevicesDataUpdateCoordinator(DataUpdateCoordinator[dict[int, Device]]):
+    """Data update coordinator for Lunatone devices."""
+
+    config_entry: LunatoneConfigEntry
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: LunatoneConfigEntry,
+        devices_api: Devices,
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            config_entry=config_entry,
+            name=f"{DOMAIN}-devices",
+            always_update=False,
+            update_interval=DEFAULT_DEVICES_SCAN_INTERVAL,
+        )
+        self.devices_api = devices_api
+
+    async def _async_update_data(self) -> dict[int, Device]:
+        """Update devices data."""
+        try:
+            await self.devices_api.async_update()
+        except aiohttp.ClientConnectionError as ex:
+            raise UpdateFailed(
+                "Unable to retrieve devices data from Lunatone REST API"
+            ) from ex
+
+        if self.devices_api.data is None:
+            raise UpdateFailed("Did not receive devices data from Lunatone REST API")
+
+        return {device.id: device for device in self.devices_api.devices}

homeassistant/components/lunatone/coordinator.py

@@ -0,0 +1,103 @@
+"""Platform for Lunatone light integration."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from homeassistant.components.light import ColorMode, LightEntity
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import LunatoneConfigEntry, LunatoneDevicesDataUpdateCoordinator
+
+PARALLEL_UPDATES = 0
+STATUS_UPDATE_DELAY = 0.04
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: LunatoneConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the Lunatone Light platform."""
+    coordinator_info = config_entry.runtime_data.coordinator_info
+    coordinator_devices = config_entry.runtime_data.coordinator_devices
+
+    async_add_entities(
+        [
+            LunatoneLight(
+                coordinator_devices, device_id, coordinator_info.data.device.serial
+            )
+            for device_id in coordinator_devices.data
+        ]
+    )
+
+
+class LunatoneLight(
+    CoordinatorEntity[LunatoneDevicesDataUpdateCoordinator], LightEntity
+):
+    """Representation of a Lunatone light."""
+
+    _attr_color_mode = ColorMode.ONOFF
+    _attr_supported_color_modes = {ColorMode.ONOFF}
+    _attr_has_entity_name = True
+    _attr_name = None
+    _attr_should_poll = False
+
+    def __init__(
+        self,
+        coordinator: LunatoneDevicesDataUpdateCoordinator,
+        device_id: int,
+        interface_serial_number: int,
+    ) -> None:
+        """Initialize a LunatoneLight."""
+        super().__init__(coordinator=coordinator)
+        self._device_id = device_id
+        self._interface_serial_number = interface_serial_number
+        self._device = self.coordinator.data.get(self._device_id)
+        self._attr_unique_id = f"{interface_serial_number}-device{device_id}"
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return the device info."""
+        assert self.unique_id
+        name = self._device.name if self._device is not None else None
+        return DeviceInfo(
+            identifiers={(DOMAIN, self.unique_id)},
+            name=name,
+            via_device=(DOMAIN, str(self._interface_serial_number)),
+        )
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return super().available and self._device is not None
+
+    @property
+    def is_on(self) -> bool:
+        """Return True if light is on."""
+        return self._device is not None and self._device.is_on
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._device = self.coordinator.data.get(self._device_id)
+        self.async_write_ha_state()
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Instruct the light to turn on."""
+        assert self._device
+        await self._device.switch_on()
+        await asyncio.sleep(STATUS_UPDATE_DELAY)
+        await self.coordinator.async_refresh()
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Instruct the light to turn off."""
+        assert self._device
+        await self._device.switch_off()
+        await asyncio.sleep(STATUS_UPDATE_DELAY)
+        await self.coordinator.async_refresh()

homeassistant/components/lunatone/light.py

@@ -0,0 +1,11 @@
+{
+  "domain": "lunatone",
+  "name": "Lunatone",
+  "codeowners": ["@MoonDevLT"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/lunatone",
+  "integration_type": "hub",
+  "iot_class": "local_polling",
+  "quality_scale": "silver",
+  "requirements": ["lunatone-rest-api-client==0.4.8"]
+}