added extra com ports for indoor and relay when using secure door module

pergolafabio · Aug 24, 2023 · 2b48cf5 · 2b48cf5
1 parent 19ffc20
commit 2b48cf5
Show file tree

Hide file tree

Showing 6 changed files with 123 additions and 39 deletions.
diff --git a/doorbell_beta/CHANGELOG.md b/doorbell_beta/CHANGELOG.md
@@ -1,5 +1,12 @@
 # Changelog
 
+## 3.0.0-beta.33 - 2023-08-24
+
+### Added
+
+- Added the 2 com ports for indoor station as a switch, have fun with it :-)
+- Optional: When you have a Secure Door Control Module(DS-K2M061), door needs to be unlocked from indoor panel, when you configyre "output_relays: 1" in the indoor config, it will create an extra switch for it
+
 ## 3.0.0-beta.32 - 2023-08-22
 
 ### Changed

diff --git a/doorbell_beta/DOCS.md b/doorbell_beta/DOCS.md
@@ -17,7 +17,7 @@ For each of your doorbells, repeat the following configuration:
 | port          | 8000          | (Optional) Port of the doorbell
 | username      | admin         | Username to access the doorbell
 | password      |               | Password to access the doorbell
-| output_relays |               | (optional) Set this option if you don't see the correct number of door switches inside HA
+| output_relays |               | (optional) Set this option if you don't see the correct number of door switche or if you have attached an secure door control module on your indoor
 | caller_info   | false         | (optional) If you want to see the button prssed as an atribute on the call sensor, usefull for intercoms with multiple buttons.
 | scenes        | false         | (optional) Extra Scene buttons for indoor panels 
 
@@ -33,12 +33,12 @@ The following configuration setups two doorbells, named `Front door` and `Rear d
   ip: 192.168.0.2
   username: admin
   password: password
-  caller_info: true
 
 - name: "Indoor"
   ip: 192.168.0.3
   username: admin
   password: password
+  caller_info: true
   scenes: true
 
 - name: "Indoor Extension"

diff --git a/doorbell_beta/config.yaml b/doorbell_beta/config.yaml
@@ -1,6 +1,6 @@
 name: Hikvision Doorbell (Beta)
 description: Connect your Hikvision doorbells to Home Assistant (Beta version)
-version: '3.0.0-beta.32'
+version: '3.0.0-beta.33'
 image: ghcr.io/pergolafabio/hikvision-doorbell
 slug: hikvision_doorbell_beta
 url: https://github.com/pergolafabio/Hikvision-Addons/tree/main/doorbell_beta

diff --git a/hikvision-doorbell/DOCS.md b/hikvision-doorbell/DOCS.md
@@ -17,7 +17,7 @@ For each of your doorbells, repeat the following configuration:
 | port          | 8000          | (Optional) Port of the doorbell
 | username      | admin         | Username to access the doorbell
 | password      |               | Password to access the doorbell
-| output_relays |               | (optional) Set this option if you don't see the correct number of door switches inside HA
+| output_relays |               | (optional) Set this option if you don't see the correct number of door switche or if you have attached an secure door control module on your indoor
 | caller_info   | false         | (optional) If you want to see the button prssed as an atribute on the call sensor, usefull for intercoms with multiple buttons.
 | scenes        | false         | (optional) Extra Scene buttons for indoor panels 
 
@@ -33,12 +33,12 @@ The following configuration setups two doorbells, named `Front door` and `Rear d
   ip: 192.168.0.2
   username: admin
   password: password
-  caller_info: true
 
 - name: "Indoor"
   ip: 192.168.0.3
   username: admin
   password: password
+  caller_info: true
   scenes: true
 
 - name: "Indoor Extension"

diff --git a/hikvision-doorbell/src/doorbell.py b/hikvision-doorbell/src/doorbell.py
@@ -1,6 +1,7 @@
 from ctypes import CDLL, CFUNCTYPE, POINTER, byref, c_byte, c_char, c_char_p, c_int, c_uint, c_void_p, pointer, sizeof, cast
 from enum import IntEnum
 import re
+import json
 from typing import Callable, Optional
 from loguru import logger
 from config import AppConfig
@@ -89,30 +90,49 @@ def logout(self):
         if not logout_result:
             logger.debug("SDK logout result {}", logout_result)
 
