Support for multiple areas (#22)

* feat: implementation of multiple areas #7
- subsystem arm/disarm
- allow opt-in using subsystems
- README.md update

README.md

@@ -4,6 +4,7 @@ HACS repository of Hikvision Ax Pro integration for home assistant
 [![hacs_badge](https://img.shields.io/badge/HACS-Custom-41BDF5.svg)](https://github.com/hacs/integration)
 
 ## Support
+- Sub zones control (Opt-in - after configuration reload integration)
 
 ### Supported Sensors / Detectors
 - Wireless external magnetic sensors

custom_components/hikvision_axpro/__init__.py

@@ -73,7 +73,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     code = entry.data[CONF_CODE]
     use_code_arming = entry.data[USE_CODE_ARMING]
     axpro = hikaxpro.HikAxPro(host, username, password)
-    update_interval = entry.data.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL.total_seconds())
+    update_interval: float = entry.data.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL.total_seconds())
 
     try:
         async with timeout(10):
@@ -125,6 +125,7 @@ class HikAxProDataUpdateCoordinator(DataUpdateCoordinator):
     device_info: Optional[dict] = None
     device_model: Optional[str] = None
     device_name: Optional[str] = None
+    sub_systems: list[SubSys] = []
 
     def __init__(
         self,
@@ -135,7 +136,7 @@ def __init__(
         code_format,
         use_code_arming,
         code,
-        update_interval
+        update_interval: float
     ):
         self.axpro = axpro
         self.state = None
@@ -162,6 +163,7 @@ def init_device(self):
         self.device_name = self.device_info['DeviceInfo']['deviceName']
         self.device_model = self.device_info['DeviceInfo']['model']
         _LOGGER.debug(self.device_info)
+        self._update_data()
 
     def _update_data(self) -> None:
         """Fetch data from axpro via sync functions."""
@@ -175,8 +177,9 @@ def _update_data(self) -> None:
                 for sublist in subsys_resp.sub_sys_list:
                     subsys_arr.append(sublist.sub_sys)
             func: Callable[[SubSys], bool] = lambda n: n.enabled
-            subsys_resp: list[SubSys] = filter(func, subsys_arr)
-            for subsys in subsys_resp:
+            subsys_arr = list(filter(func, subsys_arr))
+            self.sub_systems = subsys_arr
+            for subsys in subsys_arr:
                 if subsys.alarm:
                     status = STATE_ALARM_TRIGGERED
                     break
@@ -211,26 +214,65 @@ async def _async_update_data(self) -> None:
         except ConnectionError as error:
             raise UpdateFailed(error) from error
 
-    async def async_arm_home(self):
+    async def async_arm_home(self, sub_id: Optional[int] = None):
         """Arm alarm panel in home state."""
-        is_success = await self.hass.async_add_executor_job(self.axpro.arm_home)
+        # TODO modify AXPRO
+        if sub_id is not None:
+            is_success = await self.hass.async_add_executor_job(self._arm_home(sub_id=sub_id))
+        else:
+            is_success = await self.hass.async_add_executor_job(self.axpro.arm_home)
 
         if is_success:
             await self._async_update_data()
             await self.async_request_refresh()
 
-    async def async_arm_away(self):
+    async def async_arm_away(self, sub_id: Optional[int] = None):
         """Arm alarm panel in away state"""
-        is_success = await self.hass.async_add_executor_job(self.axpro.arm_away)
+        # TODO modify AXPRO
+        if sub_id is not None:
+            is_success = await self.hass.async_add_executor_job(self._arm_away(sub_id=sub_id))
+        else:
+            is_success = await self.hass.async_add_executor_job(self.axpro.arm_away)
 
         if is_success:
             await self._async_update_data()
             await self.async_request_refresh()
 
-    async def async_disarm(self):
+    async def async_disarm(self, sub_id: Optional[int] = None):
         """Disarm alarm control panel."""
-        is_success = await self.hass.async_add_executor_job(self.axpro.disarm)
+        # TODO modify AXPRO
+        if sub_id is not None:
+            is_success = await self.hass.async_add_executor_job(self._disarm(sub_id=sub_id))
+        else:
+            is_success = await self.hass.async_add_executor_job(self.axpro.disarm)
 
         if is_success:
             await self._async_update_data()
             await self.async_request_refresh()
+
+    def _arm_home(self, sub_id: int):
+        endpoint = self.axpro.buildUrl(f"http://{self.host}{hikaxpro.consts.Endpoints.Alarm_ArmHome.replace('0xffffffff', str(sub_id))}", True)
+        response = self.axpro.makeRequest(endpoint, hikaxpro.consts.Method.PUT)
+
+        if response.status_code != 200:
+            raise hikaxpro.errors.UnexpectedResponseCodeError(response.status_code, response.text)
+
+        return response.status_code == 200
+
+    def _arm_away(self, sub_id: int):
+        endpoint = self.axpro.buildUrl(f"http://{self.host}{hikaxpro.consts.Endpoints.Alarm_ArmAway.replace('0xffffffff', str(sub_id))}", True)
+        response = self.axpro.makeRequest(endpoint, hikaxpro.consts.Method.PUT)
+
+        if response.status_code != 200:
+            raise hikaxpro.errors.UnexpectedResponseCodeError(response.status_code, response.text)
+
+        return response.status_code == 200
+
+    def _disarm(self, sub_id: int):
+        endpoint = self.axpro.buildUrl(f"http://{self.host}{hikaxpro.consts.Endpoints.Alarm_Disarm.replace('0xffffffff', str(sub_id))}", True)
+        response = self.axpro.makeRequest(endpoint, hikaxpro.consts.Method.PUT)
+
+        if response.status_code != 200:
+            raise hikaxpro.errors.UnexpectedResponseCodeError(response.status_code, response.text)
+
+        return response.status_code == 200

custom_components/hikvision_axpro/alarm_control_panel.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import STATE_ALARM_TRIGGERED, STATE_ALARM_ARMED_AWAY, STATE_ALARM_ARMED_HOME, \
+    STATE_ALARM_ARMED_VACATION, STATE_ALARM_DISARMED
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity import DeviceInfo
@@ -11,16 +13,33 @@
     AlarmControlPanelEntityFeature,
     CodeFormat,
 )
+from homeassistant.helpers import device_registry as dr
 
-from .const import DATA_COORDINATOR, DOMAIN
+from . import HikAxProDataUpdateCoordinator, SubSys, Arming
+from .const import DATA_COORDINATOR, DOMAIN, ALLOW_SUBSYSTEMS
 
 
 async def async_setup_entry(
     hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
 ) -> None:
     """Set up a Hikvision ax pro alarm control panel based on a config entry."""
-    coordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR]
-    async_add_entities([HikAxProPanel(coordinator)], False)
+    coordinator: HikAxProDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR]
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        # connections={},
+        identifiers={(DOMAIN, coordinator.mac)},
+        manufacturer="HikVision" if coordinator.device_model is not None else "Unknown",
+        # suggested_area=zone.zone.,
+        name=coordinator.device_name,
+        #via_device=(DOMAIN, str(coordinator.mac)),
+        model=coordinator.device_model,
+    )
+    panels = [HikAxProPanel(coordinator)]
+    if bool(entry.data.get(ALLOW_SUBSYSTEMS, False)):
+        for sub_system in coordinator.sub_systems:
+            panels.append(HikAxProSubPanel(coordinator, sub_system))
+    async_add_entities(panels, False)
 
 
 class HikAxProPanel(CoordinatorEntity, AlarmControlPanelEntity):
