Add video event proxy endpoint for unifiprotect (#129980)

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Lutz 2024-11-27 18:03:21 +01:00 committed by GitHub
parent 1450fe0880
commit fda178da23
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 313 additions and 25 deletions

View File

@ -45,7 +45,7 @@ from .utils import (
async_create_api_client,
async_get_devices,
)
from .views import ThumbnailProxyView, VideoProxyView
from .views import ThumbnailProxyView, VideoEventProxyView, VideoProxyView
_LOGGER = logging.getLogger(__name__)
@ -174,6 +174,7 @@ async def _async_setup_entry(
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
hass.http.register_view(ThumbnailProxyView(hass))
hass.http.register_view(VideoProxyView(hass))
hass.http.register_view(VideoEventProxyView(hass))
async def _async_options_updated(hass: HomeAssistant, entry: UFPConfigEntry) -> None:

View File

@ -5,7 +5,7 @@ from __future__ import annotations
from datetime import datetime
from http import HTTPStatus
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
from urllib.parse import urlencode
from aiohttp import web
@ -30,7 +30,9 @@ def async_generate_thumbnail_url(
) -> str:
"""Generate URL for event thumbnail."""
url_format = ThumbnailProxyView.url or "{nvr_id}/{event_id}"
url_format = ThumbnailProxyView.url
if TYPE_CHECKING:
assert url_format is not None
url = url_format.format(nvr_id=nvr_id, event_id=event_id)
params = {}
@ -50,7 +52,9 @@ def async_generate_event_video_url(event: Event) -> str:
if event.start is None or event.end is None:
raise ValueError("Event is ongoing")
url_format = VideoProxyView.url or "{nvr_id}/{camera_id}/{start}/{end}"
url_format = VideoProxyView.url
if TYPE_CHECKING:
assert url_format is not None
return url_format.format(
nvr_id=event.api.bootstrap.nvr.id,
camera_id=event.camera_id,
@ -59,6 +63,19 @@ def async_generate_event_video_url(event: Event) -> str:
)
@callback
def async_generate_proxy_event_video_url(
nvr_id: str,
event_id: str,
) -> str:
"""Generate proxy URL for event video."""
url_format = VideoEventProxyView.url
if TYPE_CHECKING:
assert url_format is not None
return url_format.format(nvr_id=nvr_id, event_id=event_id)
@callback
def _client_error(message: Any, code: HTTPStatus) -> web.Response:
_LOGGER.warning("Client error (%s): %s", code.value, message)
@ -107,6 +124,27 @@ class ProtectProxyView(HomeAssistantView):
return data
return _404("Invalid NVR ID")
@callback
def _async_get_camera(self, data: ProtectData, camera_id: str) -> Camera | None:
if (camera := data.api.bootstrap.cameras.get(camera_id)) is not None:
return camera
entity_registry = er.async_get(self.hass)
device_registry = dr.async_get(self.hass)
if (entity := entity_registry.async_get(camera_id)) is None or (
device := device_registry.async_get(entity.device_id or "")
) is None:
return None
macs = [c[1] for c in device.connections if c[0] == dr.CONNECTION_NETWORK_MAC]
for mac in macs:
if (ufp_device := data.api.bootstrap.get_device_from_mac(mac)) is not None:
if isinstance(ufp_device, Camera):
camera = ufp_device
break
return camera
class ThumbnailProxyView(ProtectProxyView):
"""View to proxy event thumbnails from UniFi Protect."""
@ -156,27 +194,6 @@ class VideoProxyView(ProtectProxyView):
url = "/api/unifiprotect/video/{nvr_id}/{camera_id}/{start}/{end}"
name = "api:unifiprotect_thumbnail"
@callback
def _async_get_camera(self, data: ProtectData, camera_id: str) -> Camera | None:
if (camera := data.api.bootstrap.cameras.get(camera_id)) is not None:
return camera
entity_registry = er.async_get(self.hass)
device_registry = dr.async_get(self.hass)
if (entity := entity_registry.async_get(camera_id)) is None or (
device := device_registry.async_get(entity.device_id or "")
) is None:
return None
macs = [c[1] for c in device.connections if c[0] == dr.CONNECTION_NETWORK_MAC]
for mac in macs:
if (ufp_device := data.api.bootstrap.get_device_from_mac(mac)) is not None:
if isinstance(ufp_device, Camera):
camera = ufp_device
break
return camera
async def get(
self, request: web.Request, nvr_id: str, camera_id: str, start: str, end: str
) -> web.StreamResponse:
@ -226,3 +243,56 @@ class VideoProxyView(ProtectProxyView):
if response.prepared:
await response.write_eof()
return response
class VideoEventProxyView(ProtectProxyView):
"""View to proxy video clips for events from UniFi Protect."""
url = "/api/unifiprotect/video/{nvr_id}/{event_id}"
name = "api:unifiprotect_videoEventView"
async def get(
self, request: web.Request, nvr_id: str, event_id: str
) -> web.StreamResponse:
"""Get Camera Video clip for an event."""
data = self._get_data_or_404(nvr_id)
if isinstance(data, web.Response):
return data
try:
event = await data.api.get_event(event_id)
except ClientError:
return _404(f"Invalid event ID: {event_id}")
if event.start is None or event.end is None:
return _400("Event is still ongoing")
camera = self._async_get_camera(data, str(event.camera_id))
if camera is None:
return _404(f"Invalid camera ID: {event.camera_id}")
if not camera.can_read_media(data.api.bootstrap.auth_user):
return _403(f"User cannot read media from camera: {camera.id}")
response = web.StreamResponse(
status=200,
reason="OK",
headers={
"Content-Type": "video/mp4",
},
)
async def iterator(total: int, chunk: bytes | None) -> None:
if not response.prepared:
response.content_length = total
await response.prepare(request)
if chunk is not None:
await response.write(chunk)
try:
await camera.get_video(event.start, event.end, iterator_callback=iterator)
except ClientError as err:
return _404(err)
if response.prepared:
await response.write_eof()
return response

View File

@ -11,6 +11,7 @@ from uiprotect.exceptions import ClientError
from homeassistant.components.unifiprotect.views import (
async_generate_event_video_url,
async_generate_proxy_event_video_url,
async_generate_thumbnail_url,
)
from homeassistant.core import HomeAssistant
@ -520,3 +521,219 @@ async def test_video_entity_id(
assert response.status == 200
ufp.api.request.assert_called_once()
async def test_video_event_bad_nvr_id(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
camera: Camera,
ufp: MockUFPFixture,
) -> None:
"""Test video proxy URL with bad NVR id."""
ufp.api.request = AsyncMock()
await init_entry(hass, ufp, [camera])
url = async_generate_proxy_event_video_url("bad_id", "test_id")
http_client = await hass_client()
response = cast(ClientResponse, await http_client.get(url))
assert response.status == 404
ufp.api.request.assert_not_called()
async def test_video_event_bad_event(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
ufp: MockUFPFixture,
camera: Camera,
) -> None:
"""Test generating event with bad event ID."""
ufp.api.get_event = AsyncMock(side_effect=ClientError())
await init_entry(hass, ufp, [camera])
url = async_generate_proxy_event_video_url(ufp.api.bootstrap.nvr.id, "bad_event_id")
http_client = await hass_client()
response = cast(ClientResponse, await http_client.get(url))
assert response.status == 404
ufp.api.request.assert_not_called()
async def test_video_event_bad_camera(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
ufp: MockUFPFixture,
camera: Camera,
) -> None:
"""Test generating event with bad camera ID."""
ufp.api.get_event = AsyncMock(side_effect=ClientError())
await init_entry(hass, ufp, [camera])
url = async_generate_proxy_event_video_url(ufp.api.bootstrap.nvr.id, "bad_event_id")
http_client = await hass_client()
response = cast(ClientResponse, await http_client.get(url))
assert response.status == 404
ufp.api.request.assert_not_called()
async def test_video_event_bad_camera_perms(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
ufp: MockUFPFixture,
camera: Camera,
fixed_now: datetime,
) -> None:
"""Test video URL with bad camera perms."""
ufp.api.request = AsyncMock()
await init_entry(hass, ufp, [camera])
event_start = fixed_now - timedelta(seconds=30)
event = Event(
model=ModelType.EVENT,
api=ufp.api,
start=event_start,
end=fixed_now,
id="test_id",
type=EventType.MOTION,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id="bad_id",
camera=camera,
)
ufp.api.get_event = AsyncMock(return_value=event)
url = async_generate_proxy_event_video_url(ufp.api.bootstrap.nvr.id, "test_id")
ufp.api.bootstrap.auth_user.all_permissions = []
ufp.api.bootstrap.auth_user._perm_cache = {}
http_client = await hass_client()
response = cast(ClientResponse, await http_client.get(url))
assert response.status == 404
ufp.api.request.assert_not_called()
async def test_video_event_ongoing(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
ufp: MockUFPFixture,
camera: Camera,
fixed_now: datetime,
) -> None:
"""Test video URL with ongoing event."""
ufp.api.request = AsyncMock()
await init_entry(hass, ufp, [camera])
event_start = fixed_now - timedelta(seconds=30)
event = Event(
model=ModelType.EVENT,
api=ufp.api,
start=event_start,
id="test_id",
type=EventType.MOTION,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=camera.id,
camera=camera,
)
ufp.api.get_event = AsyncMock(return_value=event)
url = async_generate_proxy_event_video_url(ufp.api.bootstrap.nvr.id, "test_id")
http_client = await hass_client()
response = cast(ClientResponse, await http_client.get(url))
assert response.status == 400
ufp.api.request.assert_not_called()
async def test_event_video_no_data(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
ufp: MockUFPFixture,
camera: Camera,
fixed_now: datetime,
) -> None:
"""Test invalid no event video returned."""
await init_entry(hass, ufp, [camera])
event_start = fixed_now - timedelta(seconds=30)
event = Event(
model=ModelType.EVENT,
api=ufp.api,
start=event_start,
end=fixed_now,
id="test_id",
type=EventType.MOTION,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=camera.id,
camera=camera,
)
ufp.api.request = AsyncMock(side_effect=ClientError)
ufp.api.get_event = AsyncMock(return_value=event)
url = async_generate_proxy_event_video_url(ufp.api.bootstrap.nvr.id, "test_id")
http_client = await hass_client()
response = cast(ClientResponse, await http_client.get(url))
assert response.status == 404
async def test_event_video(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
ufp: MockUFPFixture,
camera: Camera,
fixed_now: datetime,
) -> None:
"""Test event video URL with no video."""
content = Mock()
content.__anext__ = AsyncMock(side_effect=[b"test", b"test", StopAsyncIteration()])
content.__aiter__ = Mock(return_value=content)
mock_response = Mock()
mock_response.content_length = 8
mock_response.content.iter_chunked = Mock(return_value=content)
ufp.api.request = AsyncMock(return_value=mock_response)
await init_entry(hass, ufp, [camera])
event_start = fixed_now - timedelta(seconds=30)
event = Event(
model=ModelType.EVENT,
api=ufp.api,
start=event_start,
end=fixed_now,
id="test_id",
type=EventType.MOTION,
score=100,
smart_detect_types=[],
smart_detect_event_ids=[],
camera_id=camera.id,
camera=camera,
)
ufp.api.get_event = AsyncMock(return_value=event)
url = async_generate_proxy_event_video_url(ufp.api.bootstrap.nvr.id, "test_id")
http_client = await hass_client()
response = cast(ClientResponse, await http_client.get(url))
assert await response.content.read() == b"testtest"
assert response.status == 200
ufp.api.request.assert_called_once()