Legrand Whole Home Lighting Integration - Initial Commit

Author: bas-tard

.strict-typing

@@ -297,6 +297,7 @@ homeassistant.components.lamarzocco.*
 homeassistant.components.lametric.*
 homeassistant.components.laundrify.*
 homeassistant.components.lawn_mower.*
+homeassistant.components.lc7001.*
 homeassistant.components.lcn.*
 homeassistant.components.ld2410_ble.*
 homeassistant.components.led_ble.*

CODEOWNERS

@@ -0,0 +1,41 @@
+"""The Legrand Whole Home Lighting integration."""
+
+import logging
+
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, Platform
+from homeassistant.core import HomeAssistant
+
+from .common import LcConfigEntry
+from .engine.engine import ConnectionState, Engine
+
+_PLATFORMS: list[Platform] = [Platform.LIGHT]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: LcConfigEntry) -> bool:
+    """Set up Legrand Whole Home Lighting from a config entry."""
+
+    engine = Engine(
+        host=entry.data[CONF_HOST],
+        port=entry.data[CONF_PORT],
+        password=entry.data[CONF_PASSWORD],
+    )
+
+    try:
+        engine.connect()
+        engine.start()
+        await engine.waitForState(ConnectionState.Ready)
+    except TimeoutError:
+        return False
+
+    entry.runtime_data = engine
+    await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: LcConfigEntry) -> bool:
+    """Unload a config entry."""
+
+    return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)

homeassistant/components/lc7001/__init__.py

@@ -0,0 +1,7 @@
+"""Common types and definitions."""
+
+from homeassistant.config_entries import ConfigEntry
+
+from .engine.engine import Engine
+
+type LcConfigEntry = ConfigEntry[Engine]

homeassistant/components/lc7001/common.py

@@ -0,0 +1,113 @@
+"""Config flow for the Legrand Whole Home Lighting integration."""
+
+import logging
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT
+from homeassistant.exceptions import HomeAssistantError, IntegrationError
+
+from .const import DEFAULT_HOST, DEFAULT_PORT, DEVICE_HUB, DOMAIN
+from .engine.engine import ConnectionState, Engine
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Optional(CONF_HOST, default=DEFAULT_HOST): str,
+        vol.Optional(CONF_PORT, default=DEFAULT_PORT): int,
+        vol.Required(CONF_PASSWORD): str,
+    }
+)
+
+
+class LcConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Legrand Whole Home Lighting."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize."""
+        super().__init__()
+        self.data: dict[str, Any] = {}
+
+    async def validate_input(self, data: dict[str, Any]) -> dict[str, Any]:
+        """Validate the user input allows us to connect."""
+
+        host = data[CONF_HOST]
+        port = data[CONF_PORT]
+        password = data[CONF_PASSWORD]
+
+        if host is None:
+            raise ConfigEntryNotReady
+        if port is None:
+            raise ConfigEntryNotReady
+        if password is None:
+            raise ConfigEntryNotReady
+
+        try:
+            engine = Engine(host=host, port=port, password=password)
+
+            engine.connect()
+            engine.start()
+
+            await engine.waitForState(ConnectionState.Ready)
+            mac = engine.systemInfo.MACAddress
+
+        except TimeoutError as error:
+            raise CannotConnect from error
+
+        finally:
+            engine.disconnect()
+
+        # Return info that you want to store in the config entry.
+        return {"title": DEVICE_HUB, "mac": mac}
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+
+        errors: dict[str, str] = {}
+        if user_input:
+            try:
+                info = await self.validate_input(data=user_input)
+
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+            else:
+                await self.async_set_unique_id(info["mac"])
+                self._abort_if_unique_id_configured()
+                return self.async_create_entry(title=info["title"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_reconfigure(
+        self, user_input: dict[str, Any]
+    ) -> ConfigFlowResult:
+        """Handle reconfiguration."""
+        return await self.async_step_user()
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""
+
+
+class ConfigEntryNotReady(IntegrationError):
+    """Error to indicate there is invalid config."""

homeassistant/components/lc7001/config_flow.py

@@ -0,0 +1,8 @@
+"""Constants for the Legrand Whole Home Lighting integration."""
+
+DOMAIN = "lc7001"
+
+DEVICE_HUB = "LC7001"
+
+DEFAULT_HOST = "LCM1.local"
+DEFAULT_PORT = 2112

homeassistant/components/lc7001/const.py

@@ -0,0 +1,33 @@
+"""Diagnostics support for Acaia."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from homeassistant.components.diagnostics import async_redact_data
+from homeassistant.core import HomeAssistant
+
+from .common import LcConfigEntry
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant,
+    entry: LcConfigEntry,
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry."""
+    engine = entry.runtime_data
+
+    # collect all data sources
+    diagnostics: dict[str, Any] = {
+        "config_entry": async_redact_data(entry, {}),
+        "model": engine.systemInfo.Model,
+        "device_state": engine.state.name,
+        "mac": engine.systemInfo.MACAddress,
+    }
+
+    _LOGGER.debug("Diagnostics: %s", diagnostics)
+
+    return diagnostics

homeassistant/components/lc7001/diagnostics.py

@@ -0,0 +1 @@
+"""The Engine for the LC7001 hub."""

homeassistant/components/lc7001/engine/__init__.py

@@ -0,0 +1,26 @@
+"""A BroadcastMemory packet."""
+
+from typing import Any
+
+from .packet import Packet
+
+
+class BroadcastMemory(Packet):
+    """A BroadcastMemory packet."""
+
+    _service_name = "BroadcastMemory"
+
+    def __init__(self, **kwargs: Any) -> None:
+        """Initialize a BroadcastMemory packet."""
+        super().__init__(**kwargs)
+
+        self.FreeMemory = kwargs.get("FreeMemory")
+        self.FreeMemLowWater = kwargs.get("FreeMemLowWater")
+        self.Malloc_Count = kwargs.get("Malloc_Count")
+        self.Free_Count = kwargs.get("Free_Count")
+        self.JsonConnections = kwargs.get("JsonConnections")
+        self.StaticRamUsage = kwargs.get("StaticRamUsage")
+        self.PeakRamUsage = kwargs.get("PeakRamUsage")
+        self.CurrentTime = kwargs.get("CurrentTime")
+        self.uSec = kwargs.get("uSec")
+        self.Seqno = kwargs.get("Seqno")