diff --git a/homeassistant/components/unifiprotect/const.py b/homeassistant/components/unifiprotect/const.py index f7138c24341..c8d438a53d5 100644 --- a/homeassistant/components/unifiprotect/const.py +++ b/homeassistant/components/unifiprotect/const.py @@ -83,6 +83,10 @@ EVENT_TYPE_FINGERPRINT_IDENTIFIED: Final = "identified" EVENT_TYPE_FINGERPRINT_NOT_IDENTIFIED: Final = "not_identified" EVENT_TYPE_NFC_SCANNED: Final = "scanned" EVENT_TYPE_DOORBELL_RING: Final = "ring" +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" diff --git a/homeassistant/components/unifiprotect/event.py b/homeassistant/components/unifiprotect/event.py index cb9090dd530..b87ed530817 100644 --- a/homeassistant/components/unifiprotect/event.py +++ b/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 @@ from .const import ( 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 .data import ( 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 @@ class ProtectDeviceFingerprintEventEntity( 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 @@ EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = ( ], 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, + ), ) diff --git a/homeassistant/components/unifiprotect/strings.json b/homeassistant/components/unifiprotect/strings.json index 66193d096ba..289e3b770da 100644 --- a/homeassistant/components/unifiprotect/strings.json +++ b/homeassistant/components/unifiprotect/strings.json @@ -253,6 +253,16 @@ } } } + }, + "vehicle": { + "name": "Vehicle", + "state_attributes": { + "event_type": { + "state": { + "detected": "Detected" + } + } + } } }, "media_player": { diff --git a/tests/components/unifiprotect/test_event.py b/tests/components/unifiprotect/test_event.py index 80b11c047cc..03f989b90c3 100644 --- a/tests/components/unifiprotect/test_event.py +++ b/tests/components/unifiprotect/test_event.py @@ -2,9 +2,11 @@ from __future__ import annotations +import asyncio from datetime import datetime, timedelta -from unittest.mock import Mock +from unittest.mock import Mock, patch +import pytest from uiprotect.data import Camera, Event, EventType, ModelType, SmartDetectObjectType from homeassistant.components.unifiprotect.const import ( @@ -25,6 +27,19 @@ from .utils import ( remove_entities, ) +# Short delay for testing +TEST_VEHICLE_EVENT_DELAY = 0.05 + + +@pytest.fixture(autouse=True) +def short_vehicle_delay(): + """Use a short delay for vehicle event tests.""" + with patch( + "homeassistant.components.unifiprotect.event.VEHICLE_EVENT_DELAY_SECONDS", + TEST_VEHICLE_EVENT_DELAY, + ): + yield + async def test_camera_remove( hass: HomeAssistant, ufp: MockUFPFixture, doorbell: Camera, unadopted_camera: Camera @@ -33,11 +48,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) 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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) async def test_doorbell_ring( @@ -50,7 +65,7 @@ async def test_doorbell_ring( """Test a doorbell ring event.""" await init_entry(hass, ufp, [doorbell, unadopted_camera]) - assert_entity_counts(hass, Platform.EVENT, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -164,7 +179,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -239,7 +254,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -315,7 +330,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -383,7 +398,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -444,7 +459,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -512,7 +527,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -581,7 +596,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -642,7 +657,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, 3, 3) + assert_entity_counts(hass, Platform.EVENT, 4, 4) events: list[HAEvent] = [] @callback @@ -688,3 +703,695 @@ async def test_doorbell_fingerprint_not_identified( assert state.attributes["ulp_id"] == "" unsub() + + +async def test_vehicle_detection_basic( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test basic vehicle detection event with thumbnails.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 4, 4) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + + # Create event with vehicle thumbnail + event = Event( + model=ModelType.EVENT, + id="test_vehicle_event_id", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 95, + "clock_best_wall": fixed_now, + "cropped_id": "test_thumb_id", + } + ] + }, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_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) + + # Wait for the timer + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Should have received vehicle detection event + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + assert state.attributes[ATTR_EVENT_ID] == "test_vehicle_event_id" + assert state.attributes["confidence"] == 95 + assert "clock_best_wall" in state.attributes + assert "license_plate" not in state.attributes + + unsub() + + +async def test_vehicle_detection_with_lpr_ufp6( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """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, 4, 4) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + + # Create event with vehicle thumbnail and LPR in group.matched_name (UFP 6.0+) + event = Event( + model=ModelType.EVENT, + id="test_lpr_event_id", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 98, + "clock_best_wall": fixed_now, + "cropped_id": "test_thumb_id", + "group": { + "id": "lpr_group_1", + "matched_name": "ABC123", + "confidence": 95, + }, + "attributes": { + "color": {"val": "blue", "confidence": 90}, + "vehicle_type": {"val": "sedan", "confidence": 85}, + }, + } + ] + }, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_lpr_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) + + # Wait for the timer + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Should have received vehicle detection event + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_EVENT_ID] == "test_lpr_event_id" + assert state.attributes["confidence"] == 98 + assert state.attributes["license_plate"] == "ABC123" + assert "attributes" in state.attributes + assert state.attributes["attributes"]["color"]["val"] == "blue" + assert state.attributes["attributes"]["vehicleType"]["val"] == "sedan" + + unsub() + + +async def test_vehicle_detection_with_lpr_legacy( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test vehicle detection with license plate recognition (legacy format).""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 4, 4) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + + # Create event with vehicle thumbnail and LPR in name field (legacy) + event = Event( + model=ModelType.EVENT, + id="test_lpr_legacy_event_id", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 92, + "clock_best_wall": fixed_now, + "name": "XYZ789", + "cropped_id": "test_thumb_id", + } + ] + }, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_lpr_legacy_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) + + # Wait for the timer + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Should have received vehicle detection event + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_EVENT_ID] == "test_lpr_legacy_event_id" + assert state.attributes["confidence"] == 92 + assert state.attributes["license_plate"] == "XYZ789" + + unsub() + + +async def test_vehicle_detection_multiple_thumbnails( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test vehicle detection with multiple thumbnails - should pick best LPR.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 4, 4) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + + # Create event with multiple vehicle thumbnails - best should be highest confidence LPR + event = Event( + model=ModelType.EVENT, + id="test_multi_thumbnail_id", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 70, + "clock_best_wall": fixed_now - timedelta(seconds=2), + "cropped_id": "test_thumb_id", + }, + { + "type": "vehicle", + "confidence": 85, + "clock_best_wall": fixed_now - timedelta(seconds=1), + "cropped_id": "test_thumb_id_2", + "group": { + "id": "lpr_group_2", + "matched_name": "LOW_CONF", + "confidence": 85, + }, + }, + { + "type": "vehicle", + "confidence": 99, + "clock_best_wall": fixed_now, + "cropped_id": "test_thumb_id_3", + "group": { + "id": "lpr_group_3", + "matched_name": "BEST_LPR", + "confidence": 99, + }, + }, + { + "type": "person", # Should be ignored + "confidence": 100, + "clock_best_wall": fixed_now, + "cropped_id": "test_thumb_id_person", + }, + ] + }, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_multi_thumbnail_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) + + # Wait for the timer + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Should have received vehicle detection event with highest confidence LPR + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_EVENT_ID] == "test_multi_thumbnail_id" + assert state.attributes["confidence"] == 99 + assert state.attributes["license_plate"] == "BEST_LPR" + + unsub() + + +async def test_vehicle_detection_no_thumbnails( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test vehicle detection event without thumbnails - should not fire.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 4, 4) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + + # Create event without detected_thumbnails + event = Event( + model=ModelType.EVENT, + id="test_no_thumbnails_id", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={}, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_no_thumbnails_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) + + # Wait for the timer to expire + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Should NOT have received any events (no vehicle thumbnails) + assert len(events) == 0 + + unsub() + + +async def test_vehicle_detection_timer_reset_on_new_thumbnail( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """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, 4, 4) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + + # Create event with one thumbnail + event = Event( + model=ModelType.EVENT, + id="test_timer_reset_id", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 80, + "clock_best_wall": fixed_now, + "cropped_id": "test_thumb_id_1", + } + ] + }, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_timer_reset_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() + + # Wait briefly (timer hasn't expired yet) + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY / 2) + await hass.async_block_till_done() + + # No event yet (timer hasn't expired) + assert len(events) == 0 + + # Update with better thumbnail - should reset timer + event.metadata = { + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 80, + "clock_best_wall": fixed_now, + "cropped_id": "test_thumb_id_1", + }, + { + "type": "vehicle", + "confidence": 95, + "clock_best_wall": fixed_now + timedelta(seconds=1), + "cropped_id": "test_thumb_id_2", + "group": { + "id": "lpr_group_4", + "matched_name": "BETTER_LPR", + "confidence": 95, + }, + }, + ] + } + + ufp.api.bootstrap.events = {event.id: event} + mock_msg.new_obj = event + ufp.ws_msg(mock_msg) + await hass.async_block_till_done() + + # Still no event (timer extended) + assert len(events) == 0 + + # Wait for timer to expire + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Now should have the event with the better LPR + assert len(events) == 1 + state = events[0].data["new_state"] + assert state + assert state.attributes["confidence"] == 95 + assert state.attributes["license_plate"] == "BETTER_LPR" + + unsub() + + +async def test_vehicle_detection_new_event_cancels_timer( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test that new event cancels timer for previous event.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 4, 4) + events: list[HAEvent] = [] + + @callback + def _capture_event(event: HAEvent) -> None: + events.append(event) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + unsub = async_track_state_change_event(hass, entity_id, _capture_event) + + # Create first event + event1 = Event( + model=ModelType.EVENT, + id="test_event_1", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=5), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 80, + "clock_best_wall": fixed_now - timedelta(seconds=4), + "cropped_id": "test_thumb_id", + "group": { + "id": "lpr_group_5", + "matched_name": "FIRST", + "confidence": 80, + }, + } + ] + }, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_event_1" + ufp.api.bootstrap.cameras = {new_camera.id: new_camera} + ufp.api.bootstrap.events = {event1.id: event1} + + mock_msg = Mock() + mock_msg.changed_data = {} + mock_msg.new_obj = event1 + ufp.ws_msg(mock_msg) + await hass.async_block_till_done() + + # Wait briefly (timer hasn't expired yet) + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY / 2) + await hass.async_block_till_done() + + # No event yet + assert len(events) == 0 + + # Send new event - should fire first event immediately + event2 = Event( + model=ModelType.EVENT, + id="test_event_2", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 95, + "clock_best_wall": fixed_now, + "cropped_id": "test_thumb_id", + "group": { + "id": "lpr_group_6", + "matched_name": "SECOND", + "confidence": 95, + }, + } + ] + }, + ) + + new_camera.last_smart_detect_event_id = "test_event_2" + ufp.api.bootstrap.cameras = {new_camera.id: new_camera} + ufp.api.bootstrap.events = {event2.id: event2} + + mock_msg.new_obj = event2 + ufp.ws_msg(mock_msg) + await hass.async_block_till_done() + + # Wait for second event's timer + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Should have two events - first fired immediately when second arrived + assert len(events) == 2 + # First event fired immediately when second event arrived + state = events[0].data["new_state"] + assert state + assert state.attributes[ATTR_EVENT_ID] == "test_event_1" + assert state.attributes["license_plate"] == "FIRST" + # Second event fired after timer + state = events[1].data["new_state"] + assert state + assert state.attributes[ATTR_EVENT_ID] == "test_event_2" + assert state.attributes["license_plate"] == "SECOND" + + unsub() + + +async def test_vehicle_detection_timer_cleanup_on_remove( + hass: HomeAssistant, + ufp: MockUFPFixture, + doorbell: Camera, + unadopted_camera: Camera, + fixed_now: datetime, +) -> None: + """Test that pending timer is cancelled when entity is removed.""" + + await init_entry(hass, ufp, [doorbell, unadopted_camera]) + assert_entity_counts(hass, Platform.EVENT, 4, 4) + + _, entity_id = await ids_from_device_description( + hass, Platform.EVENT, doorbell, EVENT_DESCRIPTIONS[3] + ) + + # Create event with vehicle thumbnail + event = Event( + model=ModelType.EVENT, + id="test_cleanup_event_id", + type=EventType.SMART_DETECT, + start=fixed_now - timedelta(seconds=1), + end=None, + score=100, + smart_detect_types=[], + smart_detect_event_ids=[], + camera_id=doorbell.id, + api=ufp.api, + metadata={ + "detected_thumbnails": [ + { + "type": "vehicle", + "confidence": 90, + "clock_best_wall": fixed_now, + "cropped_id": "test_cleanup_thumb_id", + } + ] + }, + ) + + new_camera = doorbell.model_copy() + new_camera.last_smart_detect_event_id = "test_cleanup_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() + + # Timer is now pending - remove the entity before it fires + await remove_entities(hass, ufp, [doorbell]) + await hass.async_block_till_done() + + # Wait past when timer would have fired + await asyncio.sleep(TEST_VEHICLE_EVENT_DELAY * 2) + await hass.async_block_till_done() + + # Entity should be gone and no event should have fired + state = hass.states.get(entity_id) + assert state is None