-    def unlock_door(self, lock_id: int):
-        """ Unlock the specified door using the SKD NET_DVR_RemoteControl.
-        If that fails, fallback to ISAPI `/ISAPI/AccessControl/RemoteControl/door/`.
+    def unlock_com(self, com_id: int):
 
-        See #83
-        """
-        gw = NET_DVR_CONTROL_GATEWAY()
-        gw.dwSize = sizeof(NET_DVR_CONTROL_GATEWAY)
-        gw.dwGatewayIndex = 1
-        gw.byCommand = 1  # opening command
-        gw.byLockType = 0  # this is normal lock not smart lock
-        gw.wLockID = lock_id  # door ID
-        gw.byControlSrc = (c_byte * 32)(*[97, 98, 99, 100])  # anything will do but can't be empty
-        gw.byControlType = 1
-
-        result = self._sdk.NET_DVR_RemoteControl(self.user_id, 16009, byref(gw), gw.dwSize)
-        if not result:
-            # SDK failed, try via ISAPI
+        url = "/ISAPI/SecurityCP/control/outputs/" + str(com_id) + "?format=json"
+        requestBody = {
+            "OutputsCtrl": {
+                "switch": "open"
+            }
+        }
+        self._call_isapi("PUT", url, json.dumps(requestBody))
+        logger.info(" Com {} unlocked by ISAPI", com_id +1)
+
+    def unlock_door(self, lock_id: int):
+        if not self._type is DeviceType.INDOOR:
+            """ Unlock the specified door using the SKD NET_DVR_RemoteControl.
+            If that fails, fallback to ISAPI `/ISAPI/AccessControl/RemoteControl/door/`.
+
+            See #83
+            """
+            gw = NET_DVR_CONTROL_GATEWAY()
+            gw.dwSize = sizeof(NET_DVR_CONTROL_GATEWAY)
+            gw.dwGatewayIndex = 1
+            gw.byCommand = 1  # opening command
+            gw.byLockType = 0  # this is normal lock not smart lock
+            gw.wLockID = lock_id  # door ID
+            gw.byControlSrc = (c_byte * 32)(*[97, 98, 99, 100])  # anything will do but can't be empty
+            gw.byControlType = 1
+
+            result = self._sdk.NET_DVR_RemoteControl(self.user_id, 16009, byref(gw), gw.dwSize)
+            if not result:
+                # SDK failed, try via ISAPI
+                url = "/ISAPI/AccessControl/RemoteControl/door/" + str(lock_id+1)
+                requestBody = "<RemoteControlDoor><cmd>open</cmd></RemoteControlDoor>"
+
+                logger.debug("NET_DVR_RemoteControl failed with code {}, trying ISAPI", self._sdk.NET_DVR_GetLastError())
+                self._call_isapi("PUT", url, requestBody)
+        else:
+                # ISAPI command for indoor
             url = "/ISAPI/AccessControl/RemoteControl/door/" + str(lock_id+1)
-            requestBody = "<RemoteControlDoor><cmd>open</cmd></RemoteControlDoor>"
+            requestBody = "<RemoteControlDoor><channelNo>1</channelNo><cmd>open</cmd><controlType>monitor</controlType></RemoteControlDoor>"
 
             logger.debug("NET_DVR_RemoteControl failed with code {}, trying ISAPI", self._sdk.NET_DVR_GetLastError())
             self._call_isapi("PUT", url, requestBody)
-   
+
         logger.info(" Door {} unlocked by SDK", lock_id + 1)
 
     def reboot_device(self):
@@ -148,6 +168,31 @@ def _call_isapi(self, http_method: str, url: str, requestBody: str = "") -> str:
 
         return response_body
 
