Add livisi integration (#76863)

CODEOWNERS

@@ -635,6 +635,8 @@ build.json @home-assistant/supervisor
 /tests/components/litejet/ @joncar
 /homeassistant/components/litterrobot/ @natekspencer @tkdrob
 /tests/components/litterrobot/ @natekspencer @tkdrob
+/homeassistant/components/livisi/ @StefanIacobLivisi
+/tests/components/livisi/ @StefanIacobLivisi
 /homeassistant/components/local_ip/ @issacg
 /tests/components/local_ip/ @issacg
 /homeassistant/components/lock/ @home-assistant/core

homeassistant/components/livisi/__init__.py

@@ -0,0 +1,57 @@
+"""The Livisi Smart Home integration."""
+from __future__ import annotations
+
+import asyncio
+from typing import Final
+
+from aiohttp import ClientConnectorError
+from aiolivisi import AioLivisi
+
+from homeassistant import core
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers import aiohttp_client, device_registry as dr
+
+from .const import DOMAIN, SWITCH_PLATFORM
+from .coordinator import LivisiDataUpdateCoordinator
+
+PLATFORMS: Final = [SWITCH_PLATFORM]
+
+
+async def async_setup_entry(hass: core.HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Livisi Smart Home from a config entry."""
+    web_session = aiohttp_client.async_get_clientsession(hass)
+    aiolivisi = AioLivisi(web_session)
+    coordinator = LivisiDataUpdateCoordinator(hass, entry, aiolivisi)
+    try:
+        await coordinator.async_setup()
+        await coordinator.async_set_all_rooms()
+    except ClientConnectorError as exception:
+        raise ConfigEntryNotReady from exception
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=coordinator.serial_number,
+        identifiers={(DOMAIN, entry.entry_id)},
+        manufacturer="Livisi",
+        name=f"SHC {coordinator.controller_type} {coordinator.serial_number}",
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    await coordinator.async_config_entry_first_refresh()
+    asyncio.create_task(coordinator.ws_connect())
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+
+    unload_success = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    await coordinator.websocket.disconnect()
+    if unload_success:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_success

homeassistant/components/livisi/config_flow.py

@@ -0,0 +1,88 @@
+"""Config flow for Livisi Home Assistant."""
+from __future__ import annotations
+
+from contextlib import suppress
+from typing import Any
+
+from aiohttp import ClientConnectorError
+from aiolivisi import AioLivisi, errors as livisi_errors
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import aiohttp_client
+
+from .const import CONF_HOST, CONF_PASSWORD, DOMAIN, LOGGER
+
+
+class LivisiFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a Livisi Smart Home config flow."""
+
+    def __init__(self) -> None:
+        """Create the configuration file."""
+        self.aio_livisi: AioLivisi = None
+        self.data_schema = vol.Schema(
+            {
+                vol.Required(CONF_HOST): str,
+                vol.Required(CONF_PASSWORD): str,
+            }
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, str] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        if user_input is None:
+            return self.async_show_form(step_id="user", data_schema=self.data_schema)
+
+        errors = {}
+        try:
+            await self._login(user_input)
+        except livisi_errors.WrongCredentialException:
+            errors["base"] = "wrong_password"
+        except livisi_errors.ShcUnreachableException:
+            errors["base"] = "cannot_connect"
+        except livisi_errors.IncorrectIpAddressException:
+            errors["base"] = "wrong_ip_address"
+        else:
+            controller_info: dict[str, Any] = {}
+            with suppress(ClientConnectorError):
+                controller_info = await self.aio_livisi.async_get_controller()
+            if controller_info:
+                return await self.create_entity(user_input, controller_info)
+            errors["base"] = "cannot_connect"
+
+        return self.async_show_form(
+            step_id="user", data_schema=self.data_schema, errors=errors
+        )
+
+    async def _login(self, user_input: dict[str, str]) -> None:
+        """Login into Livisi Smart Home."""
+        web_session = aiohttp_client.async_get_clientsession(self.hass)
+        self.aio_livisi = AioLivisi(web_session)
+        livisi_connection_data = {
+            "ip_address": user_input[CONF_HOST],
+            "password": user_input[CONF_PASSWORD],
+        }
+
+        await self.aio_livisi.async_set_token(livisi_connection_data)
+
+    async def create_entity(
+        self, user_input: dict[str, str], controller_info: dict[str, Any]
+    ) -> FlowResult:
+        """Create LIVISI entity."""
+        if (controller_data := controller_info.get("gateway")) is None:
+            controller_data = controller_info
+        controller_type = controller_data["controllerType"]
+        LOGGER.debug(
+            "Integrating SHC %s with serial number: %s",
+            controller_type,
+            controller_data["serialNumber"],
+        )
+
+        return self.async_create_entry(
+            title=f"SHC {controller_type}",
+            data={
+                **user_input,
+            },
+        )

homeassistant/components/livisi/const.py

@@ -0,0 +1,18 @@
+"""Constants for the Livisi Smart Home integration."""
+import logging
+from typing import Final
+
+LOGGER = logging.getLogger(__package__)
+DOMAIN = "livisi"
+
+CONF_HOST = "host"
+CONF_PASSWORD: Final = "password"
+AVATAR_PORT: Final = 9090
+CLASSIC_PORT: Final = 8080
+DEVICE_POLLING_DELAY: Final = 60
+LIVISI_STATE_CHANGE: Final = "livisi_state_change"
+LIVISI_REACHABILITY_CHANGE: Final = "livisi_reachability_change"
+
+SWITCH_PLATFORM: Final = "switch"
+
+PSS_DEVICE_TYPE: Final = "PSS"

homeassistant/components/livisi/coordinator.py

@@ -0,0 +1,132 @@
+"""Code to manage fetching LIVISI data API."""
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import Any
+
+from aiohttp import ClientConnectorError
+from aiolivisi import AioLivisi, LivisiEvent, Websocket
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    AVATAR_PORT,
+    CLASSIC_PORT,
+    CONF_HOST,
+    CONF_PASSWORD,
+    DEVICE_POLLING_DELAY,
+    LIVISI_REACHABILITY_CHANGE,
+    LIVISI_STATE_CHANGE,
+    LOGGER,
+)
+
+
+class LivisiDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]):
+    """Class to manage fetching LIVISI data API."""
+
+    config_entry: ConfigEntry
+
+    def __init__(
+        self, hass: HomeAssistant, config_entry: ConfigEntry, aiolivisi: AioLivisi
+    ) -> None:
+        """Initialize my coordinator."""
+        super().__init__(
+            hass,
+            LOGGER,
+            name="Livisi devices",
+            update_interval=timedelta(seconds=DEVICE_POLLING_DELAY),
+        )
+        self.config_entry = config_entry
+        self.hass = hass
+        self.aiolivisi = aiolivisi
+        self.websocket = Websocket(aiolivisi)
+        self.devices: set[str] = set()
+        self.rooms: dict[str, Any] = {}
+        self.serial_number: str = ""
+        self.controller_type: str = ""
+        self.is_avatar: bool = False
+        self.port: int = 0
+
+    async def _async_update_data(self) -> list[dict[str, Any]]:
+        """Get device configuration from LIVISI."""
+        try:
+            return await self.async_get_devices()
+        except ClientConnectorError as exc:
+            raise UpdateFailed("Failed to get LIVISI the devices") from exc
+
+    async def async_setup(self) -> None:
+        """Set up the Livisi Smart Home Controller."""
+        if not self.aiolivisi.livisi_connection_data:
+            livisi_connection_data = {
+                "ip_address": self.config_entry.data[CONF_HOST],
+                "password": self.config_entry.data[CONF_PASSWORD],
+            }
+
+            await self.aiolivisi.async_set_token(
+                livisi_connection_data=livisi_connection_data
+            )
+        controller_data = await self.aiolivisi.async_get_controller()
+        if controller_data["controllerType"] == "Avatar":
+            self.port = AVATAR_PORT
+            self.is_avatar = True
+        else:
+            self.port = CLASSIC_PORT
+            self.is_avatar = False
+        self.serial_number = controller_data["serialNumber"]
+        self.controller_type = controller_data["controllerType"]
+
+    async def async_get_devices(self) -> list[dict[str, Any]]:
+        """Set the discovered devices list."""
+        return await self.aiolivisi.async_get_devices()
+
+    async def async_get_pss_state(self, capability: str) -> bool | None:
+        """Set the PSS state."""
+        response: dict[str, Any] = await self.aiolivisi.async_get_pss_state(
+            capability[1:]
+        )
+        if response is None:
+            return None
+        on_state = response["onState"]
+        return on_state["value"]
+
+    async def async_set_all_rooms(self) -> None:
+        """Set the room list."""
+        response: list[dict[str, Any]] = await self.aiolivisi.async_get_all_rooms()
+
+        for available_room in response:
+            available_room_config: dict[str, Any] = available_room["config"]
+            self.rooms[available_room["id"]] = available_room_config["name"]
+
+    def on_data(self, event_data: LivisiEvent) -> None:
+        """Define a handler to fire when the data is received."""
+        if event_data.onState is not None:
+            async_dispatcher_send(
+                self.hass,
+                f"{LIVISI_STATE_CHANGE}_{event_data.source}",
+                event_data.onState,
+            )
+        if event_data.isReachable is not None:
+            async_dispatcher_send(
+                self.hass,
+                f"{LIVISI_REACHABILITY_CHANGE}_{event_data.source}",
+                event_data.isReachable,
+            )
+
+    async def on_close(self) -> None:
+        """Define a handler to fire when the websocket is closed."""
+        for device_id in self.devices:
+            is_reachable: bool = False
+            async_dispatcher_send(
+                self.hass,
+                f"{LIVISI_REACHABILITY_CHANGE}_{device_id}",
+                is_reachable,
+            )
+
+        await self.websocket.connect(self.on_data, self.on_close, self.port)
+
+    async def ws_connect(self) -> None:
+        """Connect the websocket."""
+        await self.websocket.connect(self.on_data, self.on_close, self.port)

