From 951c70fdf0586624e817de6d79402e1fc34b1606 Mon Sep 17 00:00:00 2001 From: Carlo Mion Date: Mon, 6 Mar 2023 14:53:50 +0100 Subject: [PATCH 1/2] fix(HomeAssistantAPI): multiple doorbells each have their own set of sensors inside the `_sensors` dict Fixes #18 --- hikvision-doorbell/src/home_assistant.py | 123 +++++++++--------- .../tests/test_homeassistant.py | 24 ++++ 2 files changed, 88 insertions(+), 59 deletions(-) diff --git a/hikvision-doorbell/src/home_assistant.py b/hikvision-doorbell/src/home_assistant.py index c7bb2f3..5b0d9ec 100644 --- a/hikvision-doorbell/src/home_assistant.py +++ b/hikvision-doorbell/src/home_assistant.py @@ -36,7 +36,7 @@ class Sensor(TypedDict): class HomeAssistantAPI(EventHandler): name = "HomeAssistantAPI" - _sensors: dict[str, Sensor] = {} + _sensors: dict[Doorbell, dict[str, Sensor]] = {} def __init__(self, config: AppConfig.HomeAssistant, doorbells: Registry) -> None: super().__init__() @@ -45,42 +45,45 @@ def __init__(self, config: AppConfig.HomeAssistant, doorbells: Registry) -> None self._config = config self._doorbells = doorbells - self._sensors['door'] = Sensor( - name="door", - type="binary_sensor", - attributes={"device_class": "door"} - ) - self._sensors['callstatus'] = Sensor( - name="callstatus", - type="binary_sensor", - attributes={"device_class": "sound"} - ) - self._sensors['motion'] = Sensor( - name="motion", - type="binary_sensor", - attributes={"device_class": "motion"} - ) - self._sensors['tamper'] = Sensor( - name='tamper', - type="binary_sensor", - attributes={"device_class": "tamper"} - ) - self._sensors['dismiss'] = Sensor( - name='dismiss', - type="binary_sensor", - attributes={} - ) - self._sensors['alarm'] = Sensor( - name='alarm', - type="binary_sensor", - attributes={"device_class": "door"} - ) # For each outdoor doorbell, initialize the sensors inside HA for doorbell in doorbells.values(): # Skip if we have an indoor unit, since it does not support all the events for now if doorbell._type == DeviceType.INDOOR: continue - for name, sensor in self._sensors.items(): + # Create dict to contain the sensors of each doorbell + self._sensors[doorbell] = {} + sensors = self._sensors[doorbell] + sensors['door'] = Sensor( + name="door", + type="binary_sensor", + attributes={"device_class": "door"} + ) + sensors['callstatus'] = Sensor( + name="callstatus", + type="binary_sensor", + attributes={"device_class": "sound"} + ) + sensors['motion'] = Sensor( + name="motion", + type="binary_sensor", + attributes={"device_class": "motion"} + ) + sensors['tamper'] = Sensor( + name='tamper', + type="binary_sensor", + attributes={"device_class": "tamper"} + ) + sensors['dismiss'] = Sensor( + name='dismiss', + type="binary_sensor", + attributes={} + ) + sensors['alarm'] = Sensor( + name='alarm', + type="binary_sensor", + attributes={"device_class": "door"} + ) + for name, sensor in sensors.items(): doorbell_name = doorbell._config.name # Add friendly name attribute to existing attributes dict containing the doorbell name + sensor name sensor['attributes'] = sensor['attributes'] | {'friendly_name': f"{doorbell_name} {name}"} @@ -113,10 +116,10 @@ async def motion_detection( alarm_info: NET_DVR_ALARMINFO_V30, buffer_length, user_pointer: c_void_p): - logger.info("Motion detected from {}, updating sensor {}", doorbell._config.name, self._sensors['motion']) - self.update_sensor(doorbell._config.name, self._sensors['motion'], 'on') + logger.info("Motion detected from {}, updating sensor {}", doorbell._config.name, self._sensors[doorbell]['motion']) + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['motion'], 'on') await asyncio.sleep(5) - self.update_sensor(doorbell._config.name, self._sensors['motion'], 'off') + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['motion'], 'off') @override async def video_intercom_event( @@ -131,19 +134,19 @@ async def video_intercom_event( logger.info("Door {} unlocked by {}, updating sensor {}", alarm_info.uEventInfo.struUnlockRecord.wLockID, list(alarm_info.uEventInfo.struUnlockRecord.byControlSrc), - self._sensors['door']) + self._sensors[doorbell]['door']) additional_attributes = { 'Unlock': list(alarm_info.uEventInfo.struUnlockRecord.byControlSrc), 'DoorID': alarm_info.uEventInfo.struUnlockRecord.wLockID } # Add additional attributes to the sensor - original_attributes = self._sensors['door']['attributes'] - self._sensors['door']['attributes'] = original_attributes | additional_attributes - self.update_sensor(doorbell._config.name, self._sensors['door'], 'on') + original_attributes = self._sensors[doorbell]['door']['attributes'] + self._sensors[doorbell]['door']['attributes'] = original_attributes | additional_attributes + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['door'], 'on') await asyncio.sleep(5) # Revert back to original attributes - self._sensors['door']['attributes'] = original_attributes - self.update_sensor(doorbell._config.name, self._sensors['door'], 'off') + self._sensors[doorbell]['door']['attributes'] = original_attributes + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['door'], 'off') elif alarm_info.byEventType == VIDEO_INTERCOM_EVENT_EVENTTYPE_ILLEGAL_CARD_SWIPING_EVENT: logger.info("Illegal card swiping") @@ -160,40 +163,42 @@ async def video_intercom_alarm( buffer_length, user_pointer: c_void_p): if alarm_info.byAlarmType == VIDEO_INTERCOM_ALARM_ALARMTYPE_DOORBELL_RINGING: - logger.info("Doorbell ringing, updating sensor {}", self._sensors['callstatus']) - self.update_sensor(doorbell._config.name, self._sensors['callstatus'], 'on') + logger.info("Doorbell ringing, updating sensor {}", self._sensors[doorbell]['callstatus']) + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['callstatus'], 'on') await asyncio.sleep(5) - self.update_sensor(doorbell._config.name, self._sensors['callstatus'], 'off') + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['callstatus'], 'off') elif alarm_info.byAlarmType == VIDEO_INTERCOM_ALARM_ALARMTYPE_DISMISS_INCOMING_CALL: - logger.info("Call dismissed, updating sensor {}", self._sensors['dismiss']) - self.update_sensor(doorbell._config.name, self._sensors['dismiss'], 'on') + logger.info("Call dismissed, updating sensor {}", self._sensors[doorbell]['dismiss']) + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['dismiss'], 'on') await asyncio.sleep(5) - self.update_sensor(doorbell._config.name, self._sensors['dismiss'], 'off') + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['dismiss'], 'off') elif alarm_info.byAlarmType == VIDEO_INTERCOM_ALARM_ALARMTYPE_TAMPERING_ALARM: - logger.info("Tampering alarm, updating sensor {}", self._sensors['tamper']) - self.update_sensor(doorbell._config.name, self._sensors['tamper'], 'on') + logger.info("Tampering alarm, updating sensor {}", self._sensors[doorbell]['tamper']) + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['tamper'], 'on') await asyncio.sleep(5) - self.update_sensor(doorbell._config.name, self._sensors['tamper'], 'off') + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['tamper'], 'off') elif alarm_info.byAlarmType == VIDEO_INTERCOM_ALARM_ALARMTYPE_DOOR_NOT_OPEN or VIDEO_INTERCOM_ALARM_ALARMTYPE_DOOR_NOT_CLOSED: logger.info("Alarm {} detected on lock {}, updating sensor {}", alarm_info.uAlarmInfo, alarm_info.wLockID, - self._sensors['alarm']) + self._sensors[doorbell]['alarm']) additional_attributes = { 'AlarmInfo': alarm_info.uAlarmInfo, 'AlarmType': alarm_info.byAlarmType, 'LockID': alarm_info.wLockID } # Add additional attributes to the sensor - original_attributes = self._sensors['alarm']['attributes'] - self._sensors['alarm']['attributes'] = original_attributes | additional_attributes - self.update_sensor(doorbell._config.name, self._sensors['alarm'], 'on') + original_attributes = self._sensors[doorbell]['alarm']['attributes'] + self._sensors[doorbell]['alarm']['attributes'] = original_attributes | additional_attributes + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['alarm'], 'on') await asyncio.sleep(6) # Revert back to original attributes - self._sensors['alarm']['attributes'] = original_attributes - self.update_sensor(doorbell._config.name, self._sensors['alarm'], 'off') + self._sensors[doorbell]['alarm']['attributes'] = original_attributes + self.update_sensor(doorbell._config.name, self._sensors[doorbell]['alarm'], 'off') else: logger.warning("Unhandled alarmType: {}", alarm_info.byAlarmType) + + @override async def isapi_alarm( self, doorbell: Doorbell, @@ -202,9 +207,9 @@ async def isapi_alarm( alarm_info: NET_DVR_ALARM_ISAPI_INFO, buffer_length, user_pointer: c_void_p): - logger.info("Isapi alarm detected from {}, file saved in: {}", - doorbell._config.name, - alarm_info.szFilename) + logger.info("Isapi alarm detected from {}, file saved in: {}", + doorbell._config.name, + alarm_info.szFilename) @override async def unhandled_event( diff --git a/hikvision-doorbell/tests/test_homeassistant.py b/hikvision-doorbell/tests/test_homeassistant.py index f88e7b6..043f79a 100644 --- a/hikvision-doorbell/tests/test_homeassistant.py +++ b/hikvision-doorbell/tests/test_homeassistant.py @@ -2,6 +2,7 @@ import asyncio from pytest_mock import MockerFixture +from doorbell import DeviceType, Doorbell from home_assistant import HomeAssistantAPI, sanitize_doorbell_name @@ -21,3 +22,26 @@ def test_unhandled_event(mocker: MockerFixture): ha = HomeAssistantAPI(config, registry) asyncio.run(ha.unhandled_event(doorbell, 0, None, None, None, None)) # type: ignore + + +def test_friendly_name_multiple_doorbells(mocker: MockerFixture): + """Test that with multiple doorbells, the sensors gets their correct friendly_name attribute. See issue #18 """ + registry = mocker.patch('doorbell.Registry') + fake_config = mocker.MagicMock() + fake_config.name = "A" + doorbell_a = mocker.Mock(Doorbell, _type=DeviceType.OUTDOOR, _config=fake_config) + fake_config = mocker.MagicMock() + fake_config.name = "B" + doorbell_b = mocker.Mock(Doorbell, _type=DeviceType.OUTDOOR, _config=fake_config) + registry.values.return_value = [doorbell_a, doorbell_b] + + config = mocker.patch('config.AppConfig.HomeAssistant') + # Mock call to API + mocker.patch("home_assistant.requests") + + ha = HomeAssistantAPI(config, registry) + + assert len(ha._sensors) == 2 + assert len(ha._sensors[doorbell_a]) == 6 + assert ha._sensors[doorbell_a]['door']['attributes']["friendly_name"] == 'A door' + From 2f553122233fd0e366d16122a8464675fcc1d040 Mon Sep 17 00:00:00 2001 From: Carlo Mion Date: Mon, 6 Mar 2023 15:06:52 +0100 Subject: [PATCH 2/2] refactor(HomeAssistantAPI): print controlSource as a string, removing suffix 0s --- hikvision-doorbell/src/home_assistant.py | 7 +++++-- hikvision-doorbell/src/sdk/hcnetsdk.py | 6 ++++++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/hikvision-doorbell/src/home_assistant.py b/hikvision-doorbell/src/home_assistant.py index 5b0d9ec..4c5c470 100644 --- a/hikvision-doorbell/src/home_assistant.py +++ b/hikvision-doorbell/src/home_assistant.py @@ -131,12 +131,15 @@ async def video_intercom_event( buffer_length, user_pointer: c_void_p): if alarm_info.byEventType == VIDEO_INTERCOM_EVENT_EVENTTYPE_UNLOCK_LOG: + # Convert the controlSource to a string, removing suffix `0`s + control_source = alarm_info.uEventInfo.struUnlockRecord.controlSource() logger.info("Door {} unlocked by {}, updating sensor {}", alarm_info.uEventInfo.struUnlockRecord.wLockID, - list(alarm_info.uEventInfo.struUnlockRecord.byControlSrc), + control_source, self._sensors[doorbell]['door']) additional_attributes = { - 'Unlock': list(alarm_info.uEventInfo.struUnlockRecord.byControlSrc), + # TODO: better rename to `control source`, similar to the original field? + 'Unlock': control_source, 'DoorID': alarm_info.uEventInfo.struUnlockRecord.wLockID } # Add additional attributes to the sensor diff --git a/hikvision-doorbell/src/sdk/hcnetsdk.py b/hikvision-doorbell/src/sdk/hcnetsdk.py index edb7200..77e0f2c 100644 --- a/hikvision-doorbell/src/sdk/hcnetsdk.py +++ b/hikvision-doorbell/src/sdk/hcnetsdk.py @@ -395,6 +395,12 @@ class NET_DVR_UNLOCK_RECORD_INFO(Structure): ("byRes", BYTE * 168), ] + def controlSource(self): + """Return the controls source number as a string representation, removing the ending `0`s""" + serial = "".join([str(number) for number in self.byControlSrc[:]]) + return re.sub(r"0*$", "", serial) + + class NET_DVR_NOTICEDATA_RECEIPT_INFO(Structure): _fields_ = [ ("byNoticeNumber", BYTE * MAX_NOTICE_NUMBER_LEN),