Add livisi window sensor (WDS) (#90220)

* Added support for livisi window sensor

* Add const strings

* added postpix for device_id

* Remove unnecessary import

* Fix imports

* Fix lint errors, remove redundant device class property

* Format code

* Update .coveragerc

* Finish basic window door sensor support

* currently, only one binary sensor (wds) is supported

* Remove unused imports

* Fix isort issue

* Simplify code as suggested in PR

* rename get_device_response to get_device_state

* fix ruff issue

* Be more defensive in interpreting what we get from aiolivisi

* Simplify coordinator

* remove window sensor specific code (isOpen)

* parameter order, type hinta

* Update homeassistant/components/livisi/coordinator.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/livisi/coordinator.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/livisi/coordinator.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/livisi/binary_sensor.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/livisi/binary_sensor.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/livisi/binary_sensor.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

---------

Co-authored-by: Tecotix <78791840+Tecotix@users.noreply.github.com>
Co-authored-by: Erik Montnemery <erik@montnemery.com>
Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

.coveragerc

@@ -642,6 +642,7 @@ omit =
     homeassistant/components/linux_battery/sensor.py
     homeassistant/components/lirc/*
     homeassistant/components/livisi/__init__.py
+    homeassistant/components/livisi/binary_sensor.py
     homeassistant/components/livisi/climate.py
     homeassistant/components/livisi/coordinator.py
     homeassistant/components/livisi/entity.py

homeassistant/components/livisi/__init__.py

@@ -16,7 +16,7 @@ from homeassistant.helpers import aiohttp_client, device_registry as dr
 from .const import DOMAIN
 from .coordinator import LivisiDataUpdateCoordinator
 
-PLATFORMS: Final = [Platform.CLIMATE, Platform.SWITCH]
+PLATFORMS: Final = [Platform.BINARY_SENSOR, Platform.CLIMATE, Platform.SWITCH]
 
 
 async def async_setup_entry(hass: core.HomeAssistant, entry: ConfigEntry) -> bool:

homeassistant/components/livisi/binary_sensor.py

@@ -0,0 +1,110 @@
+"""Code to handle a Livisi Binary Sensor."""
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN, LIVISI_STATE_CHANGE, LOGGER, WDS_DEVICE_TYPE
+from .coordinator import LivisiDataUpdateCoordinator
+from .entity import LivisiEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up binary_sensor device."""
+    coordinator: LivisiDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+    known_devices = set()
+
+    @callback
+    def handle_coordinator_update() -> None:
+        """Add Window Sensor."""
+        shc_devices: list[dict[str, Any]] = coordinator.data
+        entities: list[BinarySensorEntity] = []
+        for device in shc_devices:
+            if device["id"] not in known_devices and device["type"] == WDS_DEVICE_TYPE:
+                livisi_binary: BinarySensorEntity = LivisiWindowDoorSensor(
+                    config_entry, coordinator, device
+                )
+                LOGGER.debug("Include device type: %s", device["type"])
+                coordinator.devices.add(device["id"])
+                known_devices.add(device["id"])
+                entities.append(livisi_binary)
+        async_add_entities(entities)
+
+    config_entry.async_on_unload(
+        coordinator.async_add_listener(handle_coordinator_update)
+    )
+
+
+class LivisiBinarySensor(LivisiEntity, BinarySensorEntity):
+    """Represents a Livisi Binary Sensor."""
+
+    def __init__(
+        self,
+        config_entry: ConfigEntry,
+        coordinator: LivisiDataUpdateCoordinator,
+        device: dict[str, Any],
+        capability_name: str,
+    ) -> None:
+        """Initialize the Livisi sensor."""
+        super().__init__(config_entry, coordinator, device)
+        self._capability_id = self.capabilities[capability_name]
+
+    async def async_added_to_hass(self) -> None:
+        """Register callbacks."""
+        await super().async_added_to_hass()
+
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                f"{LIVISI_STATE_CHANGE}_{self._capability_id}",
+                self.update_states,
+            )
+        )
+
+    @callback
+    def update_states(self, state: bool) -> None:
+        """Update the state of the device."""
+        self._attr_is_on = state
+        self.async_write_ha_state()
+
+
+class LivisiWindowDoorSensor(LivisiBinarySensor):
+    """Represents a Livisi Window/Door Sensor as a Binary Sensor Entity."""
+
+    def __init__(
+        self,
+        config_entry: ConfigEntry,
+        coordinator: LivisiDataUpdateCoordinator,
+        device: dict[str, Any],
+    ) -> None:
+        """Initialize the Livisi window/door sensor."""
+        super().__init__(config_entry, coordinator, device, "WindowDoorSensor")
+
+        self._attr_device_class = (
+            BinarySensorDeviceClass.DOOR
+            if (device.get("tags", {}).get("typeCategory") == "TCDoorId")
+            else BinarySensorDeviceClass.WINDOW
+        )
+
+    async def async_added_to_hass(self) -> None:
+        """Get current state."""
+        await super().async_added_to_hass()
+        response = await self.coordinator.async_get_device_state(
+            self._capability_id, "isOpen"
+        )
+        if response is None:
+            self._attr_available = False
+        else:
+            self._attr_is_on = response

homeassistant/components/livisi/const.py

@@ -16,6 +16,8 @@ LIVISI_REACHABILITY_CHANGE: Final = "livisi_reachability_change"
 
 SWITCH_DEVICE_TYPES: Final = ["ISS", "ISS2", "PSS", "PSSO"]
 VRCC_DEVICE_TYPE: Final = "VRCC"
+WDS_DEVICE_TYPE: Final = "WDS"
+
 
 MAX_TEMPERATURE: Final = 30.0
 MIN_TEMPERATURE: Final = 6.0

homeassistant/components/livisi/coordinator.py

@@ -115,6 +115,9 @@ class LivisiDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]):
         self._async_dispatcher_send(
             LIVISI_REACHABILITY_CHANGE, event_data.source, event_data.isReachable
         )
+        self._async_dispatcher_send(
+            LIVISI_STATE_CHANGE, event_data.source, event_data.isOpen
+        )
 
     async def on_close(self) -> None:
         """Define a handler to fire when the websocket is closed."""