+    def get_num_outputs_indoor(self) -> int:
+        """
+        Get the number of output relays configured for the indoor station, manually only.
+        """
+
+        # Define various functions, each using a different method to gather this information
+        def user_config() -> int:
+            if self._config.output_relays is not None:
+                logger.debug("Using the configured number of switches: {}", self._config.output_relays)
+                return self._config.output_relays
+            raise RuntimeError("No user configuration specified")
+
+        # Define the list of available endpoints to try
+        available_endpoints: list[Callable] = [user_config]
+        for endpoint in available_endpoints:
+            # Invoke the endpoint, if it errors out try another one
+            try:
+                return endpoint()
+            except RuntimeError:
+                # This endpoint failed, try the next one
+                pass
+
+        # We have run out of available endpoints to call
+        raise RuntimeError("Unable to get the number of doors, please configure the relays manually with this option in the config: output_relays")
+
     def get_num_outputs(self) -> int:
         """
         Get the number of output relays configured for this doorbell.

diff --git a/hikvision-doorbell/src/mqtt.py b/hikvision-doorbell/src/mqtt.py
@@ -115,7 +115,7 @@ def __init__(self, config: AppConfig.MQTT, doorbells: Registry) -> None:
             # Remove spaces and - from doorbell name
             sanitized_doorbell_name = sanitize_doorbell_name(doorbell_name)
 
-            # No Callsensor and relays for indoor
+            # No Callsensor for indoor
             if not doorbell._type is DeviceType.INDOOR:
 
                 ##################
@@ -132,22 +132,54 @@ def __init__(self, config: AppConfig.MQTT, doorbells: Registry) -> None:
                 call_sensor.set_state("idle")
                 call_sensor.set_availability(True)
                 self._sensors[doorbell]['call'] = call_sensor
-                ##################
-                # Doors
-                # Create switches for output relays used to open doors
+
+
+            ##################
+            # Doors
+            # Create switches for output relays used to open doors
+
+            if not doorbell._type is DeviceType.INDOOR:
                 num_doors = doorbell.get_num_outputs()
-                logger.debug("Configuring {} door switches", num_doors)
-                for door_id in range(num_doors):
-                    door_switch_info = SwitchInfo(
-                        name=f"Door {door_id+1} relay",
-                        unique_id=f"{device.identifiers}-door_relay_{door_id}",
+            else:
+                num_doors = doorbell.get_num_outputs_indoor()
+            logger.debug("Configuring {} door switches", num_doors)
+            for door_id in range(num_doors):
+                door_switch_info = SwitchInfo(
+                    name=f"Door {door_id+1} relay",
+                    unique_id=f"{device.identifiers}-door_relay_{door_id}",
+                    device=device,
+                    object_id=f"{sanitized_doorbell_name}_door_relay_{door_id}")
+                settings = Settings(mqtt=self._mqtt_settings, entity=door_switch_info, manual_availability=True)
+                door_switch = Switch(settings, self.door_switch_callback, (doorbell, door_id))
+                door_switch.off()
+                door_switch.set_availability(True)
+                self._sensors[doorbell][f'door_{door_id}'] = door_switch
+
+            ##################
+            # Output ports
+            # Create com1 and com2 ports for indoor stations
+
+            if doorbell._type is DeviceType.INDOOR:
+                logger.debug("Configuring 2 com relays for indoor station")
+                for com_id in range(2):
+                    com_switch_info = SwitchInfo(
+                        name=f"Com {com_id+1} relay",
+                        unique_id=f"{device.identifiers}-com_relay_{com_id}",
                         device=device,
-                        object_id=f"{sanitized_doorbell_name}_door_relay_{door_id}")
-                    settings = Settings(mqtt=self._mqtt_settings, entity=door_switch_info, manual_availability=True)
-                    door_switch = Switch(settings, self.door_switch_callback, (doorbell, door_id))
-                    door_switch.off()
-                    door_switch.set_availability(True)
-                    self._sensors[doorbell][f'door_{door_id}'] = door_switch
+                        object_id=f"{sanitized_doorbell_name}_com_relay_{com_id}")
+                    settings = Settings(mqtt=self._mqtt_settings, entity=com_switch_info, manual_availability=True)
+                    com_switch = Switch(settings, self.com_switch_callback, (doorbell, com_id))
+                    com_switch.off()
+                    com_switch.set_availability(True)
+                    self._sensors[doorbell][f'com_{com_id}'] = com_switch
+
+    def com_switch_callback(self, client, user_data: tuple[Doorbell, int], message: MQTTMessage):
+        doorbell, com_id = user_data
+        command = message.payload.decode("utf-8")
+        logger.debug("Received command: {}, com_id: {}, doorbell: {}", command, com_id, doorbell._config.name)
+        match command:
+            case "ON":
+                doorbell.unlock_com(com_id)
 
     def door_switch_callback(self, client, user_data: tuple[Doorbell, int], message: MQTTMessage):
         doorbell, door_id = user_data