Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add metadata for UniFi Protect Media Source (#109389)

pull/111180/head
Christopher Bailey 2024-02-22 19:44:16 -05:00 committed by GitHub
parent f0f3773858
commit ae60f59bed
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 302 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
homeassistant/components/unifiprotect/media_source.py
View File

@ -47,6 +47,7 @@ class SimpleEventType(str, Enum):
RING = "ring"
MOTION = "motion"
SMART = "smart"
AUDIO = "audio"
class IdentifierType(str, Enum):
@ -64,21 +65,29 @@ class IdentifierTimeType(str, Enum):
RANGE = "range"
EVENT_MAP = {
SimpleEventType.ALL: None,
SimpleEventType.RING: EventType.RING,
SimpleEventType.MOTION: EventType.MOTION,
SimpleEventType.SMART: EventType.SMART_DETECT,
EVENT_MAP: dict[SimpleEventType, set[EventType]] = {
SimpleEventType.ALL: {
EventType.RING,
EventType.MOTION,
EventType.SMART_DETECT,
EventType.SMART_DETECT_LINE,
EventType.SMART_AUDIO_DETECT,
},
SimpleEventType.RING: {EventType.RING},
SimpleEventType.MOTION: {EventType.MOTION},
SimpleEventType.SMART: {EventType.SMART_DETECT, EventType.SMART_DETECT_LINE},
SimpleEventType.AUDIO: {EventType.SMART_AUDIO_DETECT},
}
EVENT_NAME_MAP = {
SimpleEventType.ALL: "All Events",
SimpleEventType.RING: "Ring Events",
SimpleEventType.MOTION: "Motion Events",
SimpleEventType.SMART: "Smart Detections",
SimpleEventType.SMART: "Object Detections",
SimpleEventType.AUDIO: "Audio Detections",
}
def get_ufp_event(event_type: SimpleEventType) -> EventType | None:
def get_ufp_event(event_type: SimpleEventType) -> set[EventType]:
"""Get UniFi Protect event type from SimpleEventType."""
return EVENT_MAP[event_type]
@ -132,6 +141,51 @@ def _format_duration(duration: timedelta) -> str:
return formatted.strip()
@callback
def _get_object_name(event: Event | dict[str, Any]) -> str:
if isinstance(event, Event):
event = event.unifi_dict()
names = []
types = set(event["smartDetectTypes"])
metadata = event.get("metadata") or {}
for thumb in metadata.get("detectedThumbnails", []):
thumb_type = thumb.get("type")
if thumb_type not in types:
continue
types.remove(thumb_type)
if thumb_type == SmartDetectObjectType.VEHICLE.value:
attributes = thumb.get("attributes") or {}
color = attributes.get("color", {}).get("val", "")
vehicle_type = attributes.get("vehicleType", {}).get("val", "vehicle")
license_plate = metadata.get("licensePlate", {}).get("name")
name = f"{color} {vehicle_type}".strip().title()
if license_plate:
types.remove(SmartDetectObjectType.LICENSE_PLATE.value)
name = f"{name}: {license_plate}"
names.append(name)
else:
smart_type = SmartDetectObjectType(thumb_type)
names.append(smart_type.name.title().replace("_", " "))
for raw in types:
smart_type = SmartDetectObjectType(raw)
names.append(smart_type.name.title().replace("_", " "))
return ", ".join(sorted(names))
@callback
def _get_audio_name(event: Event | dict[str, Any]) -> str:
if isinstance(event, Event):
event = event.unifi_dict()
smart_types = [SmartDetectObjectType(e) for e in event["smartDetectTypes"]]
return ", ".join([s.name.title().replace("_", " ") for s in smart_types])
class ProtectMediaSource(MediaSource):
"""Represents all UniFi Protect NVRs."""
@ -384,7 +438,7 @@ class ProtectMediaSource(MediaSource):
end = event.end
else:
event_id = event["id"]
event_type = event["type"]
event_type = EventType(event["type"])
start = from_js_time(event["start"])
end = from_js_time(event["end"])
@ -393,19 +447,14 @@ class ProtectMediaSource(MediaSource):
title = dt_util.as_local(start).strftime("%x %X")
duration = end - start
title += f" {_format_duration(duration)}"
if event_type == EventType.RING.value:
if event_type in EVENT_MAP[SimpleEventType.RING]:
event_text = "Ring Event"
elif event_type == EventType.MOTION.value:
elif event_type in EVENT_MAP[SimpleEventType.MOTION]:
event_text = "Motion Event"
elif event_type == EventType.SMART_DETECT.value:
if isinstance(event, Event):
smart_types = event.smart_detect_types
else:
smart_types = [
SmartDetectObjectType(e) for e in event["smartDetectTypes"]
]
smart_type_names = [s.name.title().replace("_", " ") for s in smart_types]
event_text = f"Smart Detection - {','.join(smart_type_names)}"
elif event_type in EVENT_MAP[SimpleEventType.SMART]:
event_text = f"Object Detection - {_get_object_name(event)}"
elif event_type in EVENT_MAP[SimpleEventType.AUDIO]:
event_text = f"Audio Detection - {_get_audio_name(event)}"
title += f" {event_text}"
nvr = data.api.bootstrap.nvr
@ -442,20 +491,13 @@ class ProtectMediaSource(MediaSource):
start: datetime,
end: datetime,
camera_id: str | None = None,
event_type: EventType | None = None,
event_types: set[EventType] | None = None,
reserve: bool = False,
) -> list[BrowseMediaSource]:
"""Build media source for a given range of time and event type."""
if event_type is None:
types = [
EventType.RING,
EventType.MOTION,
EventType.SMART_DETECT,
]
else:
types = [event_type]
event_types = event_types or get_ufp_event(SimpleEventType.ALL)
types = list(event_types)
sources: list[BrowseMediaSource] = []
events = await data.api.get_events_raw(
start=start, end=end, types=types, limit=data.max_events
@ -515,9 +557,8 @@ class ProtectMediaSource(MediaSource):
"start": now - timedelta(days=days),
"end": now,
"reserve": True,
"event_types": get_ufp_event(event_type),
}
if event_type != SimpleEventType.ALL:
args["event_type"] = get_ufp_event(event_type)
camera: Camera | None = None
if camera_id != "all":
@ -646,9 +687,8 @@ class ProtectMediaSource(MediaSource):
"start": start_dt,
"end": end_dt,
"reserve": False,
"event_types": get_ufp_event(event_type),
}
if event_type != SimpleEventType.ALL:
args["event_type"] = get_ufp_event(event_type)
camera: Camera | None = None
if camera_id != "all":
@ -798,6 +838,9 @@ class ProtectMediaSource(MediaSource):
source.children.append(
await self._build_events_type(data, camera_id, SimpleEventType.SMART)
)
source.children.append(
await self._build_events_type(data, camera_id, SimpleEventType.AUDIO)
)
if is_doorbell or has_smart:
source.children.insert(

239
tests/components/unifiprotect/test_media_source.py
View File

@ -421,15 +421,17 @@ async def test_browse_media_event_type(
assert browse.title == "UnifiProtect > All Cameras"
assert browse.identifier == "test_id:browse:all"
assert len(browse.children) == 4
assert len(browse.children) == 5
assert browse.children[0].title == "All Events"
assert browse.children[0].identifier == "test_id:browse:all:all"
assert browse.children[1].title == "Ring Events"
assert browse.children[1].identifier == "test_id:browse:all:ring"
assert browse.children[2].title == "Motion Events"
assert browse.children[2].identifier == "test_id:browse:all:motion"
assert browse.children[3].title == "Smart Detections"
assert browse.children[3].title == "Object Detections"
assert browse.children[3].identifier == "test_id:browse:all:smart"
assert browse.children[4].title == "Audio Detections"
assert browse.children[4].identifier == "test_id:browse:all:audio"
ONE_MONTH_SIMPLE = (
@ -649,24 +651,232 @@ async def test_browse_media_recent_truncated(
assert browse.children[0].identifier == "test_id:event:test_event_id"
@pytest.mark.parametrize(
("event", "expected_title"),
[
(
Event(
id="test_event_id",
type=EventType.RING,
start=datetime(1000, 1, 1, 0, 0, 0),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id="test",
),
"Ring Event",
),
(
Event(
id="test_event_id",
type=EventType.MOTION,
start=datetime(1000, 1, 1, 0, 0, 0),
end=None,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id="test",
),
"Motion Event",
),
(
Event(
id="test_event_id",
type=EventType.SMART_DETECT,
start=datetime(1000, 1, 1, 0, 0, 0),
end=None,
score=100,
smart_detect_types=["person"],
smart_detect_event_ids=[],
camera_id="test",
metadata={
"detected_thumbnails": [
{
"clock_best_wall": datetime(1000, 1, 1, 0, 0, 0),
"type": "person",
"cropped_id": "event_id",
}
],
},
),
"Object Detection - Person",
),
(
Event(
id="test_event_id",
type=EventType.SMART_DETECT,
start=datetime(1000, 1, 1, 0, 0, 0),
end=None,
score=100,
smart_detect_types=["vehicle", "person"],
smart_detect_event_ids=[],
camera_id="test",
),
"Object Detection - Person, Vehicle",
),
(
Event(
id="test_event_id",
type=EventType.SMART_DETECT,
start=datetime(1000, 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(
id="test_event_id",
type=EventType.SMART_DETECT,
start=datetime(1000, 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(1000, 1, 1, 0, 0, 0),
"type": "vehicle",
"cropped_id": "event_id",
}
],
},
),
"Object Detection - Vehicle: ABC1234",
),
(
Event(
id="test_event_id",
type=EventType.SMART_DETECT,
start=datetime(1000, 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(1000, 1, 1, 0, 0, 0),
"type": "vehicle",
"cropped_id": "event_id",
"attributes": {
"vehicle_type": {
"val": "car",
"confidence": 95,
}
},
}
],
},
),
"Object Detection - Car: ABC1234",
),
(
Event(
id="test_event_id",
type=EventType.SMART_DETECT,
start=datetime(1000, 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(1000, 1, 1, 0, 0, 0),
"type": "vehicle",
"cropped_id": "event_id",
"attributes": {
"color": {
"val": "black",
"confidence": 95,
}
},
},
{
"clock_best_wall": datetime(1000, 1, 1, 0, 0, 0),
"type": "person",
"cropped_id": "event_id",
},
],
},
),
"Object Detection - Black Vehicle: ABC1234",
),
(
Event(
id="test_event_id",
type=EventType.SMART_DETECT,
start=datetime(1000, 1, 1, 0, 0, 0),
end=None,
score=100,
smart_detect_types=["vehicle"],
smart_detect_event_ids=[],
camera_id="test",
metadata={
"detected_thumbnails": [
{
"clock_best_wall": datetime(1000, 1, 1, 0, 0, 0),
"type": "vehicle",
"cropped_id": "event_id",
"attributes": {
"color": {
"val": "black",
"confidence": 95,
},
"vehicle_type": {
"val": "car",
"confidence": 95,
},
},
}
]
},
),
"Object Detection - Black Car",
),
(
Event(
id="test_event_id",
type=EventType.SMART_AUDIO_DETECT,
start=datetime(1000, 1, 1, 0, 0, 0),
end=None,
score=100,
smart_detect_types=["alrmSpeak"],
smart_detect_event_ids=[],
camera_id="test",
),
"Audio Detection - Speak",
),
],
)
async def test_browse_media_event(
hass: HomeAssistant, ufp: MockUFPFixture, doorbell: Camera, fixed_now: datetime
hass: HomeAssistant,
ufp: MockUFPFixture,
doorbell: Camera,
fixed_now: datetime,
event: Event,
expected_title: str,
) -> None:
"""Test browsing specific event."""
ufp.api.get_bootstrap = AsyncMock(return_value=ufp.api.bootstrap)
await init_entry(hass, ufp, [doorbell], regenerate_ids=False)
event = Event(
id="test_event_id",
type=EventType.RING,
start=fixed_now - timedelta(seconds=20),
end=fixed_now,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=doorbell.id,
)
event.start = fixed_now - timedelta(seconds=20)
event.end = fixed_now
event.camera_id = doorbell.id
event._api = ufp.api
ufp.api.get_event = AsyncMock(return_value=event)
@ -674,10 +884,13 @@ async def test_browse_media_event(
media_item = MediaSourceItem(hass, DOMAIN, "test_id:event:test_event_id", None)
browse = await source.async_browse_media(media_item)
# chop off the datetime/duration
title = " ".join(browse.title.split(" ")[3:])
assert browse.identifier == "test_id:event:test_event_id"
assert browse.children is None
assert browse.media_class == MediaClass.VIDEO
assert title == expected_title
async def test_browse_media_eventthumb(