home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2026-08-03 20:24:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

Migrate UniFi Protect detection sensors to the public API and add category event entities (#174948)

Browse Source
This commit is contained in:
Raphael Hehl
2026-07-29 15:13:53 +02:00
committed by GitHub
parent 7a8f476a8f
commit d4cbeaa9ff
14 changed files with 1322 additions and 589 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifiprotect/binary_sensor.py
+125 -159
View File
@@ -2,17 +2,15 @@
from collections.abc import Sequence
import dataclasses
import operator
from typing import cast, override
from uiprotect.data import (
NVR,
Camera,
Event,
ModelType,
MountType,
ProtectAdoptableDeviceModel,
Sensor,
SmartDetectObjectType,
)
from uiprotect.data.nvr import UOSDisk
from uiprotect.data.public_devices import (
@@ -296,6 +294,127 @@ CAMERA_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
ufp_value="is_person_tracking_enabled",
ufp_perm=PermRequired.NO_WRITE,
),
# Sustained state via the public devices WS (uiprotect pushes a camera
# update on each detection transition).
ProtectBinaryEntityDescription(
key="motion",
device_class=BinarySensorDeviceClass.MOTION,
ufp_public_value="is_motion_detected",
ufp_event_driven=True,
),
ProtectBinaryEntityDescription(
key="smart_obj_any",
translation_key="object_detected",
ufp_required_field="feature_flags.has_smart_detect",
ufp_public_value="is_smart_currently_detected",
ufp_event_driven=True,
entity_registry_enabled_default=False,
),
ProtectBinaryEntityDescription(
key="smart_obj_person",
translation_key="person_detected",
ufp_required_field="can_detect_person",
ufp_public_value="is_person_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_person_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_obj_vehicle",
translation_key="vehicle_detected",
ufp_required_field="can_detect_vehicle",
ufp_public_value="is_vehicle_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_vehicle_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_obj_animal",
translation_key="animal_detected",
ufp_required_field="can_detect_animal",
ufp_public_value="is_animal_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_animal_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_any",
translation_key="audio_object_detected",
ufp_required_field="feature_flags.smart_detect_audio_types",
ufp_public_value="is_audio_currently_detected",
ufp_event_driven=True,
entity_registry_enabled_default=False,
),
ProtectBinaryEntityDescription(
key="smart_audio_smoke",
translation_key="smoke_alarm_detected",
ufp_required_field="can_detect_smoke",
ufp_public_value="is_smoke_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_smoke_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_cmonx",
translation_key="co_alarm_detected",
device_class=BinarySensorDeviceClass.CO,
ufp_required_field="can_detect_co",
ufp_public_value="is_cmonx_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_co_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_siren",
translation_key="siren_detected",
ufp_required_field="can_detect_siren",
ufp_public_value="is_siren_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_siren_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_baby_cry",
translation_key="baby_cry_detected",
ufp_required_field="can_detect_baby_cry",
ufp_public_value="is_baby_cry_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_baby_cry_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_speak",
translation_key="speaking_detected",
ufp_required_field="can_detect_speaking",
ufp_public_value="is_speaking_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_speaking_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_bark",
translation_key="barking_detected",
ufp_required_field="can_detect_bark",
ufp_public_value="is_bark_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_bark_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_car_alarm",
translation_key="car_alarm_detected",
ufp_required_field="can_detect_car_alarm",
ufp_public_value="is_car_alarm_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_car_alarm_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_car_horn",
translation_key="car_horn_detected",
ufp_required_field="can_detect_car_horn",
ufp_public_value="is_car_horn_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_car_horn_detection_on"),
),
ProtectBinaryEntityDescription(
key="smart_audio_glass_break",
translation_key="glass_break_detected",
ufp_required_field="can_detect_glass_break",
ufp_public_value="is_glass_break_currently_detected",
ufp_event_driven=True,
ufp_public_enabled_fn=operator.attrgetter("is_glass_break_detection_on"),
),
)
LIGHT_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
@@ -418,6 +537,8 @@ SENSE_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
),
)
# Doorbell ring is momentary (no sustained public state), so it stays on the
# private event path.
EVENT_SENSORS: tuple[ProtectBinaryEventEntityDescription, ...] = (
ProtectBinaryEventEntityDescription(
key="doorbell",
@@ -426,125 +547,6 @@ EVENT_SENSORS: tuple[ProtectBinaryEventEntityDescription, ...] = (
ufp_required_field="feature_flags.is_doorbell",
ufp_event_obj="last_ring_event",
),
ProtectBinaryEventEntityDescription(
key="motion",
device_class=BinarySensorDeviceClass.MOTION,
ufp_enabled="is_motion_detection_on",
ufp_event_obj="last_motion_event",
),
ProtectBinaryEventEntityDescription(
key="smart_obj_any",
translation_key="object_detected",
ufp_required_field="feature_flags.has_smart_detect",
ufp_event_obj="last_smart_detect_event",
entity_registry_enabled_default=False,
),
ProtectBinaryEventEntityDescription(
key="smart_obj_person",
translation_key="person_detected",
ufp_obj_type=SmartDetectObjectType.PERSON,
ufp_required_field="can_detect_person",
ufp_enabled="is_person_detection_on",
ufp_event_obj="last_person_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_obj_vehicle",
translation_key="vehicle_detected",
ufp_obj_type=SmartDetectObjectType.VEHICLE,
ufp_required_field="can_detect_vehicle",
ufp_enabled="is_vehicle_detection_on",
ufp_event_obj="last_vehicle_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_obj_animal",
translation_key="animal_detected",
ufp_obj_type=SmartDetectObjectType.ANIMAL,
ufp_required_field="can_detect_animal",
ufp_enabled="is_animal_detection_on",
ufp_event_obj="last_animal_detect_event",
),
# Package detection is a momentary smart-detect event, not a sustained state:
# it is the package event entity (event.py), not a binary sensor.
ProtectBinaryEventEntityDescription(
key="smart_audio_any",
translation_key="audio_object_detected",
ufp_required_field="feature_flags.has_smart_detect",
ufp_event_obj="last_smart_audio_detect_event",
entity_registry_enabled_default=False,
),
ProtectBinaryEventEntityDescription(
key="smart_audio_smoke",
translation_key="smoke_alarm_detected",
ufp_obj_type=SmartDetectObjectType.SMOKE,
ufp_required_field="can_detect_smoke",
ufp_enabled="is_smoke_detection_on",
ufp_event_obj="last_smoke_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_audio_cmonx",
translation_key="co_alarm_detected",
device_class=BinarySensorDeviceClass.CO,
ufp_required_field="can_detect_co",
ufp_enabled="is_co_detection_on",
ufp_event_obj="last_cmonx_detect_event",
ufp_obj_type=SmartDetectObjectType.CMONX,
),
ProtectBinaryEventEntityDescription(
key="smart_audio_siren",
translation_key="siren_detected",
ufp_obj_type=SmartDetectObjectType.SIREN,
ufp_required_field="can_detect_siren",
ufp_enabled="is_siren_detection_on",
ufp_event_obj="last_siren_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_audio_baby_cry",
translation_key="baby_cry_detected",
ufp_obj_type=SmartDetectObjectType.BABY_CRY,
ufp_required_field="can_detect_baby_cry",
ufp_enabled="is_baby_cry_detection_on",
ufp_event_obj="last_baby_cry_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_audio_speak",
translation_key="speaking_detected",
ufp_obj_type=SmartDetectObjectType.SPEAK,
ufp_required_field="can_detect_speaking",
ufp_enabled="is_speaking_detection_on",
ufp_event_obj="last_speaking_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_audio_bark",
translation_key="barking_detected",
ufp_obj_type=SmartDetectObjectType.BARK,
ufp_required_field="can_detect_bark",
ufp_enabled="is_bark_detection_on",
ufp_event_obj="last_bark_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_audio_car_alarm",
translation_key="car_alarm_detected",
ufp_obj_type=SmartDetectObjectType.BURGLAR,
ufp_required_field="can_detect_car_alarm",
ufp_enabled="is_car_alarm_detection_on",
ufp_event_obj="last_car_alarm_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_audio_car_horn",
translation_key="car_horn_detected",
ufp_obj_type=SmartDetectObjectType.CAR_HORN,
ufp_required_field="can_detect_car_horn",
ufp_enabled="is_car_horn_detection_on",
ufp_event_obj="last_car_horn_detect_event",
),
ProtectBinaryEventEntityDescription(
key="smart_audio_glass_break",
translation_key="glass_break_detected",
ufp_obj_type=SmartDetectObjectType.GLASS_BREAK,
ufp_required_field="can_detect_glass_break",
ufp_enabled="is_glass_break_detection_on",
ufp_event_obj="last_glass_break_detect_event",
),
)
VIEWER_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = (
@@ -665,31 +667,6 @@ class ProtectEventBinarySensor(EventEntityMixin, BinarySensorEntity):
self._attr_is_on = False
self._attr_extra_state_attributes = {}
@callback
def _find_active_event_with_object_type(
self, device: ProtectDeviceType
) -> Event | None:
"""Find an active event containing this sensor's object type.
Fallback for issue #152133: last_smart_detect_event_ids may not update
immediately when a new detection type is added to an ongoing event.
"""
obj_type = self.entity_description.ufp_obj_type
if obj_type is None or not isinstance(device, Camera):
return None
# Check known active event IDs from camera first (fast path)
for event_id in device.last_smart_detect_event_ids.values():
if (
event_id
and (event := self.data.api.bootstrap.events.get(event_id))
and event.end is None
and obj_type in event.smart_detect_types
):
return event
return None
@callback
@override
def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
@@ -700,22 +677,11 @@ class ProtectEventBinarySensor(EventEntityMixin, BinarySensorEntity):
super()._async_update_device_from_protect(device)
event = description.get_event_obj(device)
if event is None:
# Fallback for #152133: check active events directly
event = self._find_active_event_with_object_type(device)
if event:
self._event = event
self._event_end = event.end
if not (
event
and (
description.ufp_obj_type is None
or description.has_matching_smart(event)
)
and not self._event_already_ended(prev_event, prev_event_end)
):
if not (event and not self._event_already_ended(prev_event, prev_event_end)):
self._set_event_done()
return
homeassistant/components/unifiprotect/const.py
+1
View File
@@ -13,6 +13,7 @@ AUTH_RETRIES = 2
ATTR_EVENT_SCORE = "event_score"
ATTR_EVENT_ID = "event_id"
ATTR_SMART_DETECT_TYPES = "smart_detect_types"
ATTR_WIDTH = "width"
ATTR_HEIGHT = "height"
ATTR_FPS = "fps"
homeassistant/components/unifiprotect/data.py
+25 -7
View File
@@ -98,6 +98,7 @@ class ProtectData:
self.auth_retries = 0
self.last_update_success = False
self.last_public_update_success = False
self.last_events_update_success = False
self.api = protect
self.adopt_signal = _async_dispatch_id(entry, DISPATCH_ADOPT)
self.add_signal = _async_dispatch_id(entry, DISPATCH_ADD)
@@ -212,6 +213,7 @@ class ProtectData:
"""Subscribe and do the refresh."""
self.last_update_success = True
self.last_public_update_success = True
self.last_events_update_success = True
self._async_update_change(True, force_update=True)
api = self.api
self._unsubs = [
@@ -227,6 +229,7 @@ class ProtectData:
self._async_process_public_devices_ws_message
),
api.subscribe_devices_websocket_state(self._async_public_ws_state_changed),
api.subscribe_events_websocket_state(self._async_events_ws_state_changed),
]
@callback
@@ -310,14 +313,14 @@ class ProtectData:
) -> None:
"""Dispatch a public events websocket event to its subscribers.
Only the start of an event is dispatched, routed to the subscribers that
registered for this device and event type; an entity that cares about a
sub-type (e.g. a smart-detect object type) filters further itself.
Subscriptions are keyed by ``device_id`` (the stable cross-API join key,
shared by the private and public bootstraps), so the event routes
directly without a bootstrap lookup.
Each non-eviction change is dispatched — a detection type may surface at
the event start, on a later update, or only as it ends — routed to the
subscribers registered for this device and event type; entities fire each
``(event, type)`` once. Subscriptions are keyed by ``device_id`` (the
stable cross-API join key, shared by the private and public bootstraps),
so the event routes directly without a bootstrap lookup.
"""
if change is not EventChange.STARTED:
if change is EventChange.REMOVED:
return
if not (
subscriptions := self._public_event_subscriptions.get(
@@ -373,6 +376,21 @@ class ProtectData:
for public in list(self.api.public_bootstrap.cameras.values()):
async_dispatcher_send(self._hass, self.channels_signal, public)
@callback
def _async_events_ws_state_changed(self, state: WebsocketState) -> None:
"""Handle a change in the public events websocket state.
Entities whose values are derived from the events stream (the
detection booleans and the public event entities) include this in
their availability, since the devices websocket alone cannot tell
whether detections still flow.
"""
success = state is WebsocketState.CONNECTED
if success == self.last_events_update_success:
return
self.last_events_update_success = success
self._async_process_public_updates()
@callback
def _async_process_public_updates(self) -> None:
"""Re-signal public-API entities after a public websocket state change."""
homeassistant/components/unifiprotect/entity.py
+32 -9
View File
@@ -31,6 +31,7 @@ from homeassistant.helpers.entity import Entity, EntityDescription
from .const import (
ATTR_EVENT_ID,
ATTR_EVENT_SCORE,
ATTR_SMART_DETECT_TYPES,
DEFAULT_ATTRIBUTION,
DEFAULT_BRAND,
DOMAIN,
@@ -232,6 +233,9 @@ class BaseProtectEntity(Entity):
# (set ``ufp_public_value``); ``None`` until primed/refreshed.
_ufp_public_obj: PublicDeviceModel | None = None
_ufp_uses_public: bool = False
# Values derived from the public events websocket (detection booleans,
# public event entities) additionally require that websocket to be healthy.
_ufp_requires_events_ws: bool = False
def __init__(
self,
@@ -281,12 +285,19 @@ class BaseProtectEntity(Entity):
# Migrated entities are fully public: availability tracks the public
# websocket health and the public object's state (CONNECTED only;
# CONNECTING/DISCONNECTED/UNKNOWN and a missing object read as
# unavailable), independent of the private connection. An optional
# ``ufp_public_enabled_fn`` gate then mirrors ``ufp_enabled`` against
# the public object (e.g. a sensor feature toggled off).
# unavailable), independent of the private connection. Values fed by
# the events websocket also require it to be healthy — the devices
# websocket keeps the device state fresh, but only the events stream
# carries the detections. An optional ``ufp_public_enabled_fn`` gate
# then mirrors ``ufp_enabled`` against the public object (e.g. a
# sensor feature toggled off).
public_obj = self._ufp_public_obj
if (
self.data.last_public_update_success
and (
not self._ufp_requires_events_ws
or self.data.last_events_update_success
)
and public_obj is not None
and public_obj.state is DeviceState.CONNECTED
):
@@ -355,11 +366,18 @@ class BaseProtectEntity(Entity):
)
# Not every entity carries an entity_description (e.g. cameras), so getattr.
description = getattr(self, "entity_description", None)
if isinstance(description, ProtectEntityDescription) and (
description.ufp_public_value is not None
or description.ufp_public_value_fn is not None
):
self._ufp_uses_public = True
if isinstance(description, ProtectEntityDescription):
if (
description.ufp_public_value is not None
or description.ufp_public_value_fn is not None
):
self._ufp_uses_public = True
if description.ufp_event_driven:
self._ufp_requires_events_ws = True
# ``_ufp_uses_public`` may also be declared as a class attribute by
# entities driven by the public API without a migrated value (the
# public event entities).
if self._ufp_uses_public:
self._ufp_public_obj = self.data.async_get_public_device(self.device)
self.async_on_remove(
self.data.async_subscribe_public(
@@ -429,7 +447,9 @@ class EventEntityMixin(ProtectDeviceEntity):
"""Adds motion event attributes to sensor."""
entity_description: ProtectEventMixin
_unrecorded_attributes = frozenset({ATTR_EVENT_ID, ATTR_EVENT_SCORE})
_unrecorded_attributes = frozenset(
{ATTR_EVENT_ID, ATTR_EVENT_SCORE, ATTR_SMART_DETECT_TYPES}
)
_event: Event | None = None
_event_end: datetime | None = None
@@ -484,6 +504,9 @@ class ProtectEntityDescription(EntityDescription, Generic[T]): # noqa: UP046
ufp_public_value: str | None = None
# Callable variant of ``ufp_public_value`` for public values needing a transform.
ufp_public_value_fn: Callable[[PublicDeviceModel], Any] | None = None
# True when the public value is derived from the events websocket (the
# detection booleans); availability then also tracks that websocket.
ufp_event_driven: bool = False
ufp_enabled: str | None = None
# Public counterpart of ``ufp_enabled``; a callable because public enablement
# is often compound (e.g. mount type plus a settings flag).
homeassistant/components/unifiprotect/event.py
+184 -12
View File
@@ -1,6 +1,7 @@
"""Platform providing event entities for UniFi Protect."""
import dataclasses
import re
from typing import Any, override
from uiprotect import ProtectEvent
@@ -20,6 +21,7 @@ from homeassistant.helpers.event import async_call_at
from . import Bootstrap
from .const import (
ATTR_EVENT_ID,
ATTR_SMART_DETECT_TYPES,
EVENT_TYPE_FINGERPRINT_IDENTIFIED,
EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED,
EVENT_TYPE_NFC_SCANNED,
@@ -42,6 +44,10 @@ from .entity import EventEntityMixin, ProtectDeviceEntity, ProtectEventMixin
PARALLEL_UPDATES = 0
# Per-entity cap on tracked event ids for fire dedup (far above realistic
# concurrent/recent events per camera per category).
_MAX_TRACKED_EVENTS = 16
# Select best thumbnail
# Prefer thumbnails with LPR data, sorted by confidence
@@ -75,7 +81,69 @@ class ProtectEventEntityDescription(ProtectEventMixin, EventEntityDescription):
entity_class: type[ProtectDeviceEntity]
class ProtectDeviceRingEventEntity(EventEntityMixin, ProtectDeviceEntity, EventEntity):
# Protect emits overlapping ``smartDetectZone``, ``smartDetectLine``, and
# ``smartDetectLoiterZone`` frames for the same underlying detection, and a
# line-crossing or loitering detection can arrive as a standalone event of its
# own type — all three carry the same ``smartDetectTypes`` payload per the
# public API schema, so smart-detect entities subscribe to all of them.
_SMART_DETECT_EVENT_TYPES = (
EventType.SMART_DETECT,
EventType.SMART_DETECT_LINE,
EventType.SMART_DETECT_LOITER,
)
@dataclasses.dataclass(frozen=True, kw_only=True)
class ProtectDetectionEventEntityDescription(ProtectEventEntityDescription):
"""Describes a category detection event entity driven by the public events WS."""
ufp_public_event_types: tuple[EventType, ...]
class ProtectDevicePublicEventEntity(
EventEntityMixin, ProtectDeviceEntity, EventEntity
):
"""Base for entities driven by the public events WS.
A detection type can surface at the event start, on a later update, or only
as the event ends, and every non-eviction change is dispatched — so firing is
deduped per ``(event id, event type)``.
Availability follows the public API (device present and connected) plus the
events websocket, which is the only channel these entities fire from.
"""
_ufp_uses_public = True
_ufp_requires_events_ws = True
entity_description: ProtectEventEntityDescription
# A camera can run two overlapping events of the same category whose
# dispatches interleave, so dedup tracks fired types per recent event id
# (bounded), not just the current one.
_fired: dict[str, frozenset[str]] | None = None
@callback
def _fire_once(
self, event: ProtectEvent, event_type: str, event_data: dict[str, Any]
) -> None:
"""Fire ``event_type`` once per event, ignoring repeat dispatches."""
fired = self._fired
if fired is None:
fired = self._fired = {}
# Pop-and-reinsert so any dispatch refreshes this event id's recency; a
# long-running event that keeps updating is then not evicted below.
types = fired.pop(event.id, frozenset())
if event_type in types:
fired[event.id] = types
return
fired[event.id] = types | {event_type}
if len(fired) > _MAX_TRACKED_EVENTS:
del fired[next(iter(fired))] # evict the least-recently-seen event id
self._trigger_event(event_type, event_data)
self.async_write_ha_state()
class ProtectDeviceRingEventEntity(ProtectDevicePublicEventEntity):
"""A UniFi Protect doorbell ring event entity driven by the public events WS."""
entity_description: ProtectEventEntityDescription
@@ -92,8 +160,7 @@ class ProtectDeviceRingEventEntity(EventEntityMixin, ProtectDeviceEntity, EventE
@callback
def _async_ring_event(self, event: ProtectEvent) -> None:
self._trigger_event(DoorbellEventType.RING, {ATTR_EVENT_ID: event.id})
self.async_write_ha_state()
self._fire_once(event, DoorbellEventType.RING, {ATTR_EVENT_ID: event.id})
class ProtectDeviceNFCEventEntity(EventEntityMixin, ProtectDeviceEntity, EventEntity):
@@ -360,9 +427,7 @@ class ProtectDeviceVehicleEventEntity(
self._async_set_thumbnail_timer()
class ProtectDeviceSmartDetectEventEntity(
EventEntityMixin, ProtectDeviceEntity, EventEntity
):
class ProtectDeviceSmartDetectEventEntity(ProtectDevicePublicEventEntity):
"""A UniFi Protect smart-detect event entity driven by the public events WS.
Used for object types that Protect models as discrete, point-in-time
@@ -380,19 +445,102 @@ class ProtectDeviceSmartDetectEventEntity(
async def async_added_to_hass(self) -> None:
"""Subscribe to public smart-detect events for this camera."""
await super().async_added_to_hass()
self.async_on_remove(
self.data.async_subscribe_public_event(
self.device.id, EventType.SMART_DETECT, self._async_smart_detect_event
for event_type in _SMART_DETECT_EVENT_TYPES:
self.async_on_remove(
self.data.async_subscribe_public_event(
self.device.id, event_type, self._async_smart_detect_event
)
)
)
@callback
def _async_smart_detect_event(self, event: ProtectEvent) -> None:
description = self.entity_description
event_types = description.event_types
if event_types and description.ufp_obj_type in event.smart_detect_types:
self._trigger_event(event_types[0], {ATTR_EVENT_ID: event.id})
self.async_write_ha_state()
self._fire_once(event, event_types[0], {ATTR_EVENT_ID: event.id})
_CAMEL_BOUNDARY = re.compile(r"(?<!^)(?=[A-Z])")
# Friendly event-type slugs where the raw enum value is unclear or prefixed;
# unlisted types fall back to a snake_case slug so new ones still auto-surface.
_EVENT_TYPE_OVERRIDES = {
SmartDetectObjectType.SMOKE: "smoke",
SmartDetectObjectType.CMONX: "co",
SmartDetectObjectType.SIREN: "siren",
SmartDetectObjectType.BABY_CRY: "baby_cry",
SmartDetectObjectType.SPEAK: "speaking",
SmartDetectObjectType.BARK: "bark",
SmartDetectObjectType.BURGLAR: "car_alarm",
SmartDetectObjectType.CAR_HORN: "car_horn",
SmartDetectObjectType.GLASS_BREAK: "glass_break",
}
def _event_type(detected: SmartDetectObjectType) -> str:
"""Stable snake_case event type for a detection (HA translation-key rules)."""
return (
_EVENT_TYPE_OVERRIDES.get(detected)
or _CAMEL_BOUNDARY.sub("_", detected.value).lower()
)
_SMART_OBJECT_EVENT_TYPES = [
_event_type(t) for t in SmartDetectObjectType if t.audio_type is None
]
_SMART_AUDIO_EVENT_TYPES = [
_event_type(t) for t in SmartDetectObjectType if t.audio_type is not None
]
class ProtectDeviceDetectionEventEntity(ProtectDevicePublicEventEntity):
"""A camera smart-detect category event entity (object or audio), public WS.
Fires a momentary event for each detected type the entity surfaces. The
``event_types`` are derived from the uiprotect enum, so a new detection type
is surfaced automatically without code changes (only a state label is added).
The subscribed category comes from ``ufp_public_event_types``; the motion
variant overrides the firing.
"""
entity_description: ProtectDetectionEventEntityDescription
@override
async def async_added_to_hass(self) -> None:
"""Subscribe to the category's public detection events."""
await super().async_added_to_hass()
for event_type in self.entity_description.ufp_public_event_types:
self.async_on_remove(
self.data.async_subscribe_public_event(
self.device.id,
event_type,
self._async_detection_event,
)
)
@callback
def _async_detection_event(self, event: ProtectEvent) -> None:
allowed = self.entity_description.event_types or ()
# One fire per detected type so each stays independently automatable
# (incl. types with no binary sensor); carries the co-detected set known
# at fire time (types can still arrive on a later update).
detected = [_event_type(t) for t in event.smart_detect_types]
for event_type in detected:
if event_type in allowed:
self._fire_once(
event,
event_type,
{ATTR_EVENT_ID: event.id, ATTR_SMART_DETECT_TYPES: detected},
)
class ProtectDeviceMotionEventEntity(ProtectDeviceDetectionEventEntity):
"""A camera motion-detection event entity (public events WS)."""
@callback
@override
def _async_detection_event(self, event: ProtectEvent) -> None:
self._fire_once(event, EventType.MOTION.value, {ATTR_EVENT_ID: event.id})
EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = (
@@ -439,6 +587,30 @@ EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = (
event_types=[EVENT_TYPE_PACKAGE_DETECTED],
entity_class=ProtectDeviceSmartDetectEventEntity,
),
ProtectDetectionEventEntityDescription(
key="motion_detection",
translation_key="motion_detection",
device_class=EventDeviceClass.MOTION,
event_types=[EventType.MOTION.value],
ufp_public_event_types=(EventType.MOTION,),
entity_class=ProtectDeviceMotionEventEntity,
),
ProtectDetectionEventEntityDescription(
key="smart_detection",
translation_key="smart_detection",
ufp_required_field="feature_flags.has_smart_detect",
event_types=_SMART_OBJECT_EVENT_TYPES,
ufp_public_event_types=_SMART_DETECT_EVENT_TYPES,
entity_class=ProtectDeviceDetectionEventEntity,
),
ProtectDetectionEventEntityDescription(
key="sound_detection",
translation_key="sound_detection",
ufp_required_field="feature_flags.smart_detect_audio_types",
event_types=_SMART_AUDIO_EVENT_TYPES,
ufp_public_event_types=(EventType.SMART_AUDIO_DETECT,),
entity_class=ProtectDeviceDetectionEventEntity,
),
)
homeassistant/components/unifiprotect/icons.json
+9
View File
@@ -181,12 +181,21 @@
"fingerprint": {
"default": "mdi:fingerprint"
},
"motion_detection": {
"default": "mdi:motion-sensor"
},
"nfc": {
"default": "mdi:nfc"
},
"package": {
"default": "mdi:package-variant-closed"
},
"smart_detection": {
"default": "mdi:cctv"
},
"sound_detection": {
"default": "mdi:waveform"
},
"vehicle": {
"default": "mdi:car"
}
homeassistant/components/unifiprotect/light.py
+3 -12
View File
@@ -61,18 +61,9 @@ class ProtectLight(ProtectDeviceEntity, LightEntity):
_attr_color_mode = ColorMode.BRIGHTNESS
_attr_supported_color_modes = {ColorMode.BRIGHTNESS}
_state_attrs = ("_attr_available", "_attr_is_on", "_attr_brightness")
@override
async def async_added_to_hass(self) -> None:
"""Read state from the public API (primed before the first update)."""
self._ufp_uses_public = True
self._ufp_public_obj = self.data.async_get_public_device(self.device)
self.async_on_remove(
self.data.async_subscribe_public(
self.device.mac, self._async_public_updated
)
)
await super().async_added_to_hass()
# State comes from the public API; the base class primes the object and
# subscribes to the public devices websocket on this flag.
_ufp_uses_public = True
@callback
@override
homeassistant/components/unifiprotect/strings.json
+45
View File
@@ -289,6 +289,16 @@
}
}
},
"motion_detection": {
"name": "Motion detection",
"state_attributes": {
"event_type": {
"state": {
"motion": "Motion"
}
}
}
},
"nfc": {
"name": "NFC",
"state_attributes": {
@@ -309,6 +319,41 @@
}
}
},
"smart_detection": {
"name": "Smart detection",
"state_attributes": {
"event_type": {
"state": {
"animal": "Animal",
"car": "Car",
"face": "Face",
"license_plate": "License plate",
"package": "Package",
"person": "Person",
"pet": "Pet",
"vehicle": "Vehicle"
}
}
}
},
"sound_detection": {
"name": "Sound detection",
"state_attributes": {
"event_type": {
"state": {
"baby_cry": "Baby cry",
"bark": "Barking",
"car_alarm": "Car alarm",
"car_horn": "Car horn",
"co": "CO alarm",
"glass_break": "Glass break",
"siren": "Siren",
"smoke": "Smoke alarm",
"speaking": "Speaking"
}
}
}
},
"vehicle": {
"name": "Vehicle",
"state_attributes": {
tests/components/unifiprotect/conftest.py
+7
View File
@@ -254,11 +254,18 @@ def mock_entry(
ufp.devices_ws_state_subscription = ws_state_subscription
return Mock()
def subscribe_events_websocket_state(
ws_state_subscription: Callable[[WebsocketState], None],
) -> Any:
ufp.events_ws_state_subscription = ws_state_subscription
return Mock()
ufp_client.subscribe_websocket = subscribe
ufp_client.subscribe_websocket_state = subscribe_websocket_state
ufp_client.subscribe_devices_websocket = subscribe_devices_websocket
ufp_client.subscribe_events = subscribe_events
ufp_client.subscribe_devices_websocket_state = subscribe_devices_websocket_state
ufp_client.subscribe_events_websocket_state = subscribe_events_websocket_state
async def update_public() -> Any:
# Mirror the library prime: build each camera's public model from the
tests/components/unifiprotect/test_binary_sensor.py
+197 -304
View File
@@ -1,5 +1,6 @@
"""Test the UniFi Protect binary_sensor platform."""
from collections.abc import Callable
from datetime import datetime, timedelta
from unittest.mock import Mock
@@ -13,7 +14,7 @@ from uiprotect.data import (
ModelType,
MountType,
Sensor,
SmartDetectObjectType,
SmartDetectAudioType,
)
from uiprotect.data.public_devices import SensorFeatureCapability
from uiprotect.websocket import WebsocketState
@@ -27,11 +28,8 @@ from homeassistant.components.unifiprotect.binary_sensor import (
SENSE_SENSORS,
ProtectBinaryEntityDescription,
)
from homeassistant.components.unifiprotect.const import (
ATTR_EVENT_SCORE,
DEFAULT_ATTRIBUTION,
DOMAIN,
)
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION, DOMAIN
from homeassistant.components.unifiprotect.number import CAMERA_NUMBERS
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_DEVICE_CLASS,
@@ -41,7 +39,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
Platform,
)
from homeassistant.core import Event as HAEvent, EventStateChangedData, HomeAssistant
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .utils import (
@@ -50,10 +48,12 @@ from .utils import (
assert_entity_counts,
ids_from_device_description,
init_entry,
make_public_camera,
make_public_light,
make_public_sensor,
public_device_ws_message,
remove_entities,
setup_public_camera,
setup_public_light,
setup_public_sensor,
)
@@ -76,11 +76,11 @@ async def test_binary_sensor_camera_remove(
ufp.api.bootstrap.nvr.system_info.ustorage = None
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 8, 6)
assert_entity_counts(hass, Platform.BINARY_SENSOR, 7, 6)
await remove_entities(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 0, 0)
await adopt_devices(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 8, 6)
assert_entity_counts(hass, Platform.BINARY_SENSOR, 7, 6)
async def test_binary_sensor_light_remove(
@@ -148,8 +148,9 @@ async def test_binary_sensor_setup_camera_all(
"""Test binary_sensor entity setup for camera devices (all features)."""
ufp.api.bootstrap.nvr.system_info.ustorage = None
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 8, 6)
assert_entity_counts(hass, Platform.BINARY_SENSOR, 7, 6)
description = EVENT_SENSORS[0]
unique_id, entity_id = await ids_from_device_description(
@@ -180,8 +181,8 @@ async def test_binary_sensor_setup_camera_all(
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
# Motion
description = EVENT_SENSORS[1]
# Motion (migrated to the public path, available via setup_public_camera)
description = next(d for d in CAMERA_SENSORS if d.key == "motion")
unique_id, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, doorbell, description
)
@@ -685,49 +686,73 @@ async def test_binary_sensor_update_motion(
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test binary_sensor motion entity."""
"""Test the migrated motion binary sensor reads sustained state from the public API."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 14, 12)
assert_entity_counts(hass, Platform.BINARY_SENSOR, 13, 12)
motion = next(d for d in CAMERA_SENSORS if d.key == "motion")
_, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, doorbell, EVENT_SENSORS[1]
hass, Platform.BINARY_SENSOR, doorbell, motion
)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.MOTION,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
)
new_camera = doorbell.model_copy()
new_camera.is_motion_detected = True
new_camera.last_motion_event_id = 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()
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
assert state.state == STATE_OFF
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_SCORE] == 100
ufp.devices_ws_subscription(
public_device_ws_message(make_public_camera(doorbell, is_motion_detected=True))
)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
# Detection ends -> sustained state clears.
ufp.devices_ws_subscription(public_device_ws_message(make_public_camera(doorbell)))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF
async def test_binary_sensor_detection_unavailable_on_events_ws_disconnect(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
) -> None:
"""Detection sensors follow the events websocket their values derive from.
A migrated value fed by the devices websocket (microphone level) must not
be affected by an events websocket outage.
"""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
motion = next(d for d in CAMERA_SENSORS if d.key == "motion")
_, motion_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, doorbell, motion
)
mic_level = next(d for d in CAMERA_NUMBERS if d.key == "mic_level")
_, mic_id = await ids_from_device_description(
hass, Platform.NUMBER, doorbell, mic_level
)
assert hass.states.get(motion_id).state == STATE_OFF
assert hass.states.get(mic_id).state != STATE_UNAVAILABLE
assert ufp.events_ws_state_subscription is not None
ufp.events_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert hass.states.get(motion_id).state == STATE_UNAVAILABLE
assert hass.states.get(mic_id).state != STATE_UNAVAILABLE
ufp.events_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
assert hass.states.get(motion_id).state == STATE_OFF
async def test_binary_sensor_update_light_motion(
@@ -837,37 +862,108 @@ async def test_binary_sensor_person_detected(
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test binary_sensor person detected detection entity."""
"""Test the migrated person-detection binary sensor over the public API."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 14, 14)
doorbell.smart_detect_settings.object_types.append(SmartDetectObjectType.PERSON)
assert_entity_counts(hass, Platform.BINARY_SENSOR, 13, 13)
person = next(d for d in CAMERA_SENSORS if d.key == "smart_obj_person")
_, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, doorbell, EVENT_SENSORS[3]
hass, Platform.BINARY_SENSOR, doorbell, person
)
events = async_capture_events(hass, EVENT_STATE_CHANGED)
assert hass.states.get(entity_id).state == STATE_OFF
# Person detection starts (camera update pushed on the public devices WS).
ufp.devices_ws_subscription(
public_device_ws_message(
make_public_camera(
doorbell,
is_smart_currently_detected=True,
is_person_currently_detected=True,
)
)
)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
# Detection ends -> sustained state clears.
ufp.devices_ws_subscription(public_device_ws_message(make_public_camera(doorbell)))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("key", "make_disabled"),
[
("smart_obj_person", lambda c: make_public_camera(c, object_types=[])),
("smart_audio_smoke", lambda c: make_public_camera(c, audio_types=[])),
],
)
async def test_binary_sensor_detection_disabled_unavailable(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
key: str,
make_disabled: Callable[[Camera], Mock],
) -> None:
"""A migrated detection binary is unavailable when its type is disabled in Protect."""
# Ensure the audio-alarm capability so the smoke binary is created.
doorbell.feature_flags.smart_detect_audio_types = [SmartDetectAudioType.SMOKE]
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in CAMERA_SENSORS if d.key == key)
_, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, doorbell, description
)
assert hass.states.get(entity_id).state == STATE_OFF
# The detection type is turned off in Protect -> the enabled gate fails.
ufp.devices_ws_subscription(public_device_ws_message(make_disabled(doorbell)))
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_binary_sensor_doorbell_ring(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""The doorbell occupancy binary stays on the private ring event path."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_SENSORS if d.key == "doorbell")
_, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, doorbell, description
)
assert hass.states.get(entity_id).state == STATE_OFF
state_changes = async_capture_events(hass, EVENT_STATE_CHANGED)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.SMART_DETECT,
id="ring-1",
type=EventType.RING,
start=fixed_now - timedelta(seconds=1),
end=None,
score=50,
end=fixed_now,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
)
new_camera = doorbell.model_copy()
new_camera.is_smart_detected = True
new_camera.last_ring_event_id = event.id
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {event.id: event}
@@ -875,106 +971,12 @@ async def test_binary_sensor_person_detected(
mock_msg.changed_data = {}
mock_msg.new_obj = event
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.SMART_DETECT,
start=fixed_now - timedelta(seconds=1),
end=fixed_now + timedelta(seconds=1),
score=65,
smart_detect_types=[SmartDetectObjectType.PERSON],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
)
new_camera = doorbell.model_copy()
new_camera.is_smart_detected = True
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.PERSON] = 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()
entity_events = [event for event in events if event.data["entity_id"] == entity_id]
assert len(entity_events) == 3
assert entity_events[0].data["new_state"].state == STATE_OFF
assert entity_events[1].data["new_state"].state == STATE_ON
assert entity_events[2].data["new_state"].state == STATE_OFF
# Event is already seen and has end, should now be off
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
# Now send an event that has an end right away
event = Event(
model=ModelType.EVENT,
id="new_event_id",
type=EventType.SMART_DETECT,
start=fixed_now - timedelta(seconds=1),
end=fixed_now + timedelta(seconds=1),
score=80,
smart_detect_types=[SmartDetectObjectType.PERSON],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
)
new_camera = doorbell.model_copy()
new_camera.is_smart_detected = True
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.PERSON] = 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
state_changes: list[HAEvent[EventStateChangedData]] = async_capture_events(
hass, EVENT_STATE_CHANGED
)
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_OFF
assert len(state_changes) == 2
on_event = state_changes[0]
state = on_event.data["new_state"]
assert state
assert state.state == STATE_ON
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_SCORE] == 80
off_event = state_changes[1]
state = off_event.data["new_state"]
assert state
assert state.state == STATE_OFF
assert ATTR_EVENT_SCORE not in state.attributes
# replay and ensure ignored
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
assert len(state_changes) == 2
# A momentary ring blips on, then immediately clears.
ring_changes = [e for e in state_changes if e.data["entity_id"] == entity_id]
assert any(c.data["new_state"].state == STATE_ON for c in ring_changes)
assert hass.states.get(entity_id).state == STATE_OFF
async def test_aiport_no_binary_sensor_entities(
@@ -1003,171 +1005,62 @@ async def test_binary_sensor_simultaneous_person_and_vehicle_detection(
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test that when an event is updated with additional detection types, both trigger.
"""Person and vehicle detected at once both report ON.
This is a regression test for https://github.com/home-assistant/core/issues/152133
where an event starting with vehicle detection gets updated to also include person
detection (e.g., someone getting out of a car). Both sensors should be ON
simultaneously, not queued.
Regression for https://github.com/home-assistant/core/issues/152133 (a second
type added to an ongoing detection): on the public path each type's sustained
state is derived independently by the library, so adding person to an ongoing
vehicle detection turns both ON without queueing.
"""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.BINARY_SENSOR, 14, 14)
assert_entity_counts(hass, Platform.BINARY_SENSOR, 13, 13)
doorbell.smart_detect_settings.object_types.append(SmartDetectObjectType.PERSON)
doorbell.smart_detect_settings.object_types.append(SmartDetectObjectType.VEHICLE)
# Get entity IDs for both person and vehicle detection
person = next(d for d in CAMERA_SENSORS if d.key == "smart_obj_person")
vehicle = next(d for d in CAMERA_SENSORS if d.key == "smart_obj_vehicle")
_, person_entity_id = await ids_from_device_description(
hass,
Platform.BINARY_SENSOR,
doorbell,
EVENT_SENSORS[3], # person detected
hass, Platform.BINARY_SENSOR, doorbell, person
)
_, vehicle_entity_id = await ids_from_device_description(
hass,
Platform.BINARY_SENSOR,
doorbell,
EVENT_SENSORS[4], # vehicle detected
hass, Platform.BINARY_SENSOR, doorbell, vehicle
)
# Step 1: Initial event with only VEHICLE detection (car arriving)
event = Event(
model=ModelType.EVENT,
id="combined_event_id",
type=EventType.SMART_DETECT,
start=fixed_now - timedelta(seconds=5),
end=None, # Event is ongoing
score=90,
smart_detect_types=[SmartDetectObjectType.VEHICLE],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
# Vehicle arrives.
ufp.devices_ws_subscription(
public_device_ws_message(
make_public_camera(
doorbell,
is_smart_currently_detected=True,
is_vehicle_currently_detected=True,
)
)
)
new_camera = doorbell.model_copy()
new_camera.is_smart_detected = True
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.VEHICLE] = 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()
# Vehicle sensor should be ON
vehicle_state = hass.states.get(vehicle_entity_id)
assert vehicle_state
assert vehicle_state.state == STATE_ON, "Vehicle detection should be ON"
assert hass.states.get(vehicle_entity_id).state == STATE_ON
assert hass.states.get(person_entity_id).state == STATE_OFF
# Person sensor should still be OFF (no person detected yet)
person_state = hass.states.get(person_entity_id)
assert person_state
assert person_state.state == STATE_OFF, "Person detection should be OFF initially"
# Step 2: Same event gets updated to include PERSON detection
# (someone gets out of the car - Protect adds PERSON to the same event)
#
# BUG SCENARIO: UniFi Protect updates the event to include PERSON in
# smart_detect_types, BUT does NOT update last_smart_detect_event_ids[PERSON]
# until the event ends. This is the core issue reported in #152133.
updated_event = Event(
model=ModelType.EVENT,
id="combined_event_id", # Same event ID!
type=EventType.SMART_DETECT,
start=fixed_now - timedelta(seconds=5),
end=None, # Event still ongoing
score=90,
smart_detect_types=[
SmartDetectObjectType.VEHICLE,
SmartDetectObjectType.PERSON,
],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
# Person joins the same scene -> both ON simultaneously.
ufp.devices_ws_subscription(
public_device_ws_message(
make_public_camera(
doorbell,
is_smart_currently_detected=True,
is_vehicle_currently_detected=True,
is_person_currently_detected=True,
)
)
)
# IMPORTANT: The camera's last_smart_detect_event_ids is NOT updated for PERSON!
# This simulates the real bug where UniFi Protect doesn't immediately update
# the camera's last_smart_detect_event_ids when a new detection type is added
# to an ongoing event.
new_camera = doorbell.model_copy()
new_camera.is_smart_detected = True
# Only VEHICLE has the event ID - PERSON does not (simulating the bug)
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.VEHICLE] = (
updated_event.id
)
# NOTE: We're NOT setting last_smart_detect_event_ids[PERSON] to simulate the bug!
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {updated_event.id: updated_event}
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = updated_event
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
# CRITICAL: Both sensors should now be ON simultaneously
vehicle_state = hass.states.get(vehicle_entity_id)
assert vehicle_state
assert vehicle_state.state == STATE_ON, (
"Vehicle detection should still be ON after event update"
)
person_state = hass.states.get(person_entity_id)
assert person_state
assert person_state.state == STATE_ON, (
"Person detection should be ON immediately when added to event, "
"not waiting for vehicle detection to end"
)
# Verify both have correct attributes
assert vehicle_state.attributes[ATTR_EVENT_SCORE] == 90
assert person_state.attributes[ATTR_EVENT_SCORE] == 90
# Step 3: Event ends - both sensors should turn OFF
ended_event = Event(
model=ModelType.EVENT,
id="combined_event_id",
type=EventType.SMART_DETECT,
start=fixed_now - timedelta(seconds=5),
end=fixed_now, # Event ended now
score=90,
smart_detect_types=[
SmartDetectObjectType.VEHICLE,
SmartDetectObjectType.PERSON,
],
smart_detect_event_ids=[],
camera_id=doorbell.id,
api=ufp.api,
)
ufp.api.bootstrap.events = {ended_event.id: ended_event}
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = ended_event
ufp.ws_msg(mock_msg)
assert hass.states.get(vehicle_entity_id).state == STATE_ON
assert hass.states.get(person_entity_id).state == STATE_ON
# Scene clears -> both OFF.
ufp.devices_ws_subscription(public_device_ws_message(make_public_camera(doorbell)))
await hass.async_block_till_done()
# Both should be OFF now
vehicle_state = hass.states.get(vehicle_entity_id)
assert vehicle_state
assert vehicle_state.state == STATE_OFF, (
"Vehicle detection should be OFF after event ends"
)
person_state = hass.states.get(person_entity_id)
assert person_state
assert person_state.state == STATE_OFF, (
"Person detection should be OFF after event ends"
)
assert hass.states.get(vehicle_entity_id).state == STATE_OFF
assert hass.states.get(person_entity_id).state == STATE_OFF
tests/components/unifiprotect/test_event.py
+574 -48
View File
@@ -2,6 +2,8 @@
import asyncio
from datetime import datetime, timedelta
import json
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
@@ -12,18 +14,24 @@ from uiprotect.data import (
Event,
EventType,
ModelType,
PublicBootstrap,
SmartDetectAudioType,
SmartDetectObjectType,
)
from uiprotect.websocket import WebsocketState
from homeassistant.components.unifiprotect.const import (
ATTR_EVENT_ID,
ATTR_SMART_DETECT_TYPES,
DEFAULT_ATTRIBUTION,
EVENT_TYPE_PACKAGE_DETECTED,
)
from homeassistant.components.unifiprotect.event import EVENT_DESCRIPTIONS
from homeassistant.const import ATTR_ATTRIBUTION, Platform
from homeassistant.components.unifiprotect.event import (
_MAX_TRACKED_EVENTS,
EVENT_DESCRIPTIONS,
)
from homeassistant.const import ATTR_ATTRIBUTION, STATE_UNAVAILABLE, Platform
from homeassistant.core import Event as HAEvent, HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.event import async_track_state_change_event
from .utils import (
@@ -33,6 +41,7 @@ from .utils import (
ids_from_device_description,
init_entry,
remove_entities,
setup_public_camera,
)
# Short delay for testing
@@ -56,11 +65,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, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
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, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
async def test_doorbell_ring(
@@ -73,18 +82,12 @@ async def test_doorbell_ring(
"""Test a doorbell ring event fired from the public events websocket."""
# Ring is delivered over the public events websocket, which is only
# subscribed once update_public() has primed the public bootstrap.
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = Mock(
spec=PublicBootstrap,
relays={},
sirens={},
arm_mode=None,
arm_profiles={},
)
# subscribed once update_public() has primed the public bootstrap; the
# entity's availability also requires the public camera to resolve.
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -133,7 +136,8 @@ async def test_doorbell_ring(
await hass.async_block_till_done()
assert len(events) == 1
# Only the start of an event is dispatched; an update must be ignored.
# Updates are dispatched too, but the entity fires each event id only
# once, so a repeat dispatch of the same ring event must be suppressed.
ufp.events_msg(
ProtectEvent(
id="test_ring_event",
@@ -151,28 +155,31 @@ async def test_doorbell_ring(
unsub()
@pytest.mark.parametrize(
"event_type",
[
pytest.param(EventType.SMART_DETECT, id="zone"),
pytest.param(EventType.SMART_DETECT_LINE, id="line"),
pytest.param(EventType.SMART_DETECT_LOITER, id="loiter"),
],
)
async def test_package_detected(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
event_type: EventType,
) -> None:
"""Test a package detection event fired from the public events websocket."""
# Package detection is delivered over the public events websocket, which is
# only subscribed once update_public() has primed the public bootstrap.
ufp.api.has_public_bootstrap = True
ufp.api.public_bootstrap = Mock(
spec=PublicBootstrap,
relays={},
sirens={},
arm_mode=None,
arm_profiles={},
)
# only subscribed once update_public() has primed the public bootstrap; the
# entity's availability also requires the public camera to resolve.
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -186,12 +193,13 @@ async def test_package_detected(
unsub = async_track_state_change_event(hass, entity_id, _capture_event)
# Package detection arrives on the public events websocket as a
# smartDetectZone detection event with the package object type. Protect
# records it already-ended; the event entity fires on the detection start.
# smartDetectZone, smartDetectLine, or smartDetectLoiterZone detection
# event with the package object type. Protect records it already-ended;
# the event entity fires on the detection start.
ufp.events_msg(
ProtectEvent(
id="test_package_event",
type=EventType.SMART_DETECT,
type=event_type,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
@@ -227,7 +235,8 @@ async def test_package_detected(
await hass.async_block_till_done()
assert len(events) == 1
# Only the start of a detection is dispatched; an update must be ignored.
# Updates are dispatched too, but the entity fires each (event id, type)
# once, so a repeat dispatch of the same package event must be suppressed.
ufp.events_msg(
ProtectEvent(
id="test_package_event",
@@ -315,7 +324,7 @@ async def test_doorbell_nfc_scanned(
"""Test a doorbell NFC scanned event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -390,7 +399,7 @@ async def test_doorbell_nfc_scanned_ulpusr_deactivated(
"""Test a doorbell NFC scanned event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -466,7 +475,7 @@ async def test_doorbell_nfc_scanned_no_ulpusr(
"""Test a doorbell NFC scanned event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -534,7 +543,7 @@ async def test_doorbell_nfc_scanned_no_keyring(
"""Test a doorbell NFC scanned event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -595,7 +604,7 @@ async def test_doorbell_fingerprint_identified(
"""Test a doorbell fingerprint identified event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -663,7 +672,7 @@ async def test_doorbell_fingerprint_identified_user_deactivated(
"""Test a doorbell fingerprint identified event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -732,7 +741,7 @@ async def test_doorbell_fingerprint_identified_no_user(
"""Test a doorbell fingerprint identified event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -793,7 +802,7 @@ async def test_doorbell_fingerprint_not_identified(
"""Test a doorbell fingerprint identified event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -851,7 +860,7 @@ async def test_vehicle_detection_basic(
"""Test basic vehicle detection event with thumbnails."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -925,7 +934,7 @@ async def test_vehicle_detection_with_lpr_ufp6(
"""Test vehicle detection with license plate recognition (UFP 6.0+ format)."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1009,7 +1018,7 @@ async def test_vehicle_detection_with_lpr_legacy(
"""Test vehicle detection with license plate recognition (legacy format)."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1082,7 +1091,7 @@ async def test_vehicle_detection_multiple_thumbnails(
"""Test vehicle detection with multiple thumbnails - should pick best LPR."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1183,7 +1192,7 @@ async def test_vehicle_detection_no_thumbnails(
"""Test vehicle detection event without thumbnails - should not fire."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1241,7 +1250,7 @@ async def test_vehicle_detection_timer_reset_on_new_thumbnail(
"""Test that timer resets when new thumbnails arrive for same event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1352,7 +1361,7 @@ async def test_vehicle_detection_new_event_cancels_timer(
"""Test that new event cancels timer for previous event."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1479,7 +1488,7 @@ async def test_vehicle_detection_timer_cleanup_on_remove(
"""Test that pending timer is cancelled when entity is removed."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3]
@@ -1544,7 +1553,7 @@ async def test_vehicle_detection_refire_on_lpr_data(
"""Test that event refires when LPR data arrives after initial detection."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1655,7 +1664,7 @@ async def test_vehicle_detection_no_refire_same_data(
"""Test that event does NOT refire when same data arrives again."""
await init_entry(hass, ufp, [doorbell, unadopted_camera])
assert_entity_counts(hass, Platform.EVENT, 5, 5)
assert_entity_counts(hass, Platform.EVENT, 7, 7)
events: list[HAEvent] = []
@callback
@@ -1740,3 +1749,520 @@ async def test_aiport_no_event_entities(
# AI Port should not create any camera-specific event entities
# (doorbell, motion, etc.)
assert_entity_counts(hass, Platform.EVENT, 0, 0)
async def test_motion_detection_event(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""The motion event entity fires from the public events websocket."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "motion_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
ufp.events_msg(
ProtectEvent(
id="motion-1",
type=EventType.MOTION,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
start=fixed_now - timedelta(seconds=1),
end=fixed_now,
),
EventChange.STARTED,
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.attributes["event_type"] == "motion"
assert state.attributes[ATTR_EVENT_ID] == "motion-1"
@pytest.mark.parametrize(
"event_type",
[
pytest.param(EventType.SMART_DETECT, id="zone"),
pytest.param(EventType.SMART_DETECT_LINE, id="line"),
pytest.param(EventType.SMART_DETECT_LOITER, id="loiter"),
],
)
async def test_smart_detection_event(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
event_type: EventType,
) -> None:
"""The smart-detection event entity fires per object type with the full type set.
smartDetectZone, smartDetectLine, and smartDetectLoiterZone events all carry
smart detections, so a standalone line-crossing or loitering event must fire
the entity too.
"""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "smart_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, entity_id, _capture)
ufp.events_msg(
ProtectEvent(
id="smart-1",
type=event_type,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
start=fixed_now - timedelta(seconds=1),
end=fixed_now,
smart_detect_types=(
SmartDetectObjectType.PERSON,
SmartDetectObjectType.VEHICLE,
),
),
EventChange.STARTED,
)
await hass.async_block_till_done()
unsub()
# One fire per surfaced type, each carrying the full co-detected set.
fired = [event.data["new_state"].attributes["event_type"] for event in events]
assert fired == ["person", "vehicle"]
last = events[-1].data["new_state"]
assert last.attributes[ATTR_EVENT_ID] == "smart-1"
assert last.attributes[ATTR_SMART_DETECT_TYPES] == ["person", "vehicle"]
async def test_sound_detection_event(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""The sound-detection event entity fires for audio types (slugged event type)."""
doorbell.feature_flags.smart_detect_audio_types = [SmartDetectAudioType.SMOKE]
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "sound_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
ufp.events_msg(
ProtectEvent(
id="audio-1",
type=EventType.SMART_AUDIO_DETECT,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
start=fixed_now - timedelta(seconds=1),
end=fixed_now,
smart_detect_types=(SmartDetectObjectType.SMOKE,),
),
EventChange.STARTED,
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.attributes["event_type"] == "smoke"
assert state.attributes[ATTR_EVENT_ID] == "audio-1"
async def test_sound_detection_event_late_type(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Audio types arrive on a later update, not at start; fire once when they appear."""
doorbell.feature_flags.smart_detect_audio_types = [SmartDetectAudioType.SMOKE]
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "sound_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, entity_id, _capture)
base = {
"id": "audio-1",
"type": EventType.SMART_AUDIO_DETECT,
"channel": ProtectEventChannel.DETECTION,
"device_id": doorbell.id,
"device_mac": doorbell.mac,
"start": fixed_now - timedelta(seconds=1),
}
# Start carries no type yet -> nothing fires.
ufp.events_msg(
ProtectEvent(**base, end=None, smart_detect_types=()), EventChange.STARTED
)
await hass.async_block_till_done()
assert events == []
# The type appears on a later update -> fires once.
ufp.events_msg(
ProtectEvent(
**base, end=None, smart_detect_types=(SmartDetectObjectType.SMOKE,)
),
EventChange.UPDATED,
)
await hass.async_block_till_done()
# A further update for the same type is deduped (no re-fire).
ufp.events_msg(
ProtectEvent(
**base, end=fixed_now, smart_detect_types=(SmartDetectObjectType.SMOKE,)
),
EventChange.UPDATED,
)
await hass.async_block_till_done()
unsub()
fired = [event.data["new_state"].attributes["event_type"] for event in events]
assert fired == ["smoke"]
async def test_sound_detection_absent_without_audio_types(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
) -> None:
"""Object smart-detect without audio support creates no sound-detection entity."""
doorbell.feature_flags.smart_detect_audio_types = []
await init_entry(hass, ufp, [doorbell])
entity_registry = er.async_get(hass)
smart = next(d for d in EVENT_DESCRIPTIONS if d.key == "smart_detection")
sound = next(d for d in EVENT_DESCRIPTIONS if d.key == "sound_detection")
_, smart_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, smart
)
_, sound_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, sound
)
# object detection stays, audio (sound) detection is gated out
assert entity_registry.async_get(smart_id) is not None
assert entity_registry.async_get(sound_id) is None
async def test_detection_event_removed_change_ignored(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""A REMOVED (eviction) change does not fire a detection event."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "smart_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, entity_id, _capture)
ufp.events_msg(
ProtectEvent(
id="ev-removed",
type=EventType.SMART_DETECT,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
start=fixed_now - timedelta(seconds=1),
end=fixed_now,
smart_detect_types=(SmartDetectObjectType.PERSON,),
),
EventChange.REMOVED,
)
await hass.async_block_till_done()
unsub()
assert events == []
async def test_doorbell_ring_dedup_across_dispatches(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""A ring fires once even though start, update and end are all dispatched."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[0]
)
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, entity_id, _capture)
base = {
"id": "ring-1",
"type": EventType.RING,
"channel": ProtectEventChannel.DETECTION,
"device_id": doorbell.id,
"device_mac": doorbell.mac,
"start": fixed_now - timedelta(seconds=1),
}
for change, end in (
(EventChange.STARTED, None),
(EventChange.UPDATED, None),
(EventChange.ENDED, fixed_now),
):
ufp.events_msg(ProtectEvent(**base, end=end), change)
await hass.async_block_till_done()
unsub()
assert len(events) == 1
def test_detection_event_types_have_translations() -> None:
"""Every category detection event type has a strings.json state label.
Guards against a uiprotect enum addition slugging into ``event_types`` (and
firing) without a matching translation label.
"""
strings = json.loads(
(
Path(__file__).parents[3]
/ "homeassistant/components/unifiprotect/strings.json"
).read_text()
)
event_states = strings["entity"]["event"]
for key in ("motion_detection", "smart_detection", "sound_detection"):
description = next(d for d in EVENT_DESCRIPTIONS if d.key == key)
labels = event_states[key]["state_attributes"]["event_type"]["state"]
missing = [t for t in description.event_types or () if t not in labels]
assert not missing, f"{key} missing event_type labels: {missing}"
async def test_smart_detection_event_interleaved_dedup(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Two overlapping same-category events whose dispatches interleave don't re-fire."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "smart_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, entity_id, _capture)
common = {
"type": EventType.SMART_DETECT,
"channel": ProtectEventChannel.DETECTION,
"device_id": doorbell.id,
"device_mac": doorbell.mac,
"start": fixed_now - timedelta(seconds=1),
"end": None,
}
# Event A and B overlap; A re-dispatches after B started.
ufp.events_msg(
ProtectEvent(
id="evt-a", smart_detect_types=(SmartDetectObjectType.PERSON,), **common
),
EventChange.STARTED,
)
ufp.events_msg(
ProtectEvent(
id="evt-b", smart_detect_types=(SmartDetectObjectType.VEHICLE,), **common
),
EventChange.STARTED,
)
ufp.events_msg(
ProtectEvent(
id="evt-a", smart_detect_types=(SmartDetectObjectType.PERSON,), **common
),
EventChange.UPDATED,
)
await hass.async_block_till_done()
unsub()
# A's person is not re-fired when its update arrives after B's dispatch.
fired = [event.data["new_state"].attributes["event_type"] for event in events]
assert fired == ["person", "vehicle"]
async def test_detection_event_dedup_is_bounded(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""The fire-dedup tracker is bounded; distinct events keep firing."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "motion_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, entity_id, _capture)
for index in range(20):
ufp.events_msg(
ProtectEvent(
id=f"motion-{index}",
type=EventType.MOTION,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
start=fixed_now - timedelta(seconds=1),
end=fixed_now,
),
EventChange.STARTED,
)
await hass.async_block_till_done()
unsub()
assert len(events) == 20
async def test_detection_event_dedup_evicts_oldest(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Exceeding the dedup cap evicts the oldest event id, letting it refire.
An unbounded tracker would also pass ``test_detection_event_dedup_is_bounded``
(it only sends distinct ids), so this replays an id that should have aged out.
"""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "motion_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, description
)
events: list[HAEvent] = []
@callback
def _capture(event: HAEvent) -> None:
events.append(event)
unsub = async_track_state_change_event(hass, entity_id, _capture)
def _send(index: int) -> None:
ufp.events_msg(
ProtectEvent(
id=f"motion-{index}",
type=EventType.MOTION,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
start=fixed_now - timedelta(seconds=1),
end=fixed_now,
),
EventChange.STARTED,
)
# A (cap + 1)th distinct id evicts the oldest tracked id ("motion-0").
for index in range(_MAX_TRACKED_EVENTS + 1):
_send(index)
await hass.async_block_till_done()
assert len(events) == _MAX_TRACKED_EVENTS + 1
# Replaying the evicted id fires again; a bug that never evicts would dedup it.
_send(0)
await hass.async_block_till_done()
unsub()
assert len(events) == _MAX_TRACKED_EVENTS + 2
async def test_event_entities_unavailable_on_events_ws_disconnect(
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
unadopted_camera: Camera,
) -> None:
"""Public event entities follow the events websocket they fire from."""
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
ring = next(d for d in EVENT_DESCRIPTIONS if d.key == "doorbell")
_, ring_id = await ids_from_device_description(hass, Platform.EVENT, doorbell, ring)
motion = next(d for d in EVENT_DESCRIPTIONS if d.key == "motion_detection")
_, motion_id = await ids_from_device_description(
hass, Platform.EVENT, doorbell, motion
)
assert hass.states.get(ring_id).state != STATE_UNAVAILABLE
assert hass.states.get(motion_id).state != STATE_UNAVAILABLE
assert ufp.events_ws_state_subscription is not None
ufp.events_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert hass.states.get(ring_id).state == STATE_UNAVAILABLE
assert hass.states.get(motion_id).state == STATE_UNAVAILABLE
ufp.events_ws_state_subscription(WebsocketState.CONNECTED)
await hass.async_block_till_done()
assert hass.states.get(ring_id).state != STATE_UNAVAILABLE
assert hass.states.get(motion_id).state != STATE_UNAVAILABLE
tests/components/unifiprotect/test_recorder.py
+32 -35
View File
@@ -1,22 +1,26 @@
"""The tests for unifiprotect recorder."""
from datetime import datetime, timedelta
from unittest.mock import Mock
from uiprotect.data import Camera, Event, EventType, ModelType
from uiprotect import EventChange, ProtectEvent, ProtectEventChannel
from uiprotect.data import Camera, EventType, SmartDetectObjectType
from homeassistant.components.recorder import Recorder
from homeassistant.components.recorder.history import get_significant_states
from homeassistant.components.unifiprotect.binary_sensor import EVENT_SENSORS
from homeassistant.components.unifiprotect.const import (
ATTR_EVENT_ID,
ATTR_EVENT_SCORE,
DEFAULT_ATTRIBUTION,
ATTR_SMART_DETECT_TYPES,
)
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_FRIENDLY_NAME, STATE_ON, Platform
from homeassistant.components.unifiprotect.event import EVENT_DESCRIPTIONS
from homeassistant.const import ATTR_FRIENDLY_NAME, Platform
from homeassistant.core import HomeAssistant
from .utils import MockUFPFixture, ids_from_device_description, init_entry
from .utils import (
MockUFPFixture,
ids_from_device_description,
init_entry,
setup_public_camera,
)
from tests.components.recorder.common import async_wait_recording_done
@@ -29,44 +33,37 @@ async def test_exclude_attributes(
unadopted_camera: Camera,
fixed_now: datetime,
) -> None:
"""Test binary_sensor has event_id and event_score excluded from recording."""
"""The smart-detect event entity excludes event_id/smart_detect_types from recording."""
now = fixed_now
# Smart-detect events arrive on the public events websocket; the entity's
# availability also requires the public camera to resolve.
setup_public_camera(ufp)
await init_entry(hass, ufp, [doorbell, unadopted_camera])
description = next(d for d in EVENT_DESCRIPTIONS if d.key == "smart_detection")
_, entity_id = await ids_from_device_description(
hass, Platform.BINARY_SENSOR, doorbell, EVENT_SENSORS[1]
hass, Platform.EVENT, doorbell, description
)
event = Event(
model=ModelType.EVENT,
id="test_event_id",
type=EventType.MOTION,
start=fixed_now - timedelta(seconds=1),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
ufp.events_msg(
ProtectEvent(
id="test_event_id",
type=EventType.SMART_DETECT,
channel=ProtectEventChannel.DETECTION,
device_id=doorbell.id,
device_mac=doorbell.mac,
start=fixed_now - timedelta(seconds=1),
end=fixed_now,
smart_detect_types=(SmartDetectObjectType.PERSON,),
),
EventChange.STARTED,
)
new_camera = doorbell.model_copy()
new_camera.is_motion_detected = True
new_camera.last_motion_event_id = event.id
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_camera
ufp.api.bootstrap.cameras = {new_camera.id: new_camera}
ufp.api.bootstrap.events = {event.id: event}
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_ON
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
assert state.attributes[ATTR_EVENT_SCORE] == 100
assert state.attributes[ATTR_EVENT_ID] == "test_event_id"
assert ATTR_SMART_DETECT_TYPES in state.attributes
await async_wait_recording_done(hass)
states = await hass.async_add_executor_job(
@@ -75,6 +72,6 @@ async def test_exclude_attributes(
assert len(states) >= 1
for entity_states in states.values():
for state in entity_states:
assert ATTR_EVENT_SCORE not in state.attributes
assert ATTR_EVENT_ID not in state.attributes
assert ATTR_SMART_DETECT_TYPES not in state.attributes
assert ATTR_FRIENDLY_NAME in state.attributes
tests/components/unifiprotect/test_relay.py
+21
View File
@@ -455,6 +455,27 @@ async def test_public_ws_state_change_without_public_bootstrap(
assert data.last_public_update_success is False
async def test_events_ws_state_change_without_public_bootstrap(
hass: HomeAssistant,
ufp: MockUFPFixture,
) -> None:
"""Events WS state changes flip the flag but no-op without a bootstrap."""
await init_entry(hass, ufp, [])
data = ufp.entry.runtime_data
assert data.last_events_update_success is True
assert ufp.events_ws_state_subscription is not None
# No public bootstrap -> re-signal step returns early.
ufp.events_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert data.last_events_update_success is False
# Same state again -> handler early-returns.
ufp.events_ws_state_subscription(WebsocketState.DISCONNECTED)
await hass.async_block_till_done()
assert data.last_events_update_success is False
async def test_relay_public_ws_message_without_public_old_obj(
hass: HomeAssistant,
ufp_with_relay: tuple[MockUFPFixture, Mock],
tests/components/unifiprotect/utils.py
+67 -3
View File
@@ -23,6 +23,8 @@ from uiprotect.data import (
ProtectModelWithId,
PublicBootstrap,
Sensor,
SmartDetectAudioType,
SmartDetectObjectType,
WSSubscriptionMessage,
)
from uiprotect.data.bootstrap import ProtectDeviceRef
@@ -35,6 +37,7 @@ from uiprotect.data.public_devices import (
PublicSensor,
PublicSensorLeakSettings,
PublicSensorMotionSettingsRead,
PublicSmartDetectSettings,
PublicWirelessBatteryStatus,
PublicWirelessConnectionState,
SensorFeatureCapability,
@@ -62,6 +65,7 @@ class MockUFPFixture:
devices_ws_subscription: Callable[[WSSubscriptionMessage], None] | None = None
events_subscription: Callable[[ProtectEvent, EventChange], None] | None = None
devices_ws_state_subscription: Callable[[WebsocketState], None] | None = None
events_ws_state_subscription: Callable[[WebsocketState], None] | None = None
def ws_msg(self, msg: WSSubscriptionMessage) -> None:
"""Emit WS message for testing."""
@@ -408,18 +412,41 @@ _HDR_DISPLAY_TO_PUBLIC = {
}
_ALL_OBJECT_TYPES = [t for t in SmartDetectObjectType if t.audio_type is None]
_ALL_AUDIO_TYPES = list(SmartDetectAudioType)
def make_public_camera(
camera: Camera,
*,
state: DeviceState | None = None,
is_motion_detected: bool = False,
is_smart_currently_detected: bool = False,
is_person_currently_detected: bool = False,
is_vehicle_currently_detected: bool = False,
is_animal_currently_detected: bool = False,
is_audio_currently_detected: bool = False,
is_smoke_currently_detected: bool = False,
is_cmonx_currently_detected: bool = False,
is_siren_currently_detected: bool = False,
is_baby_cry_currently_detected: bool = False,
is_speaking_currently_detected: bool = False,
is_bark_currently_detected: bool = False,
is_car_alarm_currently_detected: bool = False,
is_car_horn_currently_detected: bool = False,
is_glass_break_currently_detected: bool = False,
object_types: list[SmartDetectObjectType] | None = None,
audio_types: list[SmartDetectAudioType] | None = None,
mic_volume: int | None = None,
hdr_type: PublicHdrMode | None = None,
) -> Mock:
"""Build a public-API camera mirroring a private camera's migrated fields.
``mic_volume`` and ``hdr_type`` default to values derived from the private
fixture so the public mirror matches it; pass an override to assert a value
the private object would not produce.
The stream tiers/mic/HDR back the migrated stream and select entities; the
``is_*`` flags back the migrated ``ufp_public_value`` detection paths and the
``smart_detect_settings`` types back the per-type ``ufp_public_enabled_fn``
gates (default: all types enabled). ``mic_volume`` and ``hdr_type`` default to
values derived from the private fixture so the public mirror matches it.
"""
public = Mock(spec=PublicCamera)
public.id = camera.id
@@ -430,6 +457,43 @@ def make_public_camera(
public.model = ModelType.CAMERA
public.state = DeviceState[camera.state.name] if state is None else state
public.mic_volume = camera.mic_volume if mic_volume is None else mic_volume
public.is_motion_detected = is_motion_detected
public.is_smart_currently_detected = is_smart_currently_detected
public.is_person_currently_detected = is_person_currently_detected
public.is_vehicle_currently_detected = is_vehicle_currently_detected
public.is_animal_currently_detected = is_animal_currently_detected
public.is_audio_currently_detected = is_audio_currently_detected
public.is_smoke_currently_detected = is_smoke_currently_detected
public.is_cmonx_currently_detected = is_cmonx_currently_detected
public.is_siren_currently_detected = is_siren_currently_detected
public.is_baby_cry_currently_detected = is_baby_cry_currently_detected
public.is_speaking_currently_detected = is_speaking_currently_detected
public.is_bark_currently_detected = is_bark_currently_detected
public.is_car_alarm_currently_detected = is_car_alarm_currently_detected
public.is_car_horn_currently_detected = is_car_horn_currently_detected
public.is_glass_break_currently_detected = is_glass_break_currently_detected
public.smart_detect_settings = PublicSmartDetectSettings(
object_types=_ALL_OBJECT_TYPES if object_types is None else object_types,
audio_types=_ALL_AUDIO_TYPES if audio_types is None else audio_types,
)
# A Mock(spec) does not evaluate properties, so mirror the PublicCamera
# parity properties the migrated detection sensors gate on using the
# library's own logic.
for name in (
"is_person_detection_on",
"is_vehicle_detection_on",
"is_animal_detection_on",
"is_smoke_detection_on",
"is_co_detection_on",
"is_siren_detection_on",
"is_baby_cry_detection_on",
"is_speaking_detection_on",
"is_bark_detection_on",
"is_car_alarm_detection_on",
"is_car_horn_detection_on",
"is_glass_break_detection_on",
):
setattr(public, name, getattr(PublicCamera, name).fget(public))
public.hdr_type = (
_HDR_DISPLAY_TO_PUBLIC[camera.hdr_mode_display]
if hdr_type is None