Add alarm control panel platform to NASweb integration (#141582)

Co-authored-by: Erik Montnemery <[email protected]>

Author: nasWebio

homeassistant/components/nasweb/__init__.py

@@ -19,7 +19,11 @@
 from .coordinator import NASwebCoordinator
 from .nasweb_data import NASwebData
 
-PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.SWITCH]
+PLATFORMS: list[Platform] = [
+    Platform.ALARM_CONTROL_PANEL,
+    Platform.SENSOR,
+    Platform.SWITCH,
+]
 
 NASWEB_CONFIG_URL = "https://{host}/page"
 

homeassistant/components/nasweb/alarm_control_panel.py

@@ -0,0 +1,154 @@
+"""Platform for NASweb alarms."""
+
+from __future__ import annotations
+
+import logging
+import time
+
+from webio_api import Zone as NASwebZone
+from webio_api.const import STATE_ZONE_ALARM, STATE_ZONE_ARMED, STATE_ZONE_DISARMED
+
+from homeassistant.components.alarm_control_panel import (
+    DOMAIN as DOMAIN_ALARM_CONTROL_PANEL,
+    AlarmControlPanelEntity,
+    AlarmControlPanelEntityFeature,
+    AlarmControlPanelState,
+    CodeFormat,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+import homeassistant.helpers.entity_registry as er
+from homeassistant.helpers.typing import DiscoveryInfoType
+from homeassistant.helpers.update_coordinator import (
+    BaseCoordinatorEntity,
+    BaseDataUpdateCoordinatorProtocol,
+)
+
+from . import NASwebConfigEntry
+from .const import DOMAIN, STATUS_UPDATE_MAX_TIME_INTERVAL
+
+_LOGGER = logging.getLogger(__name__)
+ALARM_CONTROL_PANEL_TRANSLATION_KEY = "zone"
+
+NASWEB_STATE_TO_HA_STATE = {
+    STATE_ZONE_ALARM: AlarmControlPanelState.TRIGGERED,
+    STATE_ZONE_ARMED: AlarmControlPanelState.ARMED_AWAY,
+    STATE_ZONE_DISARMED: AlarmControlPanelState.DISARMED,
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config: NASwebConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> None:
+    """Set up alarm control panel platform."""
+    coordinator = config.runtime_data
+    current_zones: set[int] = set()
+
+    @callback
+    def _check_entities() -> None:
+        received_zones: dict[int, NASwebZone] = {
+            entry.index: entry for entry in coordinator.webio_api.zones
+        }
+        added = {i for i in received_zones if i not in current_zones}
+        removed = {i for i in current_zones if i not in received_zones}
+        entities_to_add: list[ZoneEntity] = []
+        for index in added:
+            webio_zone = received_zones[index]
+            if not isinstance(webio_zone, NASwebZone):
+                _LOGGER.error("Cannot create ZoneEntity without NASwebZone")
+                continue
+            new_zone = ZoneEntity(coordinator, webio_zone)
+            entities_to_add.append(new_zone)
+            current_zones.add(index)
+        async_add_entities(entities_to_add)
+        entity_registry = er.async_get(hass)
+        for index in removed:
+            unique_id = f"{DOMAIN}.{config.unique_id}.zone.{index}"
+            if entity_id := entity_registry.async_get_entity_id(
+                DOMAIN_ALARM_CONTROL_PANEL, DOMAIN, unique_id
+            ):
+                entity_registry.async_remove(entity_id)
+                current_zones.remove(index)
+            else:
+                _LOGGER.warning("Failed to remove old zone: no entity_id")
+
+    coordinator.async_add_listener(_check_entities)
+    _check_entities()
+
+
+class ZoneEntity(AlarmControlPanelEntity, BaseCoordinatorEntity):
+    """Entity representing NASweb zone."""
+
+    _attr_has_entity_name = True
+    _attr_should_poll = False
+    _attr_translation_key = ALARM_CONTROL_PANEL_TRANSLATION_KEY
+
+    def __init__(
+        self, coordinator: BaseDataUpdateCoordinatorProtocol, nasweb_zone: NASwebZone
+    ) -> None:
+        """Initialize zone entity."""
+        super().__init__(coordinator)
+        self._zone = nasweb_zone
+        self._attr_name = nasweb_zone.name
+        self._attr_translation_placeholders = {"index": f"{nasweb_zone.index:2d}"}
+        self._attr_unique_id = (
+            f"{DOMAIN}.{self._zone.webio_serial}.zone.{self._zone.index}"
+        )
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, self._zone.webio_serial)},
+        )
+
+    async def async_added_to_hass(self) -> None:
+        """When entity is added to hass."""
+        await super().async_added_to_hass()
+        self._handle_coordinator_update()
+
+    def _set_attr_available(
+        self, entity_last_update: float, available: bool | None
+    ) -> None:
+        if (
+            self.coordinator.last_update is None
+            or time.time() - entity_last_update >= STATUS_UPDATE_MAX_TIME_INTERVAL
+        ):
+            self._attr_available = False
+        else:
+            self._attr_available = available if available is not None else False
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._attr_alarm_state = NASWEB_STATE_TO_HA_STATE[self._zone.state]
+        if self._zone.pass_type == 0:
+            self._attr_code_format = CodeFormat.TEXT
+        elif self._zone.pass_type == 1:
+            self._attr_code_format = CodeFormat.NUMBER
+        else:
+            self._attr_code_format = None
+        self._attr_code_arm_required = self._attr_code_format is not None
+
+        self._set_attr_available(self._zone.last_update, self._zone.available)
+        self.async_write_ha_state()
+
+    async def async_update(self) -> None:
+        """Update the entity.
+
+        Only used by the generic entity update service.
+        Scheduling updates is not necessary, the coordinator takes care of updates via push notifications.
+        """
+
+    @property
+    def supported_features(self) -> AlarmControlPanelEntityFeature:
+        """Return the list of supported features."""
+        return AlarmControlPanelEntityFeature.ARM_AWAY
+
+    async def async_alarm_arm_away(self, code: str | None = None) -> None:
+        """Arm away ZoneEntity."""
+        await self._zone.arm(code)
+
+    async def async_alarm_disarm(self, code: str | None = None) -> None:
+        """Disarm ZoneEntity."""
+        await self._zone.disarm(code)

homeassistant/components/nasweb/coordinator.py

@@ -23,6 +23,7 @@
 
 KEY_INPUTS = "inputs"
 KEY_OUTPUTS = "outputs"
+KEY_ZONES = "zones"
 
 
 class NotificationCoordinator:
@@ -103,6 +104,7 @@ def __init__(
             KEY_OUTPUTS: self.webio_api.outputs,
             KEY_INPUTS: self.webio_api.inputs,
             KEY_TEMP_SENSOR: self.webio_api.temp_sensor,
+            KEY_ZONES: self.webio_api.zones,
         }
         self.async_set_updated_data(data)
 
@@ -197,5 +199,6 @@ async def process_status_update(self, new_status: dict) -> None:
             KEY_OUTPUTS: self.webio_api.outputs,
             KEY_INPUTS: self.webio_api.inputs,
             KEY_TEMP_SENSOR: self.webio_api.temp_sensor,
+            KEY_ZONES: self.webio_api.zones,
         }
         self.async_set_updated_data(new_data)

homeassistant/components/nasweb/strings.json

@@ -24,6 +24,11 @@
     }
   },
   "entity": {
+    "alarm_control_panel": {
+      "zone": {
+        "name": "Zone {index}"
+      }
+    },
     "sensor": {
       "sensor_input": {
         "name": "Input {index}",