homeassistant/components/livisi/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "livisi",
+  "name": "LIVISI Smart Home",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/livisi",
+  "requirements": ["aiolivisi==0.0.14"],
+  "codeowners": ["@StefanIacobLivisi"],
+  "iot_class": "local_polling"
+}

homeassistant/components/livisi/strings.json

@@ -0,0 +1,18 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "description": "Enter the IP address and the (local) password of the SHC.",
+        "data": {
+          "host": "[%key:common::config_flow::data::ip%]",
+          "password": "[%key:common::config_flow::data::password%]"
+        }
+      }
+    },
+    "error": {
+      "wrong_password": "The password is incorrect.",
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "wrong_ip_address": "The IP address is incorrect or the SHC cannot be reached locally."
+    }
+  }
+}

homeassistant/components/livisi/switch.py

@@ -0,0 +1,161 @@
+"""Code to handle a Livisi switches."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.switch import SwitchEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import (
+    DOMAIN,
+    LIVISI_REACHABILITY_CHANGE,
+    LIVISI_STATE_CHANGE,
+    LOGGER,
+    PSS_DEVICE_TYPE,
+)
+from .coordinator import LivisiDataUpdateCoordinator
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up switch device."""
+    coordinator: LivisiDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+
+    @callback
+    def handle_coordinator_update() -> None:
+        """Add switch."""
+        shc_devices: list[dict[str, Any]] = coordinator.data
+        entities: list[SwitchEntity] = []
+        for device in shc_devices:
+            if (
+                device["type"] == PSS_DEVICE_TYPE
+                and device["id"] not in coordinator.devices
+            ):
+                livisi_switch: SwitchEntity = create_entity(
+                    config_entry, device, coordinator
+                )
+                LOGGER.debug("Include device type: %s", device["type"])
+                coordinator.devices.add(device["id"])
+                entities.append(livisi_switch)
+        async_add_entities(entities)
+
+    config_entry.async_on_unload(
+        coordinator.async_add_listener(handle_coordinator_update)
+    )
+
+
+def create_entity(
+    config_entry: ConfigEntry,
+    device: dict[str, Any],
+    coordinator: LivisiDataUpdateCoordinator,
+) -> SwitchEntity:
+    """Create Switch Entity."""
+    config_details: dict[str, Any] = device["config"]
+    capabilities: list = device["capabilities"]
+    room_id: str = device["location"]
+    room_name: str = coordinator.rooms[room_id]
+    livisi_switch = LivisiSwitch(
+        config_entry,
+        coordinator,
+        unique_id=device["id"],
+        manufacturer=device["manufacturer"],
+        device_type=device["type"],
+        name=config_details["name"],
+        capability_id=capabilities[0],
+        room=room_name,
+    )
+    return livisi_switch
+
+
+class LivisiSwitch(CoordinatorEntity[LivisiDataUpdateCoordinator], SwitchEntity):
+    """Represents the Livisi Switch."""
+
+    def __init__(
+        self,
+        config_entry: ConfigEntry,
+        coordinator: LivisiDataUpdateCoordinator,
+        unique_id: str,
+        manufacturer: str,
+        device_type: str,
+        name: str,
+        capability_id: str,
+        room: str,
+    ) -> None:
+        """Initialize the Livisi Switch."""
+        self.config_entry = config_entry
+        self._attr_unique_id = unique_id
+        self._attr_name = name
+        self._capability_id = capability_id
+        self.aio_livisi = coordinator.aiolivisi
+        self._attr_available = False
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, unique_id)},
+            manufacturer=manufacturer,
+            model=device_type,
+            name=name,
+            suggested_area=room,
+            via_device=(DOMAIN, config_entry.entry_id),
+        )
+        super().__init__(coordinator)
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn the entity on."""
+        response = await self.aio_livisi.async_pss_set_state(
+            self._capability_id, is_on=True
+        )
+        if response is None:
+            self._attr_available = False
+            raise HomeAssistantError(f"Failed to turn on {self._attr_name}")
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn the entity off."""
+        response = await self.aio_livisi.async_pss_set_state(
+            self._capability_id, is_on=False
+        )
+        if response is None:
+            self._attr_available = False
+            raise HomeAssistantError(f"Failed to turn off {self._attr_name}")
+
+    async def async_added_to_hass(self) -> None:
+        """Register callbacks."""
+        response = await self.coordinator.async_get_pss_state(self._capability_id)
+        if response is None:
+            self._attr_is_on = False
+            self._attr_available = False
+        else:
+            self._attr_is_on = response
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{LIVISI_STATE_CHANGE}_{self._capability_id}",
+                self.update_states,
+            )
+        )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{LIVISI_REACHABILITY_CHANGE}_{self.unique_id}",
+                self.update_reachability,
+            )
+        )
+
+    @callback
+    def update_states(self, state: bool) -> None:
+        """Update the states of the switch device."""
+        self._attr_is_on = state
+        self.async_write_ha_state()
+
+    @callback
+    def update_reachability(self, is_reachable: bool) -> None:
+        """Update the reachability of the switch device."""
+        self._attr_available = is_reachable
+        self.async_write_ha_state()

