Add fingerprint and nfc event support to unifiprotect (#130840)

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Raphael Hehl 2024-11-26 10:00:34 +01:00 committed by GitHub
parent 521cc67d45
commit 91e4939bf0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 335 additions and 30 deletions

View File

@ -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"

View File

@ -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(

View File

@ -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"
}
}
}
}
}
},

View File

@ -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

View File

@ -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()

View File

@ -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}"