@@ -106,3 +125,105 @@ async def async_alarm_arm_away(self, code=None):
 
     def __is_code_valid(self, code):
         return code == self.coordinator.code
+
+
+class HikAxProSubPanel(CoordinatorEntity, AlarmControlPanelEntity):
+    """Representation of Hikvision Ax Pro alarm panel."""
+    sys: SubSys
+
+    def __init__(self, coordinator: CoordinatorEntity, sys: SubSys):
+        self.sys = sys
+        super().__init__(coordinator=coordinator)
+
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self.async_write_ha_state()
+
+    _attr_supported_features = (
+        AlarmControlPanelEntityFeature.ARM_HOME
+        | AlarmControlPanelEntityFeature.ARM_AWAY
+    )
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return device info for this device."""
+        return DeviceInfo(
+            identifiers={(DOMAIN, self.coordinator.mac)},
+            manufacturer="Hikvision - Ax Pro",
+            model=self.coordinator.device_model,
+            name=self.coordinator.device_name,
+        )
+
+    @property
+    def unique_id(self):
+        """Return a unique id."""
+        return "subsys-" + str(self.sys.id)
+
+    @property
+    def name(self):
+        """Return the name."""
+        return self.sys.name
+        # "HikvisionAxPro"
+
+    @property
+    def state(self):
+        """Return the state of the device."""
+        if self.sys.alarm:
+            return STATE_ALARM_TRIGGERED
+        if self.sys.arming == Arming.AWAY:
+            return STATE_ALARM_ARMED_AWAY
+        if self.sys.arming == Arming.STAY:
+            return STATE_ALARM_ARMED_HOME
+        if self.sys.arming == Arming.VACATION:
+            return STATE_ALARM_ARMED_VACATION
+        if self.sys.arming == Arming.DISARM:
+            return STATE_ALARM_DISARMED
+        return None
+
+    @property
+    def code_format(self) -> CodeFormat | None:
+        """Return the code format."""
+        return self.__get_code_format(self.coordinator.code_format)
+
+    def __get_code_format(self, code_format_str) -> CodeFormat:
+        """Returns CodeFormat according to the given code format string."""
+        code_format: CodeFormat = None
+
+        if not self.coordinator.use_code:
+            code_format = None
+        elif code_format_str == "NUMBER":
+            code_format = CodeFormat.NUMBER
+        elif code_format_str == "TEXT":
+            code_format = CodeFormat.TEXT
+
+        return code_format
+
+    async def async_alarm_disarm(self, code=None):
+        """Send disarm command."""
+        if self.coordinator.use_code:
+            if not self.__is_code_valid(code):
+                return
+
+        await self.coordinator.async_disarm(self.sys.id)
+
+    async def async_alarm_arm_home(self, code=None):
+        """Send arm home command."""
+        if self.coordinator.use_code and self.coordinator.use_code_arming:
+            if not self.__is_code_valid(code):
+                return
+
+        await self.coordinator.async_arm_home(self.sys.id)
+
+    async def async_alarm_arm_away(self, code=None):
+        """Send arm away command."""
+        if self.coordinator.use_code and self.coordinator.use_code_arming:
+            if not self.__is_code_valid(code):
+                return
+
+        await self.coordinator.async_arm_away(self.sys.id)
+
+    def __is_code_valid(self, code):
+        return code == self.coordinator.code
+

custom_components/hikvision_axpro/config_flow.py

@@ -21,7 +21,7 @@
 )
 from homeassistant.components.alarm_control_panel import SCAN_INTERVAL
 
-from .const import DOMAIN, USE_CODE_ARMING
+from .const import DOMAIN, USE_CODE_ARMING, ALLOW_SUBSYSTEMS
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -34,7 +34,8 @@
         vol.Optional(ATTR_CODE_FORMAT, default="NUMBER"): vol.In(["TEXT", "NUMBER"]),
         vol.Optional(CONF_CODE, default=""): str,
         vol.Optional(USE_CODE_ARMING, default=False): bool,
-        vol.Required(CONF_SCAN_INTERVAL, default=SCAN_INTERVAL.total_seconds()): int
+        vol.Required(CONF_SCAN_INTERVAL, default=SCAN_INTERVAL.total_seconds()): int,
+        vol.Optional(ALLOW_SUBSYSTEMS, default=False): bool,
     }
 )
 
@@ -48,7 +49,8 @@
         vol.Optional(ATTR_CODE_FORMAT, default="NUMBER"): vol.In(["TEXT", "NUMBER"]),
         vol.Optional(CONF_CODE, default=""): str,
         vol.Optional(USE_CODE_ARMING, default=False): bool,
-        vol.Required(CONF_SCAN_INTERVAL, default=SCAN_INTERVAL.total_seconds()): int
+        vol.Required(CONF_SCAN_INTERVAL, default=SCAN_INTERVAL.total_seconds()): int,
+        vol.Optional(ALLOW_SUBSYSTEMS, default=False): bool,
     }
 )
 

custom_components/hikvision_axpro/const.py

@@ -8,3 +8,5 @@
 DATA_COORDINATOR: Final[str] = "hikaxpro"
 
 USE_CODE_ARMING: Final[str] = "use_code_arming"
+
+ALLOW_SUBSYSTEMS: Final[str] = "allow_subsystems"

custom_components/hikvision_axpro/translations/cs.json

@@ -20,6 +20,7 @@
                     "enabled": "Kód pro odjištění",
                     "code_format": "Formát kódu (TEXT nebo NUMBER)",
                     "use_code_arming": "Kód pro zajištění",
+                    "allow_subsystems": "Zahrnutí zón jako samostatných oblastí pro zapnutí/vypnutí alarmů",
                     "code": "Kód",
                     "scan_interval": "Aktualizační interval"
                 }
@@ -36,6 +37,7 @@
                     "enabled": "Kód pro odjištění",
                     "code_format": "Formát kódu (TEXT nebo NUMBER)",
                     "use_code_arming": "Kód pro zajištění",
+                    "allow_subsystems": "Zahrnutí zón jako samostatných oblastí pro zapnutí/vypnutí alarmů",
                     "code": "Kód",
                     "scan_interval": "Aktualizační interval"
                 }

custom_components/hikvision_axpro/translations/en.json

@@ -20,6 +20,7 @@
                     "enabled": "Use code to disarm",
                     "code_format": "Code format (TEXT or NUMBER)",
                     "use_code_arming": "Use code for arming",
+                    "allow_subsystems": "Include areas as separate zones for arm/disarm",
                     "code": "Code",
                     "scan_interval": "Pull interval from system"
                 }
@@ -36,6 +37,7 @@
                     "enabled": "Use code to disarm",
                     "code_format": "Code format (TEXT or NUMBER)",
                     "use_code_arming": "Use code for arming",
+                    "allow_subsystems": "Include areas as separate zones for arm/disarm",
                     "code": "Code",
                     "scan_interval": "Pull interval from system"
                 }

custom_components/hikvision_axpro/translations/sl.json

@@ -20,6 +20,7 @@
                     "enabled": "Uporabi kodo za izklop",
                     "code_format": "Oblika kode (TEXT ali NUMBER)",
                     "use_code_arming": "Uporabi kodo za vklop",
+                    "allow_subsystems": "Vključite območja kot ločena območja za vklop/izklop alarmiranja",
                     "code": "Koda",
                     "scan_interval": "Interval preverjanja sistema"
                 }
@@ -36,6 +37,7 @@
                     "enabled": "Uporabi kodo za izklop",
                     "code_format": "Oblika kode (TEXT ali NUMBER)",
                     "use_code_arming": "Uporabi kodo za vklop",
+                    "allow_subsystems": "Vključite območja kot ločena območja za vklop/izklop alarmiranja",
                     "code": "Koda",
                     "scan_interval": "Interval preverjanja sistema"
                 }