homeassistant/components/livisi/translations/en.json

@@ -0,0 +1,18 @@
+{
+    "config": {
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "wrong_ip_address": "The IP address is incorrect or the SHC cannot be reached locally.",
+            "wrong_password": "The password is incorrect."
+        },
+        "step": {
+            "user": {
+                "data": {
+                    "host": "IP Address",
+                    "password": "Password"
+                },
+                "description": "Enter the IP address and the (local) password of the SHC."
+            }
+        }
+    }
+}

homeassistant/generated/config_flows.py

@@ -215,6 +215,7 @@ FLOWS = {
         "lifx",
         "litejet",
         "litterrobot",
+        "livisi",
         "local_ip",
         "locative",
         "logi_circle",

homeassistant/generated/integrations.json

@@ -2821,6 +2821,12 @@
       "config_flow": true,
       "iot_class": "cloud_push"
     },
+    "livisi": {
+      "name": "LIVISI Smart Home",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "llamalab_automate": {
       "name": "LlamaLab Automate",
       "integration_type": "hub",

requirements_all.txt

@@ -201,6 +201,9 @@ aiolifx_effects==0.3.0
 # homeassistant.components.lifx
 aiolifx_themes==0.2.0
 
+# homeassistant.components.livisi
+aiolivisi==0.0.14
+
 # homeassistant.components.lookin
 aiolookin==0.1.1
 

requirements_test_all.txt

@@ -179,6 +179,9 @@ aiolifx_effects==0.3.0
 # homeassistant.components.lifx
 aiolifx_themes==0.2.0
 
+# homeassistant.components.livisi
+aiolivisi==0.0.14
+
 # homeassistant.components.lookin
 aiolookin==0.1.1
 

tests/components/livisi/__init__.py

@@ -0,0 +1,37 @@
+"""Tests for the LIVISI Smart Home integration."""
+from unittest.mock import patch
+
+from homeassistant.components.livisi.const import CONF_HOST, CONF_PASSWORD
+
+VALID_CONFIG = {
+    CONF_HOST: "1.1.1.1",
+    CONF_PASSWORD: "test",
+}
+
+DEVICE_CONFIG = {
+    "serialNumber": "1234",
+    "controllerType": "Classic",
+}
+
+
+def mocked_livisi_login():
+    """Create mock for LIVISI login."""
+    return patch(
+        "homeassistant.components.livisi.config_flow.AioLivisi.async_set_token"
+    )
+
+
+def mocked_livisi_controller():
+    """Create mock data for LIVISI controller."""
+    return patch(
+        "homeassistant.components.livisi.config_flow.AioLivisi.async_get_controller",
+        return_value=DEVICE_CONFIG,
+    )
+
+
+def mocked_livisi_setup_entry():
+    """Create mock for LIVISI setup entry."""
+    return patch(
+        "homeassistant.components.livisi.async_setup_entry",
+        return_value=True,
+    )

tests/components/livisi/test_config_flow.py

@@ -0,0 +1,68 @@
+"""Test the Livisi Home Assistant config flow."""
+
+from unittest.mock import patch
+
+from aiolivisi import errors as livisi_errors
+import pytest
+
+from homeassistant import data_entry_flow
+from homeassistant.components.livisi.const import DOMAIN
+from homeassistant.config_entries import SOURCE_USER
+
+from . import (
+    VALID_CONFIG,
+    mocked_livisi_controller,
+    mocked_livisi_login,
+    mocked_livisi_setup_entry,
+)
+
+
+async def test_create_entry(hass):
+    """Test create LIVISI entity."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    with mocked_livisi_login(), mocked_livisi_controller(), mocked_livisi_setup_entry():
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            VALID_CONFIG,
+        )
+
+        assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
+        assert result["title"] == "SHC Classic"
+        assert result["data"]["host"] == "1.1.1.1"
+        assert result["data"]["password"] == "test"
+
+
+@pytest.mark.parametrize(
+    "exception,expected_reason",
+    [
+        (livisi_errors.ShcUnreachableException(), "cannot_connect"),
+        (livisi_errors.IncorrectIpAddressException(), "wrong_ip_address"),
+        (livisi_errors.WrongCredentialException(), "wrong_password"),
+    ],
+)
+async def test_create_entity_after_login_error(
+    hass, exception: livisi_errors.LivisiException, expected_reason: str
+):
+    """Test the LIVISI integration can create an entity after the user had login errors."""
+    with patch(
+        "homeassistant.components.livisi.config_flow.AioLivisi.async_set_token",
+        side_effect=exception,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": SOURCE_USER},
+        )
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], VALID_CONFIG
+        )
+        assert result["type"] == data_entry_flow.FlowResultType.FORM
+        assert result["errors"]["base"] == expected_reason
+    with mocked_livisi_login(), mocked_livisi_controller(), mocked_livisi_setup_entry():
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input=VALID_CONFIG,
+        )
+    assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY