mirror of
https://github.com/home-assistant/core.git
synced 2026-06-06 23:46:56 +01:00
Remove license plate event sensor (#157196)
This commit is contained in:
@@ -17,7 +17,6 @@ from uiprotect.data import (
|
||||
ProtectAdoptableDeviceModel,
|
||||
ProtectDeviceModel,
|
||||
Sensor,
|
||||
SmartDetectObjectType,
|
||||
)
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
@@ -533,18 +532,6 @@ NVR_DISABLED_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
|
||||
),
|
||||
)
|
||||
|
||||
LICENSE_PLATE_EVENT_SENSORS: tuple[ProtectSensorEventEntityDescription, ...] = (
|
||||
ProtectSensorEventEntityDescription(
|
||||
key="smart_obj_licenseplate",
|
||||
icon="mdi:car",
|
||||
translation_key="license_plate_detected",
|
||||
ufp_obj_type=SmartDetectObjectType.LICENSE_PLATE,
|
||||
ufp_required_field="can_detect_license_plate",
|
||||
ufp_event_obj="last_license_plate_detect_event",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
LIGHT_SENSORS: tuple[ProtectSensorEntityDescription, ...] = (
|
||||
ProtectSensorEntityDescription(
|
||||
key="motion_last_trip_time",
|
||||
@@ -680,20 +667,6 @@ def _async_event_entities(
|
||||
camera.display_name,
|
||||
)
|
||||
|
||||
if not camera.feature_flags.has_smart_detect:
|
||||
continue
|
||||
|
||||
for event_desc in LICENSE_PLATE_EVENT_SENSORS:
|
||||
if not event_desc.has_required(camera):
|
||||
continue
|
||||
|
||||
entities.append(ProtectLicensePlateEventSensor(data, camera, event_desc))
|
||||
_LOGGER.debug(
|
||||
"Adding sensor entity %s for %s",
|
||||
description.name,
|
||||
camera.display_name,
|
||||
)
|
||||
|
||||
return entities
|
||||
|
||||
|
||||
@@ -738,40 +711,3 @@ class ProtectEventSensor(EventEntityMixin, SensorEntity):
|
||||
"_attr_native_value",
|
||||
"_attr_extra_state_attributes",
|
||||
)
|
||||
|
||||
|
||||
class ProtectLicensePlateEventSensor(ProtectEventSensor):
|
||||
"""A UniFi Protect license plate sensor."""
|
||||
|
||||
device: Camera
|
||||
|
||||
@callback
|
||||
def _set_event_done(self) -> None:
|
||||
self._attr_native_value = OBJECT_TYPE_NONE
|
||||
self._attr_extra_state_attributes = {}
|
||||
|
||||
@callback
|
||||
def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
|
||||
description = self.entity_description
|
||||
|
||||
prev_event = self._event
|
||||
prev_event_end = self._event_end
|
||||
super()._async_update_device_from_protect(device)
|
||||
if event := description.get_event_obj(device):
|
||||
self._event = event
|
||||
self._event_end = event.end
|
||||
|
||||
if not (
|
||||
event
|
||||
and (metadata := event.metadata)
|
||||
and (license_plate := metadata.license_plate)
|
||||
and description.has_matching_smart(event)
|
||||
and not self._event_already_ended(prev_event, prev_event_end)
|
||||
):
|
||||
self._set_event_done()
|
||||
return
|
||||
|
||||
self._attr_native_value = license_plate.name
|
||||
self._set_event_attrs(event)
|
||||
if event.end:
|
||||
self._async_event_with_immediate_end()
|
||||
|
||||
@@ -730,109 +730,6 @@ async def test_browse_media_recent_truncated(
|
||||
),
|
||||
"Object Detection - Person, Vehicle",
|
||||
),
|
||||
(
|
||||
Event(
|
||||
model=ModelType.EVENT,
|
||||
id="test_event_id",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=datetime(2000, 1, 1, 0, 0, 0),
|
||||
end=None,
|
||||
score=100,
|
||||
smart_detect_types=["vehicle", "licensePlate"],
|
||||
smart_detect_event_ids=[],
|
||||
camera_id="test",
|
||||
),
|
||||
"Object Detection - License Plate, Vehicle",
|
||||
),
|
||||
(
|
||||
Event(
|
||||
model=ModelType.EVENT,
|
||||
id="test_event_id",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=datetime(2000, 1, 1, 0, 0, 0),
|
||||
end=None,
|
||||
score=100,
|
||||
smart_detect_types=["vehicle", "licensePlate"],
|
||||
smart_detect_event_ids=[],
|
||||
camera_id="test",
|
||||
metadata={
|
||||
"license_plate": {"name": "ABC1234", "confidence_level": 95},
|
||||
"detected_thumbnails": [
|
||||
{
|
||||
"clock_best_wall": datetime(2000, 1, 1, 0, 0, 0),
|
||||
"type": "vehicle",
|
||||
"cropped_id": "event_id",
|
||||
}
|
||||
],
|
||||
},
|
||||
),
|
||||
"Object Detection - Vehicle: ABC1234",
|
||||
),
|
||||
(
|
||||
Event(
|
||||
model=ModelType.EVENT,
|
||||
id="test_event_id",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=datetime(2000, 1, 1, 0, 0, 0),
|
||||
end=None,
|
||||
score=100,
|
||||
smart_detect_types=["vehicle", "licensePlate"],
|
||||
smart_detect_event_ids=[],
|
||||
camera_id="test",
|
||||
metadata={
|
||||
"license_plate": {"name": "ABC1234", "confidence_level": 95},
|
||||
"detected_thumbnails": [
|
||||
{
|
||||
"clock_best_wall": datetime(2000, 1, 1, 0, 0, 0),
|
||||
"type": "vehicle",
|
||||
"cropped_id": "event_id",
|
||||
"attributes": {
|
||||
"vehicle_type": {
|
||||
"val": "car",
|
||||
"confidence": 95,
|
||||
}
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
),
|
||||
"Object Detection - Car: ABC1234",
|
||||
),
|
||||
(
|
||||
Event(
|
||||
model=ModelType.EVENT,
|
||||
id="test_event_id",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=datetime(2000, 1, 1, 0, 0, 0),
|
||||
end=None,
|
||||
score=100,
|
||||
smart_detect_types=["vehicle", "licensePlate"],
|
||||
smart_detect_event_ids=[],
|
||||
camera_id="test",
|
||||
metadata={
|
||||
"license_plate": {"name": "ABC1234", "confidence_level": 95},
|
||||
"detected_thumbnails": [
|
||||
{
|
||||
"clock_best_wall": datetime(2000, 1, 1, 0, 0, 0),
|
||||
"type": "vehicle",
|
||||
"cropped_id": "event_id",
|
||||
"attributes": {
|
||||
"color": {
|
||||
"val": "black",
|
||||
"confidence": 95,
|
||||
}
|
||||
},
|
||||
},
|
||||
{
|
||||
"clock_best_wall": datetime(2000, 1, 1, 0, 0, 0),
|
||||
"type": "person",
|
||||
"cropped_id": "event_id",
|
||||
},
|
||||
],
|
||||
},
|
||||
),
|
||||
"Object Detection - Black Vehicle: ABC1234",
|
||||
),
|
||||
(
|
||||
Event(
|
||||
model=ModelType.EVENT,
|
||||
|
||||
@@ -6,26 +6,14 @@ from datetime import datetime, timedelta
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
from uiprotect.data import (
|
||||
NVR,
|
||||
Camera,
|
||||
Event,
|
||||
EventType,
|
||||
ModelType,
|
||||
Sensor,
|
||||
SmartDetectObjectType,
|
||||
)
|
||||
from uiprotect.data.nvr import EventMetadata, LicensePlateMetadata
|
||||
from uiprotect.data import NVR, Camera, Event, EventType, ModelType, Sensor
|
||||
from uiprotect.data.nvr import EventMetadata
|
||||
|
||||
from homeassistant.components.unifiprotect.const import (
|
||||
ATTR_EVENT_SCORE,
|
||||
DEFAULT_ATTRIBUTION,
|
||||
)
|
||||
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
|
||||
from homeassistant.components.unifiprotect.sensor import (
|
||||
ALL_DEVICES_SENSORS,
|
||||
CAMERA_DISABLED_SENSORS,
|
||||
CAMERA_SENSORS,
|
||||
LICENSE_PLATE_EVENT_SENSORS,
|
||||
MOTION_TRIP_SENSORS,
|
||||
NVR_DISABLED_SENSORS,
|
||||
NVR_SENSORS,
|
||||
@@ -34,12 +22,11 @@ from homeassistant.components.unifiprotect.sensor import (
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ATTRIBUTION,
|
||||
EVENT_STATE_CHANGED,
|
||||
STATE_UNAVAILABLE,
|
||||
STATE_UNKNOWN,
|
||||
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 (
|
||||
@@ -54,8 +41,6 @@ from .utils import (
|
||||
time_changed,
|
||||
)
|
||||
|
||||
from tests.common import async_capture_events
|
||||
|
||||
|
||||
def get_sensor_by_key(sensors: tuple, key: str) -> ProtectSensorEntityDescription:
|
||||
"""Get sensor description by key."""
|
||||
@@ -543,441 +528,6 @@ async def test_sensor_update_alarm_with_last_trip_time(
|
||||
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
|
||||
|
||||
|
||||
async def test_camera_update_license_plate(
|
||||
hass: HomeAssistant, ufp: MockUFPFixture, camera: Camera, fixed_now: datetime
|
||||
) -> None:
|
||||
"""Test license plate sensor."""
|
||||
|
||||
camera.feature_flags.smart_detect_types.append(SmartDetectObjectType.LICENSE_PLATE)
|
||||
camera.feature_flags.has_smart_detect = True
|
||||
camera.smart_detect_settings.object_types.append(
|
||||
SmartDetectObjectType.LICENSE_PLATE
|
||||
)
|
||||
|
||||
await init_entry(hass, ufp, [camera])
|
||||
assert_entity_counts(hass, Platform.SENSOR, 23, 13)
|
||||
|
||||
_, entity_id = await ids_from_device_description(
|
||||
hass,
|
||||
Platform.SENSOR,
|
||||
camera,
|
||||
get_sensor_by_key(LICENSE_PLATE_EVENT_SENSORS, "smart_obj_licenseplate"),
|
||||
)
|
||||
|
||||
event_metadata = EventMetadata(
|
||||
license_plate=LicensePlateMetadata(name="ABCD1234", confidence_level=95)
|
||||
)
|
||||
event = Event(
|
||||
model=ModelType.EVENT,
|
||||
id="test_event_id",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=fixed_now - timedelta(seconds=1),
|
||||
end=None,
|
||||
score=100,
|
||||
smart_detect_types=[SmartDetectObjectType.LICENSE_PLATE],
|
||||
smart_detect_event_ids=[],
|
||||
metadata=event_metadata,
|
||||
api=ufp.api,
|
||||
)
|
||||
|
||||
new_camera = camera.model_copy()
|
||||
new_camera.is_smart_detected = True
|
||||
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.LICENSE_PLATE] = (
|
||||
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}
|
||||
|
||||
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 == "ABCD1234"
|
||||
|
||||
assert len(state_changes) == 1
|
||||
|
||||
# ensure reply is ignored
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 1
|
||||
|
||||
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=100,
|
||||
smart_detect_types=[SmartDetectObjectType.LICENSE_PLATE],
|
||||
smart_detect_event_ids=[],
|
||||
metadata=event_metadata,
|
||||
api=ufp.api,
|
||||
)
|
||||
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.LICENSE_PLATE] = (
|
||||
event.id
|
||||
)
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 2
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "none"
|
||||
|
||||
# Now send a new event with end already set
|
||||
event = Event(
|
||||
model=ModelType.EVENT,
|
||||
id="new_event",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=fixed_now - timedelta(seconds=1),
|
||||
end=fixed_now + timedelta(seconds=1),
|
||||
score=100,
|
||||
smart_detect_types=[SmartDetectObjectType.LICENSE_PLATE],
|
||||
smart_detect_event_ids=[],
|
||||
metadata=event_metadata,
|
||||
api=ufp.api,
|
||||
)
|
||||
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.LICENSE_PLATE] = (
|
||||
event.id
|
||||
)
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 4
|
||||
assert state_changes[2].data["new_state"].state == "ABCD1234"
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "none"
|
||||
|
||||
|
||||
async def test_camera_update_license_plate_changes_number_during_detect(
|
||||
hass: HomeAssistant, ufp: MockUFPFixture, camera: Camera, fixed_now: datetime
|
||||
) -> None:
|
||||
"""Test license plate sensor that changes number during detect."""
|
||||
|
||||
camera.feature_flags.smart_detect_types.append(SmartDetectObjectType.LICENSE_PLATE)
|
||||
camera.feature_flags.has_smart_detect = True
|
||||
camera.smart_detect_settings.object_types.append(
|
||||
SmartDetectObjectType.LICENSE_PLATE
|
||||
)
|
||||
|
||||
await init_entry(hass, ufp, [camera])
|
||||
assert_entity_counts(hass, Platform.SENSOR, 23, 13)
|
||||
|
||||
_, entity_id = await ids_from_device_description(
|
||||
hass,
|
||||
Platform.SENSOR,
|
||||
camera,
|
||||
get_sensor_by_key(LICENSE_PLATE_EVENT_SENSORS, "smart_obj_licenseplate"),
|
||||
)
|
||||
|
||||
event_metadata = EventMetadata(
|
||||
license_plate=LicensePlateMetadata(name="ABCD1234", confidence_level=95)
|
||||
)
|
||||
event = Event(
|
||||
model=ModelType.EVENT,
|
||||
id="test_event_id",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=fixed_now - timedelta(seconds=1),
|
||||
end=None,
|
||||
score=100,
|
||||
smart_detect_types=[SmartDetectObjectType.LICENSE_PLATE],
|
||||
smart_detect_event_ids=[],
|
||||
metadata=event_metadata,
|
||||
api=ufp.api,
|
||||
)
|
||||
|
||||
new_camera = camera.model_copy()
|
||||
new_camera.is_smart_detected = True
|
||||
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.LICENSE_PLATE] = (
|
||||
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}
|
||||
|
||||
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 == "ABCD1234"
|
||||
|
||||
assert len(state_changes) == 1
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 1
|
||||
|
||||
# Now mutate the original event so it ends
|
||||
# Also change the metadata to a different license plate
|
||||
# since the model may not get the plate correct on
|
||||
# the first update.
|
||||
event.score = 99
|
||||
event.end = fixed_now + timedelta(seconds=1)
|
||||
event_metadata.license_plate.name = "DCBA4321"
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 3
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "none"
|
||||
|
||||
assert state_changes[0].data["new_state"].state == "ABCD1234"
|
||||
assert state_changes[1].data["new_state"].state == "DCBA4321"
|
||||
assert state_changes[2].data["new_state"].state == "none"
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "none"
|
||||
|
||||
|
||||
async def test_camera_update_license_plate_multiple_updates(
|
||||
hass: HomeAssistant, ufp: MockUFPFixture, camera: Camera, fixed_now: datetime
|
||||
) -> None:
|
||||
"""Test license plate sensor that updates multiple times."""
|
||||
|
||||
camera.feature_flags.smart_detect_types.append(SmartDetectObjectType.LICENSE_PLATE)
|
||||
camera.feature_flags.has_smart_detect = True
|
||||
camera.smart_detect_settings.object_types.append(
|
||||
SmartDetectObjectType.LICENSE_PLATE
|
||||
)
|
||||
|
||||
await init_entry(hass, ufp, [camera])
|
||||
assert_entity_counts(hass, Platform.SENSOR, 23, 13)
|
||||
|
||||
_, entity_id = await ids_from_device_description(
|
||||
hass,
|
||||
Platform.SENSOR,
|
||||
camera,
|
||||
get_sensor_by_key(LICENSE_PLATE_EVENT_SENSORS, "smart_obj_licenseplate"),
|
||||
)
|
||||
|
||||
event_metadata = EventMetadata(
|
||||
license_plate=LicensePlateMetadata(name="ABCD1234", confidence_level=95)
|
||||
)
|
||||
event = Event(
|
||||
model=ModelType.EVENT,
|
||||
id="test_event_id",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=fixed_now - timedelta(seconds=1),
|
||||
end=None,
|
||||
score=100,
|
||||
smart_detect_types=[SmartDetectObjectType.LICENSE_PLATE],
|
||||
smart_detect_event_ids=[],
|
||||
metadata=event_metadata,
|
||||
api=ufp.api,
|
||||
)
|
||||
|
||||
new_camera = camera.model_copy()
|
||||
new_camera.is_smart_detected = True
|
||||
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.LICENSE_PLATE] = (
|
||||
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}
|
||||
|
||||
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 == "ABCD1234"
|
||||
assert state.attributes[ATTR_EVENT_SCORE] == 100
|
||||
|
||||
assert len(state_changes) == 1
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 1
|
||||
|
||||
# Now mutate the original event so the score changes
|
||||
event.score = 99
|
||||
event_metadata.license_plate.name = "DCBA4321"
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 2
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "DCBA4321"
|
||||
assert state.attributes[ATTR_EVENT_SCORE] == 99
|
||||
|
||||
# Now mutate the original event so the score changes again
|
||||
event.score = 40
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 3
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "DCBA4321"
|
||||
assert state.attributes[ATTR_EVENT_SCORE] == 40
|
||||
|
||||
# Now send the event again
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 3
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "DCBA4321"
|
||||
assert state.attributes[ATTR_EVENT_SCORE] == 40
|
||||
|
||||
# Now mutate the original event to add an end time
|
||||
event.end = fixed_now + timedelta(seconds=1)
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 4
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "none"
|
||||
|
||||
# Now send the event again
|
||||
event.end = fixed_now + timedelta(seconds=1)
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 4
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "none"
|
||||
|
||||
|
||||
async def test_camera_update_license_no_dupes(
|
||||
hass: HomeAssistant, ufp: MockUFPFixture, camera: Camera, fixed_now: datetime
|
||||
) -> None:
|
||||
"""Test license plate sensor does not generate duplicate reads."""
|
||||
|
||||
camera.feature_flags.smart_detect_types.append(SmartDetectObjectType.LICENSE_PLATE)
|
||||
camera.feature_flags.has_smart_detect = True
|
||||
camera.smart_detect_settings.object_types.append(
|
||||
SmartDetectObjectType.LICENSE_PLATE
|
||||
)
|
||||
|
||||
await init_entry(hass, ufp, [camera])
|
||||
assert_entity_counts(hass, Platform.SENSOR, 23, 13)
|
||||
|
||||
_, entity_id = await ids_from_device_description(
|
||||
hass,
|
||||
Platform.SENSOR,
|
||||
camera,
|
||||
get_sensor_by_key(LICENSE_PLATE_EVENT_SENSORS, "smart_obj_licenseplate"),
|
||||
)
|
||||
|
||||
event_metadata = EventMetadata(
|
||||
license_plate=LicensePlateMetadata(name="FPR2238", confidence_level=91)
|
||||
)
|
||||
event = Event(
|
||||
model=ModelType.EVENT,
|
||||
id="6675e36400de8c03e40bd5e3",
|
||||
type=EventType.SMART_DETECT,
|
||||
start=fixed_now - timedelta(seconds=1),
|
||||
end=None,
|
||||
score=83,
|
||||
smart_detect_types=[SmartDetectObjectType.LICENSE_PLATE],
|
||||
smart_detect_event_ids=[],
|
||||
metadata=event_metadata,
|
||||
api=ufp.api,
|
||||
)
|
||||
|
||||
new_camera = camera.model_copy()
|
||||
new_camera.is_smart_detected = True
|
||||
new_camera.last_smart_detect_event_ids[SmartDetectObjectType.LICENSE_PLATE] = (
|
||||
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}
|
||||
|
||||
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 == "FPR2238"
|
||||
assert state.attributes[ATTR_EVENT_SCORE] == 83
|
||||
|
||||
assert len(state_changes) == 1
|
||||
|
||||
# Now send it again
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 1
|
||||
|
||||
# Again send it again
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 1
|
||||
|
||||
# Now add the end time and change the confidence level
|
||||
event.end = fixed_now + timedelta(seconds=1)
|
||||
event.metadata.license_plate.confidence_level = 96
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 2
|
||||
|
||||
state = hass.states.get(entity_id)
|
||||
assert state
|
||||
assert state.state == "none"
|
||||
|
||||
# Now send it 3 more times
|
||||
for _ in range(3):
|
||||
ufp.api.bootstrap.events = {event.id: event}
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 2
|
||||
|
||||
# Now clear the event
|
||||
ufp.api.bootstrap.events = {}
|
||||
ufp.ws_msg(mock_msg)
|
||||
await hass.async_block_till_done()
|
||||
assert len(state_changes) == 2
|
||||
|
||||
|
||||
async def test_sensor_precision(
|
||||
hass: HomeAssistant, ufp: MockUFPFixture, sensor_all: Sensor, fixed_now: datetime
|
||||
) -> None:
|
||||
|
||||
Reference in New Issue
Block a user