Alarm support

.gitignore

@@ -1,2 +1,3 @@
 .DS_Store
 .vscode
+captured

README.md

@@ -1,8 +1,56 @@
 # Home Assistant Ksenia Lares integration
 
-[![hacs_badge](https://img.shields.io/badge/HACS-Custom-orange.svg)](https://github.com/custom-components/hacs)
+[![GitHub Activity][commits-shield]][commits]
+[![License][license-shield]](LICENSE)
 
-Ksenia Lares (pre 4.0) integration for home assistant. This is integration is in early stages, use at own risk.
+[![hacs][hacsbadge]][hacs]
+![Project Maintenance][maintenance-shield]
+
+Ksenia Lares 48IP integration for home assistant. This is integration is in early stages, use at own risk.
+
+**This integration will set up the following platforms.**
+
+Platform | Description
+-- | --
+`binary_sensor` | For each zone, defaults to movement sensor.
+`sensor` | For each partition, showing the ARM status. 
+`alarm_panel` | ARM and disarm based on partitions and scenarios
+
+## Installation
+### Installation via HACS
+1. Open HACS
+2. Goto the menu in to top right corner
+3. Goto custom respositories
+4. Add this repositry `https://github.com/johnnybegood/ha-ksenia-lares`
+5. Click _explore & download respositories_
+6. Search for and add "Ksenia Lares"
+7. Restart Home Assistant
+8. In the HA UI go to "Configuration" -> "Integrations" click "+" and search for "Ksenia Lares" or click the button below.
+
+[![Open your Home Assistant instance and start setting up a new integration.](https://my.home-assistant.io/badges/config_flow_start.svg)](https://my.home-assistant.io/redirect/config_flow_start/?domain=ksenia_lares)
+
+### Installation directly
+
+1. Using the tool of choice open the directory (folder) for your HA configuration (where you find `configuration.yaml`).
+2. If you do not have a `custom_components` directory (folder) there, you need to create it.
+3. In the `custom_components` directory (folder) create a new folder called `ksenia_lares`.
+4. Download _all_ the files from the `custom_components/ksenia_lares/` directory (folder) in this repository.
+5. Place the files you downloaded in the new directory (folder) you created.
+6. Restart Home Assistant
+7. In the HA UI go to "Configuration" -> "Integrations" click "+" and search for "Ksenia Lares" or click the button below.
+
+[![Open your Home Assistant instance and start setting up a new integration.](https://my.home-assistant.io/badges/config_flow_start.svg)](https://my.home-assistant.io/redirect/config_flow_start/?domain=ksenia_lares)
+
+## Quickstart
+1. Check your Ksenia Lares has the web interface running
+2. Enter the following information from Ksenia Lares 48IP web interface:
+   1. Hostname or IP adress (port 4202 will be used)
+   2. Username
+   3. Password
+3. To be able to use the alarm panel, click the 'configure' button on the Ksenia Lares [integration](https://my.home-assistant.io/redirect/integrations/)
+4.  Select the partitions and scenarios that match different ARM states.
+    1.  When the selected partitions are armed, the alarm panel will show the corresponding state.
+    2.  The selected scenario will be activated when changing arm state of the alarm panel
 
 ## WIP
 - [x] Detect zones from Lares alarm system and show as binary sensor 

custom_components/ksenia_lares/__init__.py

@@ -5,13 +5,14 @@
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
+from homeassistant.const import Platform
 
 from .base import LaresBase
 from .coordinator import LaresDataUpdateCoordinator
-from .const import DOMAIN
+from .const import DOMAIN, DATA_COORDINATOR, DATA_UPDATE_LISTENER
 
 CONFIG_SCHEMA = vol.Schema({DOMAIN: vol.Schema({})}, extra=vol.ALLOW_EXTRA)
-PLATFORMS = ["binary_sensor", "sensor"]
+PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR, Platform.ALARM_CONTROL_PANEL]
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
@@ -23,7 +24,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
     # Preload device info
     await client.device_info()
 
-    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
+    unsub_options_update_listener = entry.add_update_listener(options_update_listener)
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
+        DATA_COORDINATOR: coordinator,
+        DATA_UPDATE_LISTENER: unsub_options_update_listener,
+    }
 
     hass.async_create_task(
         hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@@ -32,6 +38,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
     return True
 
 
+async def options_update_listener(hass: HomeAssistant, config_entry: ConfigEntry):
+    """Handle options update."""
+    await hass.config_entries.async_reload(config_entry.entry_id)
+
+
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
     """Unload a config entry."""
     unload_ok = all(
@@ -43,4 +54,8 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
         )
     )
 
+    if unload_ok:
+        hass.data[DOMAIN][entry.entry_id][DATA_UPDATE_LISTENER]()
+        hass.data[DOMAIN].pop(entry.entry_id)
+
     return unload_ok

custom_components/ksenia_lares/alarm_control_panel.py

@@ -0,0 +1,221 @@
+"""This component provides support for Lares alarm  control panel."""
+import logging
+from datetime import timedelta
+
+from homeassistant.components.alarm_control_panel import (
+    AlarmControlPanelEntity,
+    AlarmControlPanelEntityFeature,
+    CodeFormat,
+)
+
+from homeassistant.helpers.typing import StateType
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from homeassistant.const import (
+    STATE_ALARM_ARMED_AWAY,
+    STATE_ALARM_ARMED_HOME,
+    STATE_ALARM_DISARMED,
+    STATE_ALARM_ARMED_NIGHT,
+    STATE_ALARM_ARMED_CUSTOM_BYPASS,
+    STATE_ALARM_ARMING,
+)
+
+from .coordinator import LaresDataUpdateCoordinator
+from .const import (
+    DOMAIN,
+    DATA_COORDINATOR,
+    DATA_PARTITIONS,
+    PARTITION_STATUS_ARMED,
+    PARTITION_STATUS_ARMED_IMMEDIATE,
+    PARTITION_STATUS_ARMING,
+    CONF_PARTITION_AWAY,
+    CONF_PARTITION_HOME,
+    CONF_PARTITION_NIGHT,
+    CONF_SCENARIO_AWAY,
+    CONF_SCENARIO_HOME,
+    CONF_SCENARIO_NIGHT,
+    CONF_SCENARIO_DISARM,
+)
+
+_LOGGER = logging.getLogger(__name__)
+SCAN_INTERVAL = timedelta(seconds=10)
+
+
+async def async_setup_entry(hass, config_entry, async_add_devices):
+    """Set up alarm control panel of the Lares alarm device from a config entry."""
+
+    coordinator = hass.data[DOMAIN][config_entry.entry_id][DATA_COORDINATOR]
+    device_info = await coordinator.client.device_info()
+    partition_descriptions = await coordinator.client.partition_descriptions()
+    scenario_descriptions = await coordinator.client.scenario_descriptions()
+
+    options = {
+        CONF_PARTITION_AWAY: config_entry.options.get(CONF_PARTITION_AWAY, []),
+        CONF_PARTITION_HOME: config_entry.options.get(CONF_PARTITION_HOME, []),
+        CONF_PARTITION_NIGHT: config_entry.options.get(CONF_PARTITION_NIGHT, []),
+        CONF_SCENARIO_NIGHT: config_entry.options.get(CONF_SCENARIO_NIGHT, []),
+        CONF_SCENARIO_HOME: config_entry.options.get(CONF_SCENARIO_HOME, []),
+        CONF_SCENARIO_AWAY: config_entry.options.get(CONF_SCENARIO_AWAY, []),
+        CONF_SCENARIO_DISARM: config_entry.options.get(CONF_SCENARIO_DISARM, []),
+    }
+
+    # Fetch initial data so we have data when entities subscribe
+    await coordinator.async_refresh()
+
+    async_add_devices(
+        [
+            LaresAlarmControlPanel(
+                coordinator,
+                device_info,
+                partition_descriptions,
+                scenario_descriptions,
+                options,
+            )
+        ]
+    )
+
+
+class LaresAlarmControlPanel(CoordinatorEntity, AlarmControlPanelEntity):
+    """An implementation of a Lares alarm control panel."""
+
+    TYPE = DOMAIN
+    ARMED_STATUS = [PARTITION_STATUS_ARMED, PARTITION_STATUS_ARMED_IMMEDIATE]
+
+    def __init__(
+        self,
+        coordinator: LaresDataUpdateCoordinator,
+        device_info: dict,
+        partition_descriptions: dict,
+        scenario_descriptions: dict,
+        options: dict,
+    ) -> None:
+        """Initialize a the switch."""
+        super().__init__(coordinator)
+
+        self._coordinator = coordinator
+        self._partition_descriptions = partition_descriptions
+        self._scenario_descriptions = scenario_descriptions
+        self._options = options
+        self._attr_code_format = CodeFormat.NUMBER
+        self._attr_device_info = device_info
+        self._attr_code_arm_required = True
+
+    @property
+    def unique_id(self) -> str:
+        """Return the unique ID for this entity."""
+        name = self._attr_device_info["name"].replace(" ", "_")
+        return f"lares_panel_{name}"
+
+    @property
+    def name(self):
+        """Return the name of this panel."""
+        name = self._attr_device_info["name"]
+        return f"Panel {name}"
+
+    @property
+    def supported_features(self) -> AlarmControlPanelEntityFeature:
+        """Return the list of supported features."""
+        supported_features = AlarmControlPanelEntityFeature(0)
+
+        if self._options[CONF_SCENARIO_AWAY] != "":
+            supported_features |= AlarmControlPanelEntityFeature.ARM_AWAY
+
+        if self._options[CONF_SCENARIO_HOME] != "":
+            supported_features |= AlarmControlPanelEntityFeature.ARM_HOME
+
+        if self._options[CONF_SCENARIO_NIGHT] != "":
+            supported_features |= AlarmControlPanelEntityFeature.ARM_NIGHT
+
+        return supported_features
+
+    @property
+    def state(self) -> StateType:
+        """Return the state of this panel."""
+        if self.__has_partition_with_status(PARTITION_STATUS_ARMING):
+            return STATE_ALARM_ARMING
+
+        if self.__is_armed(CONF_PARTITION_AWAY):
+            return STATE_ALARM_ARMED_AWAY
+
+        if self.__is_armed(CONF_PARTITION_HOME):
+            return STATE_ALARM_ARMED_HOME
+
+        if self.__is_armed(CONF_PARTITION_NIGHT):
+            return STATE_ALARM_ARMED_NIGHT
+
+        # If any of the not mapped partitions is armed, show custom as fallback
+        if self.__has_partition_with_status(self.ARMED_STATUS):
+            return STATE_ALARM_ARMED_CUSTOM_BYPASS
+
+        return STATE_ALARM_DISARMED
+
+    async def async_alarm_arm_home(self, code: str | None = None) -> None:
+        """Send arm home command."""
+        await self.__command(CONF_SCENARIO_HOME, code)
+
+    async def async_alarm_arm_away(self, code: str | None = None) -> None:
+        """Send arm home command."""
+        await self.__command(CONF_SCENARIO_AWAY, code)
+
+    async def async_alarm_arm_night(self, code: str | None = None) -> None:
+        """Send arm home command."""
+        await self.__command(CONF_SCENARIO_NIGHT, code)
+
+    async def async_alarm_disarm(self, code: str | None = None) -> None:
+        """Send disarm command."""
+        await self.__command(CONF_SCENARIO_DISARM, code)
+
+    def __has_partition_with_status(self, status_list: list[str]) -> bool:
+        """Return if any partitions is arming."""
+        partitions = enumerate(self._coordinator.data[DATA_PARTITIONS])
+        in_state = list(
+            idx for idx, partition in partitions if partition["status"] in status_list
+        )
+
+        _LOGGER.debug("%s in status %s", in_state, status_list)
+
+        return len(in_state) > 0
+
+    def __is_armed(self, key: str) -> bool:
+        """Return if all partitions linked to the configuration key are armed."""
+        partition_names = self._options[key]
+
+        # Skip the check if no partitions are linked
+        if len(partition_names) == 0:
+            _LOGGER.debug("Skipping %s armed check, no definition", key)
+            return False
+
+        descriptions = enumerate(self._partition_descriptions)
+        to_check = (idx for idx, name in descriptions if name in partition_names)
+
+        _LOGGER.debug("Checking %s (%s) for %s", partition_names, to_check, key)
+
+        for idx in to_check:
+            if (
+                self._coordinator.data[DATA_PARTITIONS][idx]["status"]
+                not in self.ARMED_STATUS
+            ):
+                return False
+
+        return True
+
+    async def __command(self, key: str, code: str | None = None) -> None:
+        """Send arm home command."""
+        scenario_name = self._options[key]
+
+        if scenario_name is None:
+            _LOGGER.warning("Skipping command, no definition for %s", key)
+            return
+
+        descriptions = enumerate(self._scenario_descriptions)
+        match_gen = (idx for idx, name in descriptions if name == scenario_name)
+        matches = list(match_gen)
+
+        if len(matches) != 1:
+            _LOGGER.error("No match for %s (%s found)", key, len(matches))
+            return
+
+        scenario = matches[0]
+        _LOGGER.debug("Activating scenario %s", scenario)
+
+        await self._coordinator.client.activate_scenario(scenario, code)