Skip to content

Commit

Permalink
Merge pull request #81 from pergolafabio/fix/ha_events_multiple_doorb…
Browse files Browse the repository at this point in the history 
…ells

Fix/HA events with multiple doorbells
  • Loading branch information
@mion00
mion00 authored Mar 6, 2023
2 parents bd24e4c + 2f55312 commit eb6b00a
Show file tree
Hide file tree
Showing 3 changed files with 99 additions and 61 deletions.
130 changes: 69 additions & 61 deletions hikvision-doorbell/src/home_assistant.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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__()
Expand All @@ -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}"}
Expand Down Expand Up @@ -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(
Expand All @@ -128,22 +131,25 @@ 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),
self._sensors['door'])
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
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")
Expand All @@ -160,40 +166,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,
Expand All @@ -202,9 +210,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(
Expand Down
6 changes: 6 additions & 0 deletions hikvision-doorbell/src/sdk/hcnetsdk.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
24 changes: 24 additions & 0 deletions hikvision-doorbell/tests/test_homeassistant.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import asyncio

from pytest_mock import MockerFixture
from doorbell import DeviceType, Doorbell
from home_assistant import HomeAssistantAPI, sanitize_doorbell_name


Expand All @@ -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'

0 comments on commit eb6b00a

Please sign in to comment.