Update nest to only include the image attachment payload for cameras that support fetching media (#124590)

Only include the image attachment payload for cameras that support fetching media
pull/124880/head
Allen Porter 2024-08-31 11:38:45 -07:00 committed by Bram Kragten
parent 9a690ed421
commit b81d7a0ed8
2 changed files with 57 additions and 17 deletions

View File

@ -166,38 +166,43 @@ class SignalUpdateCallback:
)
if not device_entry:
return
supported_traits = self._supported_traits(device_id)
for api_event_type, image_event in events.items():
if not (event_type := EVENT_NAME_MAP.get(api_event_type)):
continue
nest_event_id = image_event.event_token
attachment = {
"image": EVENT_THUMBNAIL_URL_FORMAT.format(
device_id=device_entry.id, event_token=image_event.event_token
),
}
if self._supports_clip(device_id):
attachment["video"] = EVENT_MEDIA_API_URL_FORMAT.format(
device_id=device_entry.id, event_token=image_event.event_token
)
message = {
"device_id": device_entry.id,
"type": event_type,
"timestamp": event_message.timestamp,
"nest_event_id": nest_event_id,
"attachment": attachment,
}
if (
TraitType.CAMERA_EVENT_IMAGE in supported_traits
or TraitType.CAMERA_CLIP_PREVIEW in supported_traits
):
attachment = {
"image": EVENT_THUMBNAIL_URL_FORMAT.format(
device_id=device_entry.id, event_token=image_event.event_token
)
}
if TraitType.CAMERA_CLIP_PREVIEW in supported_traits:
attachment["video"] = EVENT_MEDIA_API_URL_FORMAT.format(
device_id=device_entry.id, event_token=image_event.event_token
)
message["attachment"] = attachment
if image_event.zones:
message["zones"] = image_event.zones
self._hass.bus.async_fire(NEST_EVENT, message)
def _supports_clip(self, device_id: str) -> bool:
def _supported_traits(self, device_id: str) -> list[TraitType]:
if not (
device_manager := self._hass.data[DOMAIN]
.get(self._config_entry_id, {})
.get(DATA_DEVICE_MANAGER)
) or not (device := device_manager.devices.get(device_id)):
return False
return TraitType.CAMERA_CLIP_PREVIEW in device.traits
return []
return list(device.traits)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

View File

@ -122,28 +122,28 @@ def create_events(events, device_id=DEVICE_ID, timestamp=None):
[
(
"sdm.devices.types.DOORBELL",
["sdm.devices.traits.DoorbellChime"],
["sdm.devices.traits.DoorbellChime", "sdm.devices.traits.CameraEventImage"],
"sdm.devices.events.DoorbellChime.Chime",
"Doorbell",
"doorbell_chime",
),
(
"sdm.devices.types.CAMERA",
["sdm.devices.traits.CameraMotion"],
["sdm.devices.traits.CameraMotion", "sdm.devices.traits.CameraEventImage"],
"sdm.devices.events.CameraMotion.Motion",
"Camera",
"camera_motion",
),
(
"sdm.devices.types.CAMERA",
["sdm.devices.traits.CameraPerson"],
["sdm.devices.traits.CameraPerson", "sdm.devices.traits.CameraEventImage"],
"sdm.devices.events.CameraPerson.Person",
"Camera",
"camera_person",
),
(
"sdm.devices.types.CAMERA",
["sdm.devices.traits.CameraSound"],
["sdm.devices.traits.CameraSound", "sdm.devices.traits.CameraEventImage"],
"sdm.devices.events.CameraSound.Sound",
"Camera",
"camera_sound",
@ -234,6 +234,41 @@ async def test_camera_multiple_event(
}
@pytest.mark.parametrize(
"device_traits",
[(["sdm.devices.traits.CameraMotion"])],
)
async def test_media_not_supported(
hass: HomeAssistant, entity_registry: er.EntityRegistry, subscriber, setup_platform
) -> None:
"""Test a pubsub message for a camera person event."""
events = async_capture_events(hass, NEST_EVENT)
await setup_platform()
entry = entity_registry.async_get("camera.front")
assert entry is not None
event_map = {
"sdm.devices.events.CameraMotion.Motion": {
"eventSessionId": EVENT_SESSION_ID,
"eventId": EVENT_ID,
},
}
timestamp = utcnow()
await subscriber.async_receive_event(create_events(event_map, timestamp=timestamp))
await hass.async_block_till_done()
event_time = timestamp.replace(microsecond=0)
assert len(events) == 1
assert event_view(events[0].data) == {
"device_id": entry.device_id,
"type": "camera_motion",
"timestamp": event_time,
}
# Media fetching not supported by this device
assert "attachment" not in events[0].data
async def test_unknown_event(hass: HomeAssistant, subscriber, setup_platform) -> None:
"""Test a pubsub message for an unknown event type."""
events = async_capture_events(hass, NEST_EVENT)