UnifiProtect add vehicle detection event entity with license plate recognition support (home-assistant#157203)

Co-authored-by: J. Nick Koston <[email protected]>

Author: RaHehl

homeassistant/components/unifiprotect/const.py

@@ -83,6 +83,10 @@
 EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED: Final = "not_identified"
 EVENT_TYPE_NFC_SCANNED: Final = "scanned"
 EVENT_TYPE_DOORBELL_RING: Final = "ring"
+EVENT_TYPE_VEHICLE_DETECTED: Final = "detected"
+
+# Delay in seconds before firing vehicle event after last thumbnail
+VEHICLE_EVENT_DELAY_SECONDS: Final = 3
 
 KEYRINGS_ULP_ID: Final = "ulp_id"
 KEYRINGS_USER_STATUS: Final = "user_status"

homeassistant/components/unifiprotect/event.py

@@ -3,14 +3,18 @@
 from __future__ import annotations
 
 import dataclasses
+from typing import Any
+
+from uiprotect.data.nvr import Event, EventDetectedThumbnail
 
 from homeassistant.components.event import (
     EventDeviceClass,
     EventEntity,
     EventEntityDescription,
 )
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.event import async_call_at
 
 from . import Bootstrap
 from .const import (
@@ -19,10 +23,12 @@
     EVENT_TYPE_FINGERPRINT_IDENTIFIED,
     EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED,
     EVENT_TYPE_NFC_SCANNED,
+    EVENT_TYPE_VEHICLE_DETECTED,
     KEYRINGS_KEY_TYPE_ID_NFC,
     KEYRINGS_ULP_ID,
     KEYRINGS_USER_FULL_NAME,
     KEYRINGS_USER_STATUS,
+    VEHICLE_EVENT_DELAY_SECONDS,
 )
 from .data import (
     Camera,
@@ -35,6 +41,17 @@
 from .entity import EventEntityMixin, ProtectDeviceEntity, ProtectEventMixin
 
 
+# Select best thumbnail
+# Prefer thumbnails with LPR data, sorted by confidence
+# LPR can be in: 1) group.matched_name (UFP 6.0+) or 2) name field
+def _thumbnail_sort_key(t: EventDetectedThumbnail) -> tuple[bool, float, float]:
+    """Sort key: (has_lpr, confidence, clock_best_wall)."""
+    has_lpr = bool((t.group and t.group.matched_name) or (t.name and len(t.name) > 0))
+    confidence = t.confidence if t.confidence else 0.0
+    clock = t.clock_best_wall.timestamp() if t.clock_best_wall else 0.0
+    return (has_lpr, confidence, clock)
+
+
 def _add_ulp_user_infos(
     bootstrap: Bootstrap, event_data: dict[str, str], ulp_id: str
 ) -> None:
@@ -167,6 +184,152 @@ def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
             self.async_write_ha_state()
 
 
+class ProtectDeviceVehicleEventEntity(
+    EventEntityMixin, ProtectDeviceEntity, EventEntity
+):
+    """A UniFi Protect vehicle detection event entity.
+
+    Vehicle detection events use a delayed firing mechanism to allow time for
+    the best thumbnail (with license plate recognition data) to arrive. The
+    timer is extended each time new thumbnails arrive for the same event. If
+    a new event arrives while a timer is pending, the old event fires immediately
+    with its stored thumbnails, then a new timer starts for the new event.
+    """
+
+    entity_description: ProtectEventEntityDescription
+    _thumbnail_timer_cancel: CALLBACK_TYPE | None = None
+    _latest_event_id: str | None = None
+    _latest_thumbnails: list[EventDetectedThumbnail] | None = None
+    _thumbnail_timer_due: float = 0.0  # Loop time when timer should fire
+
+    async def async_added_to_hass(self) -> None:
+        """Register cleanup callback when entity is added."""
+        await super().async_added_to_hass()
+        self.async_on_remove(self._cancel_thumbnail_timer)
+
+    @callback
+    def _cancel_thumbnail_timer(self) -> None:
+        """Cancel pending thumbnail timer if one exists."""
+        if self._thumbnail_timer_cancel:
+            self._thumbnail_timer_cancel()
+            self._thumbnail_timer_cancel = None
+
+    @callback
+    def _async_timer_callback(self, *_: Any) -> None:
+        """Handle timer expiration - fire the vehicle event.
+
+        If the due time was extended (new thumbnails arrived), re-arm the timer.
+        Otherwise, fire the event with the stored thumbnails.
+        """
+        self._thumbnail_timer_cancel = None
+        if self._thumbnail_timer_due > self.hass.loop.time():
+            # Timer fired early because due time was extended; re-arm
+            self._async_set_thumbnail_timer()
+            return
+
+        if self._latest_event_id:
+            self._fire_vehicle_event(self._latest_event_id, self._latest_thumbnails)
+
+    @staticmethod
+    def _get_vehicle_thumbnails(event: Event) -> list[EventDetectedThumbnail]:
+        """Get vehicle thumbnails from event."""
+        if event.metadata and event.metadata.detected_thumbnails:
+            return [
+                t for t in event.metadata.detected_thumbnails if t.type == "vehicle"
+            ]
+        return []
+
+    @callback
+    def _fire_vehicle_event(
+        self, event_id: str, thumbnails: list[EventDetectedThumbnail] | None = None
+    ) -> None:
+        """Fire the vehicle detection event with best available thumbnail.
+
+        Args:
+            event_id: The event ID to include in the fired event data.
+            thumbnails: Pre-stored thumbnails to use. If None, fetches from
+                the current event (used when event is still active).
+        """
+        if thumbnails is None:
+            # No stored thumbnails; try to get from current event
+            event = self.entity_description.get_event_obj(self.device)
+            if not event or event.id != event_id:
+                return
+            thumbnails = self._get_vehicle_thumbnails(event)
+
+        if not thumbnails:
+            return
+
+        # Start with just the event ID
+        event_data: dict[str, Any] = {
+            ATTR_EVENT_ID: event_id,
+            "thumbnail_count": len(thumbnails),
+        }
+
+        thumbnail = max(thumbnails, key=_thumbnail_sort_key)
+
+        # Add confidence if available
+        if thumbnail.confidence is not None:
+            event_data["confidence"] = thumbnail.confidence
+
+        # Add best detection frame timestamp
+        if thumbnail.clock_best_wall is not None:
+            event_data["clock_best_wall"] = thumbnail.clock_best_wall.isoformat()
+
+        # License plate from group.matched_name (UFP 6.0+) or name field (older)
+        if thumbnail.group and thumbnail.group.matched_name:
+            event_data["license_plate"] = thumbnail.group.matched_name
+        elif thumbnail.name:
+            event_data["license_plate"] = thumbnail.name
+
+        # Add all thumbnail attributes as dict
+        if thumbnail.attributes:
+            event_data["attributes"] = thumbnail.attributes.unifi_dict()
+
+        self._trigger_event(EVENT_TYPE_VEHICLE_DETECTED, event_data)
+        self.async_write_ha_state()
+
+    @callback
+    def _async_set_thumbnail_timer(self) -> None:
+        """Schedule the thumbnail timer to fire at _thumbnail_timer_due."""
+        self._thumbnail_timer_cancel = async_call_at(
+            self.hass,
+            self._async_timer_callback,
+            self._thumbnail_timer_due,
+        )
+
+    @callback
+    def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
+        description = self.entity_description
+
+        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
+
+        # Process vehicle detection events with thumbnails
+        if (
+            event
+            and event.type is EventType.SMART_DETECT
+            and (thumbnails := self._get_vehicle_thumbnails(event))
+        ):
+            # New event arrived while timer pending for different event?
+            # Fire the old event immediately since it has completed
+            if self._latest_event_id and self._latest_event_id != event.id:
+                self._fire_vehicle_event(self._latest_event_id, self._latest_thumbnails)
+
+            # Store event data and extend/start the timer
+            # Timer extension allows better thumbnails (with LPR) to arrive
+            self._latest_event_id = event.id
+            self._latest_thumbnails = thumbnails
+            self._thumbnail_timer_due = (
+                self.hass.loop.time() + VEHICLE_EVENT_DELAY_SECONDS
+            )
+            # Only schedule if no timer running; existing timer will re-arm
+            if self._thumbnail_timer_cancel is None:
+                self._async_set_thumbnail_timer()
+
+
 EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = (
     ProtectEventEntityDescription(
         key="doorbell",
@@ -199,6 +362,15 @@ def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
         ],
         entity_class=ProtectDeviceFingerprintEventEntity,
     ),
+    ProtectEventEntityDescription(
+        key="vehicle",
+        translation_key="vehicle",
+        icon="mdi:car",
+        ufp_required_field="feature_flags.has_smart_detect",
+        ufp_event_obj="last_smart_detect_event",
+        event_types=[EVENT_TYPE_VEHICLE_DETECTED],
+        entity_class=ProtectDeviceVehicleEventEntity,
+    ),
 )
 
 

homeassistant/components/unifiprotect/strings.json

@@ -253,6 +253,16 @@
             }
           }
         }
+      },
+      "vehicle": {
+        "name": "Vehicle",
+        "state_attributes": {
+          "event_type": {
+            "state": {
+              "detected": "Detected"
+            }
+          }
+        }
       }
     },
     "media_player": {