From 91e4939bf04501410a6cf5e71ef6f6ed11333544 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 26 Nov 2024 10:00:34 +0100 Subject: [PATCH] Add fingerprint and nfc event support to unifiprotect (#130840) Co-authored-by: J. Nick Koston --- .../components/unifiprotect/const.py | 7 + .../components/unifiprotect/event.py | 148 +++++++++++--- .../components/unifiprotect/strings.json | 22 +++ tests/components/unifiprotect/conftest.py | 2 + tests/components/unifiprotect/test_event.py | 180 +++++++++++++++++- tests/components/unifiprotect/utils.py | 6 +- 6 files changed, 335 insertions(+), 30 deletions(-) diff --git a/homeassistant/components/unifiprotect/const.py b/homeassistant/components/unifiprotect/const.py index ad251ba6153..7d1e5b55d3f 100644 --- a/homeassistant/components/unifiprotect/const.py +++ b/homeassistant/components/unifiprotect/const.py @@ -1,5 +1,7 @@ """Constant definitions for UniFi Protect Integration.""" +from typing import Final + from uiprotect.data import ModelType, Version from homeassistant.const import Platform @@ -75,3 +77,8 @@ PLATFORMS = [ DISPATCH_ADD = "add_device" DISPATCH_ADOPT = "adopt_device" DISPATCH_CHANNELS = "new_camera_channels" + +EVENT_TYPE_FINGERPRINT_IDENTIFIED: Final = "identified" +EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED: Final = "not_identified" +EVENT_TYPE_NFC_SCANNED: Final = "scanned" +EVENT_TYPE_DOORBELL_RING: Final = "ring" diff --git a/homeassistant/components/unifiprotect/event.py b/homeassistant/components/unifiprotect/event.py index 8bbe568242b..f126920fb18 100644 --- a/homeassistant/components/unifiprotect/event.py +++ b/homeassistant/components/unifiprotect/event.py @@ -14,7 +14,13 @@ from homeassistant.components.event import ( from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .const import ATTR_EVENT_ID +from .const import ( + ATTR_EVENT_ID, + EVENT_TYPE_DOORBELL_RING, + EVENT_TYPE_FINGERPRINT_IDENTIFIED, + EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED, + EVENT_TYPE_NFC_SCANNED, +) from .data import ProtectData, ProtectDeviceType, UFPConfigEntry from .entity import EventEntityMixin, ProtectDeviceEntity, ProtectEventMixin @@ -23,22 +29,10 @@ from .entity import EventEntityMixin, ProtectDeviceEntity, ProtectEventMixin class ProtectEventEntityDescription(ProtectEventMixin, EventEntityDescription): """Describes UniFi Protect event entity.""" - -EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = ( - ProtectEventEntityDescription( - key="doorbell", - translation_key="doorbell", - name="Doorbell", - device_class=EventDeviceClass.DOORBELL, - icon="mdi:doorbell-video", - ufp_required_field="feature_flags.is_doorbell", - ufp_event_obj="last_ring_event", - event_types=[EventType.RING], - ), -) + entity_class: type[ProtectDeviceEntity] -class ProtectDeviceEventEntity(EventEntityMixin, ProtectDeviceEntity, EventEntity): +class ProtectDeviceRingEventEntity(EventEntityMixin, ProtectDeviceEntity, EventEntity): """A UniFi Protect event entity.""" entity_description: ProtectEventEntityDescription @@ -57,26 +51,128 @@ class ProtectDeviceEventEntity(EventEntityMixin, ProtectDeviceEntity, EventEntit if ( event and not self._event_already_ended(prev_event, prev_event_end) - and (event_types := description.event_types) - and (event_type := event.type) in event_types + and event.type is EventType.RING ): - self._trigger_event(event_type, {ATTR_EVENT_ID: event.id}) + self._trigger_event(EVENT_TYPE_DOORBELL_RING, {ATTR_EVENT_ID: event.id}) self.async_write_ha_state() +class ProtectDeviceNFCEventEntity(EventEntityMixin, ProtectDeviceEntity, EventEntity): + """A UniFi Protect NFC event entity.""" + + entity_description: ProtectEventEntityDescription + + @callback + def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None: + description = self.entity_description + + prev_event = self._event + prev_event_end = self._event_end + super()._async_update_device_from_protect(device) + if event := description.get_event_obj(device): + self._event = event + self._event_end = event.end if event else None + + if ( + event + and not self._event_already_ended(prev_event, prev_event_end) + and event.type is EventType.NFC_CARD_SCANNED + ): + event_data = {ATTR_EVENT_ID: event.id} + if event.metadata and event.metadata.nfc and event.metadata.nfc.nfc_id: + event_data["nfc_id"] = event.metadata.nfc.nfc_id + + self._trigger_event(EVENT_TYPE_NFC_SCANNED, event_data) + self.async_write_ha_state() + + +class ProtectDeviceFingerprintEventEntity( + EventEntityMixin, ProtectDeviceEntity, EventEntity +): + """A UniFi Protect fingerprint event entity.""" + + entity_description: ProtectEventEntityDescription + + @callback + def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None: + description = self.entity_description + + prev_event = self._event + prev_event_end = self._event_end + super()._async_update_device_from_protect(device) + if event := description.get_event_obj(device): + self._event = event + self._event_end = event.end if event else None + + if ( + event + and not self._event_already_ended(prev_event, prev_event_end) + and event.type is EventType.FINGERPRINT_IDENTIFIED + ): + event_data = {ATTR_EVENT_ID: event.id} + if ( + event.metadata + and event.metadata.fingerprint + and event.metadata.fingerprint.ulp_id + ): + event_data["ulp_id"] = event.metadata.fingerprint.ulp_id + event_identified = EVENT_TYPE_FINGERPRINT_IDENTIFIED + else: + event_data["ulp_id"] = "" + event_identified = EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED + + self._trigger_event(event_identified, event_data) + self.async_write_ha_state() + + +EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = ( + ProtectEventEntityDescription( + key="doorbell", + translation_key="doorbell", + device_class=EventDeviceClass.DOORBELL, + icon="mdi:doorbell-video", + ufp_required_field="feature_flags.is_doorbell", + ufp_event_obj="last_ring_event", + event_types=[EVENT_TYPE_DOORBELL_RING], + entity_class=ProtectDeviceRingEventEntity, + ), + ProtectEventEntityDescription( + key="nfc", + translation_key="nfc", + device_class=EventDeviceClass.DOORBELL, + icon="mdi:nfc", + ufp_required_field="feature_flags.support_nfc", + ufp_event_obj="last_nfc_card_scanned_event", + event_types=[EVENT_TYPE_NFC_SCANNED], + entity_class=ProtectDeviceNFCEventEntity, + ), + ProtectEventEntityDescription( + key="fingerprint", + translation_key="fingerprint", + device_class=EventDeviceClass.DOORBELL, + icon="mdi:fingerprint", + ufp_required_field="feature_flags.has_fingerprint_sensor", + ufp_event_obj="last_fingerprint_identified_event", + event_types=[ + EVENT_TYPE_FINGERPRINT_IDENTIFIED, + EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED, + ], + entity_class=ProtectDeviceFingerprintEventEntity, + ), +) + + @callback def _async_event_entities( data: ProtectData, ufp_device: ProtectAdoptableDeviceModel | None = None, ) -> list[ProtectDeviceEntity]: - entities: list[ProtectDeviceEntity] = [] - for device in data.get_cameras() if ufp_device is None else [ufp_device]: - entities.extend( - ProtectDeviceEventEntity(data, device, description) - for description in EVENT_DESCRIPTIONS - if description.has_required(device) - ) - return entities + return [ + description.entity_class(data, device, description) + for device in (data.get_cameras() if ufp_device is None else [ufp_device]) + for description in EVENT_DESCRIPTIONS + if description.has_required(device) + ] async def async_setup_entry( diff --git a/homeassistant/components/unifiprotect/strings.json b/homeassistant/components/unifiprotect/strings.json index 7b2a06dfef7..8ecb4076409 100644 --- a/homeassistant/components/unifiprotect/strings.json +++ b/homeassistant/components/unifiprotect/strings.json @@ -137,6 +137,7 @@ }, "event": { "doorbell": { + "name": "Doorbell", "state_attributes": { "event_type": { "state": { @@ -144,6 +145,27 @@ } } } + }, + "nfc": { + "name": "NFC", + "state_attributes": { + "event_type": { + "state": { + "scanned": "Scanned" + } + } + } + }, + "fingerprint": { + "name": "Fingerprint", + "state_attributes": { + "event_type": { + "state": { + "identified": "Identified", + "not_identified": "Not identified" + } + } + } } } }, diff --git a/tests/components/unifiprotect/conftest.py b/tests/components/unifiprotect/conftest.py index 0bef1ff0eb9..fad65c095df 100644 --- a/tests/components/unifiprotect/conftest.py +++ b/tests/components/unifiprotect/conftest.py @@ -233,6 +233,8 @@ def doorbell_fixture(camera: Camera, fixed_now: datetime): doorbell.feature_flags.has_speaker = True doorbell.feature_flags.has_privacy_mask = True doorbell.feature_flags.is_doorbell = True + doorbell.feature_flags.has_fingerprint_sensor = True + doorbell.feature_flags.support_nfc = True doorbell.feature_flags.has_chime = True doorbell.feature_flags.has_smart_detect = True doorbell.feature_flags.has_package_camera = True diff --git a/tests/components/unifiprotect/test_event.py b/tests/components/unifiprotect/test_event.py index 9d1a701fe39..cc2195c1dba 100644 --- a/tests/components/unifiprotect/test_event.py +++ b/tests/components/unifiprotect/test_event.py @@ -33,11 +33,11 @@ async def test_camera_remove( ufp.api.bootstrap.nvr.system_info.ustorage = None await init_entry(hass, ufp, [doorbell, unadopted_camera]) - assert_entity_counts(hass, Platform.EVENT, 1, 1) + assert_entity_counts(hass, Platform.EVENT, 3, 3) await remove_entities(hass, ufp, [doorbell, unadopted_camera]) assert_entity_counts(hass, Platform.EVENT, 0, 0) await adopt_devices(hass, ufp, [doorbell, unadopted_camera]) - assert_entity_counts(hass, Platform.EVENT, 1, 1) + assert_entity_counts(hass, Platform.EVENT, 3, 3) async def test_doorbell_ring( @@ -50,7 +50,7 @@ async def test_doorbell_ring( """Test a doorbell ring event.""" await init_entry(hass, ufp, [doorbell, unadopted_camera]) - assert_entity_counts(hass, Platform.EVENT, 1, 1) + assert_entity_counts(hass, Platform.EVENT, 3, 3) events: list[HAEvent] = [] @callback @@ -152,3 +152,177 @@ async def test_doorbell_ring( assert state assert state.state == timestamp unsub() + + +async def test_doorbell_nfc_scanned( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test a doorbell NFC scanned event.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 3, 3) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = ids_from_device_description( + Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[1] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + event = Event( + model=ModelType.EVENT, + id="test_event_id", + type=EventType.NFC_CARD_SCANNED, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={"nfc": {"nfc_id": "test_nfc_id", "user_id": "test_user_id"}}, + ) + + new_camera = doorbell.copy() + new_camera.last_nfc_card_scanned_event_id = "test_event_id" + ufp.api.bootstrap.cameras = {new_camera.id: new_camera} + ufp.api.bootstrap.events = {event.id: event} + + mock_msg = Mock() + mock_msg.changed_data = {} + mock_msg.new_obj = event + ufp.ws_msg(mock_msg) + + await hass.async_block_till_done() + + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + assert state.attributes[ATTR_EVENT_ID] == "test_event_id" + assert state.attributes["nfc_id"] == "test_nfc_id" + + unsub() + + +async def test_doorbell_fingerprint_identified( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test a doorbell fingerprint identified event.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 3, 3) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = ids_from_device_description( + Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[2] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + event = Event( + model=ModelType.EVENT, + id="test_event_id", + type=EventType.FINGERPRINT_IDENTIFIED, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={"fingerprint": {"ulp_id": "test_ulp_id"}}, + ) + + new_camera = doorbell.copy() + new_camera.last_fingerprint_identified_event_id = "test_event_id" + ufp.api.bootstrap.cameras = {new_camera.id: new_camera} + ufp.api.bootstrap.events = {event.id: event} + + mock_msg = Mock() + mock_msg.changed_data = {} + mock_msg.new_obj = event + ufp.ws_msg(mock_msg) + + await hass.async_block_till_done() + + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + assert state.attributes[ATTR_EVENT_ID] == "test_event_id" + assert state.attributes["ulp_id"] == "test_ulp_id" + + unsub() + + +async def test_doorbell_fingerprint_not_identified( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test a doorbell fingerprint identified event.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 3, 3) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = ids_from_device_description( + Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[2] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + event = Event( + model=ModelType.EVENT, + id="test_event_id", + type=EventType.FINGERPRINT_IDENTIFIED, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={"fingerprint": {}}, + ) + + new_camera = doorbell.copy() + new_camera.last_fingerprint_identified_event_id = "test_event_id" + ufp.api.bootstrap.cameras = {new_camera.id: new_camera} + ufp.api.bootstrap.events = {event.id: event} + + mock_msg = Mock() + mock_msg.changed_data = {} + mock_msg.new_obj = event + ufp.ws_msg(mock_msg) + + await hass.async_block_till_done() + + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + assert state.attributes[ATTR_EVENT_ID] == "test_event_id" + assert state.attributes["ulp_id"] == "" + + unsub() diff --git a/tests/components/unifiprotect/utils.py b/tests/components/unifiprotect/utils.py index 25a9ddcbb92..5a1ffa8258e 100644 --- a/tests/components/unifiprotect/utils.py +++ b/tests/components/unifiprotect/utils.py @@ -109,7 +109,11 @@ def ids_from_device_description( """Return expected unique_id and entity_id for a give platform/device/description combination.""" entity_name = normalize_name(device.display_name) - description_entity_name = normalize_name(str(description.name)) + + if description.name and isinstance(description.name, str): + description_entity_name = normalize_name(description.name) + else: + description_entity_name = normalize_name(description.key) unique_id = f"{device.mac}_{description.key}" entity_id = f"{platform.value}.{entity_name}_{description_entity_name}"