diff --git a/homeassistant/components/reolink/camera.py b/homeassistant/components/reolink/camera.py index e9e4bd8ab4fa..f84c5e09d9b5 100644 --- a/homeassistant/components/reolink/camera.py +++ b/homeassistant/components/reolink/camera.py @@ -1,6 +1,7 @@ """Component providing support for Reolink IP cameras.""" from dataclasses import dataclass +from datetime import datetime import logging from typing import override @@ -9,8 +10,10 @@ from homeassistant.components.camera import ( CameraEntityDescription, CameraEntityFeature, ) +from homeassistant.components.ffmpeg import get_ffmpeg_manager from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from homeassistant.util import dt as dt_util from .entity import ReolinkChannelCoordinatorEntity, ReolinkChannelEntityDescription from .util import ReolinkConfigEntry, ReolinkData, raise_translated_error @@ -176,3 +179,18 @@ class ReolinkCamera(ReolinkChannelCoordinatorEntity, Camera): return await self._host.api.get_snapshot( self._channel, self.entity_description.stream ) + + @raise_translated_error + async def async_camera_image_past(self, timestamp: datetime) -> bytes: + """Return a still image from a past recording.""" + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=dt_util.get_default_time_zone()) + if (camera_tz := self._host.api.timezone()) is not None: + timestamp = timestamp.astimezone(camera_tz) + + return await self._host.api.baichuan.snapshot_past( + self._channel, + timestamp, + self.entity_description.stream, + get_ffmpeg_manager(self.hass).binary, + ) diff --git a/homeassistant/components/reolink/icons.json b/homeassistant/components/reolink/icons.json index 62b5222d3ec0..6da96ab48136 100644 --- a/homeassistant/components/reolink/icons.json +++ b/homeassistant/components/reolink/icons.json @@ -669,6 +669,9 @@ }, "ptz_move": { "service": "mdi:pan" + }, + "snapshot_past": { + "service": "mdi:camera-timer" } } } diff --git a/homeassistant/components/reolink/manifest.json b/homeassistant/components/reolink/manifest.json index c6ec0e7f5fb4..0169df036390 100644 --- a/homeassistant/components/reolink/manifest.json +++ b/homeassistant/components/reolink/manifest.json @@ -3,7 +3,7 @@ "name": "Reolink", "codeowners": ["@starkillerOG"], "config_flow": true, - "dependencies": ["http", "webhook"], + "dependencies": ["ffmpeg", "http", "webhook"], "dhcp": [ { "hostname": "reolink*" diff --git a/homeassistant/components/reolink/services.py b/homeassistant/components/reolink/services.py index 1328d363a07d..1fe107e8aa9f 100644 --- a/homeassistant/components/reolink/services.py +++ b/homeassistant/components/reolink/services.py @@ -1,22 +1,76 @@ """Reolink additional services.""" +import logging +import os +from typing import TYPE_CHECKING + import probatio from reolink_aio.api import Chime from reolink_aio.enums import ChimeToneEnum from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN -from homeassistant.const import ATTR_DEVICE_ID +from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN +from homeassistant.const import ATTR_DEVICE_ID, CONF_FILENAME from homeassistant.core import HomeAssistant, ServiceCall, callback -from homeassistant.exceptions import ServiceValidationError +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError from homeassistant.helpers import config_validation as cv, service +from homeassistant.helpers.template import Template from .const import DOMAIN, SUPPORT_PTZ_SPEED from .host import ReolinkHost from .util import get_device_uid_and_ch, raise_translated_error +_LOGGER = logging.getLogger(__name__) + +if TYPE_CHECKING: + from .camera import ReolinkCamera + ATTR_RINGTONE = "ringtone" ATTR_SPEED = "speed" +ATTR_TIMESTAMP = "timestamp" SERVICE_PTZ_MOVE = "ptz_move" +SERVICE_SNAPSHOT_PAST = "snapshot_past" + + +def _write_image(to_file: str, image: bytes) -> None: + """Write the image to a file, called in the executor.""" + os.makedirs(os.path.dirname(to_file), exist_ok=True) + with open(to_file, "wb") as img_file: + img_file.write(image) + + +async def _async_snapshot_past( + camera: ReolinkCamera, service_call: ServiceCall +) -> None: + """Save a snapshot of a past moment in time to a file.""" + hass = camera.hass + filename: Template = service_call.data[CONF_FILENAME] + snapshot_file = filename.async_render() + + if not await hass.async_add_executor_job( + hass.config.is_allowed_path, snapshot_file + ): + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="no_access_to_path", + translation_placeholders={"filename": snapshot_file}, + ) + + image = await camera.async_camera_image_past(service_call.data[ATTR_TIMESTAMP]) + + try: + await hass.async_add_executor_job(_write_image, snapshot_file, image) + except OSError as err: + _LOGGER.error( + "Reolink snapshot_past: Can't write image to '%s': %s", + snapshot_file, + err, + ) + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="write_image_error", + translation_placeholders={"filename": snapshot_file}, + ) from err @raise_translated_error @@ -70,3 +124,14 @@ def async_setup_services(hass: HomeAssistant) -> None: func="async_ptz_move", required_features=[SUPPORT_PTZ_SPEED], ) + service.async_register_platform_entity_service( + hass, + DOMAIN, + SERVICE_SNAPSHOT_PAST, + entity_domain=CAMERA_DOMAIN, + schema={ + probatio.Required(CONF_FILENAME): cv.template, + probatio.Required(ATTR_TIMESTAMP): cv.datetime, + }, + func=_async_snapshot_past, + ) diff --git a/homeassistant/components/reolink/services.yaml b/homeassistant/components/reolink/services.yaml index fe7fba9cdc7d..6cdf81a0cf49 100644 --- a/homeassistant/components/reolink/services.yaml +++ b/homeassistant/components/reolink/services.yaml @@ -43,3 +43,20 @@ play_chime: - operetta - moonlight - waybackhome + +snapshot_past: + target: + entity: + integration: reolink + domain: camera + fields: + timestamp: + required: true + example: "2025-09-29 14:30:00" + selector: + datetime: + filename: + required: true + example: "/tmp/reolink_snapshot.jpg" + selector: + text: diff --git a/homeassistant/components/reolink/strings.json b/homeassistant/components/reolink/strings.json index 1fcb860f906a..eeca670b902a 100644 --- a/homeassistant/components/reolink/strings.json +++ b/homeassistant/components/reolink/strings.json @@ -954,6 +954,9 @@ "login_error": { "message": "Error during login attempt: {err}" }, + "no_access_to_path": { + "message": "Cannot write `{filename}`, no access to path; `allowlist_external_dirs` may need to be adjusted in `configuration.yaml`" + }, "no_data": { "message": "Device returned no data: {err}" }, @@ -983,6 +986,9 @@ }, "webhook_no_url": { "message": "Error registering URL for webhook {event_id}: Home Assistant URL is not available" + }, + "write_image_error": { + "message": "Can't write image to `{filename}`, see the log for details" } }, "issues": { @@ -1063,6 +1069,20 @@ } }, "name": "PTZ move" + }, + "snapshot_past": { + "description": "Takes a snapshot from a past moment in time out of the recordings of a camera and stores it in a file.", + "fields": { + "filename": { + "description": "Full path to filename.", + "name": "Filename" + }, + "timestamp": { + "description": "Moment in time to take the snapshot from, in the local time of Home Assistant.", + "name": "Timestamp" + } + }, + "name": "Take camera snapshot from past" } } } diff --git a/tests/components/reolink/test_services.py b/tests/components/reolink/test_services.py index 38819bbd51d9..3af1fbb6c500 100644 --- a/tests/components/reolink/test_services.py +++ b/tests/components/reolink/test_services.py @@ -1,21 +1,33 @@ """Test the Reolink services.""" -from unittest.mock import AsyncMock, MagicMock, patch +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, mock_open, patch import pytest from reolink_aio.api import Chime from reolink_aio.exceptions import InvalidParameterError, ReolinkError from homeassistant.components.reolink.const import DOMAIN -from homeassistant.components.reolink.services import ATTR_RINGTONE +from homeassistant.components.reolink.services import ( + ATTR_RINGTONE, + ATTR_TIMESTAMP, + SERVICE_SNAPSHOT_PAST, +) from homeassistant.config_entries import ConfigEntryState -from homeassistant.const import ATTR_DEVICE_ID, Platform +from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID, CONF_FILENAME, Platform from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError, ServiceValidationError from homeassistant.helpers import entity_registry as er +from homeassistant.util import dt as dt_util + +from . import setup_integration +from .conftest import TEST_CAM_NAME from tests.common import MockConfigEntry +TEST_FILE = "/test/snapshot.jpg" +TEST_CAMERA_ID = f"{Platform.CAMERA}.{TEST_CAM_NAME}_fluent" + async def test_play_chime_service_entity( hass: HomeAssistant, @@ -114,3 +126,142 @@ async def test_play_chime_service_unloaded( {ATTR_DEVICE_ID: [device_id], ATTR_RINGTONE: "attraction"}, blocking=True, ) + + +@pytest.fixture +async def camera_config_entry( + hass: HomeAssistant, + config_entry: MockConfigEntry, + reolink_host: MagicMock, +) -> MockConfigEntry: + """Set up the Reolink camera platform.""" + # the camera is in UTC while Home Assistant is in US/Pacific + reolink_host.timezone.return_value = dt_util.UTC + + with patch("homeassistant.components.reolink.PLATFORMS", [Platform.CAMERA]): + await setup_integration(hass, config_entry) + return config_entry + + +@pytest.mark.usefixtures("camera_config_entry") +async def test_snapshot_past_service( + hass: HomeAssistant, + reolink_host: MagicMock, +) -> None: + """Test the snapshot_past service writes the image of the camera to a file.""" + reolink_host.baichuan.snapshot_past = AsyncMock(return_value=b"image") + mopen = mock_open() + + with ( + patch("homeassistant.components.reolink.services.open", mopen, create=True), + patch("homeassistant.components.reolink.services.os.makedirs"), + patch.object(hass.config, "is_allowed_path", return_value=True), + ): + await hass.services.async_call( + DOMAIN, + SERVICE_SNAPSHOT_PAST, + { + ATTR_ENTITY_ID: TEST_CAMERA_ID, + ATTR_TIMESTAMP: "2025-09-29 14:30:00", + CONF_FILENAME: TEST_FILE, + }, + blocking=True, + ) + + # a naive timestamp is interpreted in the Home Assistant timezone (US/Pacific) + # and converted to the timezone of the camera (UTC) + reolink_host.baichuan.snapshot_past.assert_called_once_with( + 0, datetime(2025, 9, 29, 21, 30, tzinfo=dt_util.UTC), "sub", "ffmpeg" + ) + mopen.assert_called_once_with(TEST_FILE, "wb") + assert mopen().write.mock_calls[0][1][0] == b"image" + + +@pytest.mark.usefixtures("camera_config_entry") +async def test_snapshot_past_service_not_allowed_path( + hass: HomeAssistant, + reolink_host: MagicMock, +) -> None: + """Test the snapshot_past service with a path that is not allowed.""" + reolink_host.baichuan.snapshot_past = AsyncMock(return_value=b"image") + + with ( + patch.object(hass.config, "is_allowed_path", return_value=False), + pytest.raises(ServiceValidationError), + ): + await hass.services.async_call( + DOMAIN, + SERVICE_SNAPSHOT_PAST, + { + ATTR_ENTITY_ID: TEST_CAMERA_ID, + ATTR_TIMESTAMP: "2025-09-29 14:30:00", + CONF_FILENAME: TEST_FILE, + }, + blocking=True, + ) + + reolink_host.baichuan.snapshot_past.assert_not_called() + + +@pytest.mark.parametrize( + ("side_effect", "expected"), + [ + (ReolinkError("Test error"), HomeAssistantError), + (InvalidParameterError("Test error"), ServiceValidationError), + ], + ids=["reolink_error", "invalid_parameter"], +) +@pytest.mark.usefixtures("camera_config_entry") +async def test_snapshot_past_service_errors( + hass: HomeAssistant, + reolink_host: MagicMock, + side_effect: Exception, + expected: type[Exception], +) -> None: + """Test the snapshot_past service when the camera returns an error.""" + reolink_host.baichuan.snapshot_past = AsyncMock(side_effect=side_effect) + + with ( + patch.object(hass.config, "is_allowed_path", return_value=True), + pytest.raises(expected), + ): + await hass.services.async_call( + DOMAIN, + SERVICE_SNAPSHOT_PAST, + { + ATTR_ENTITY_ID: TEST_CAMERA_ID, + ATTR_TIMESTAMP: "2025-09-29 14:30:00", + CONF_FILENAME: TEST_FILE, + }, + blocking=True, + ) + + +@pytest.mark.usefixtures("camera_config_entry") +async def test_snapshot_past_service_write_error( + hass: HomeAssistant, + reolink_host: MagicMock, +) -> None: + """Test the snapshot_past service when the image can not be written to disk.""" + reolink_host.baichuan.snapshot_past = AsyncMock(return_value=b"image") + + with ( + patch( + "homeassistant.components.reolink.services.open", + side_effect=OSError("Test error"), + create=True, + ), + patch("homeassistant.components.reolink.services.os.makedirs"), + patch.object(hass.config, "is_allowed_path", return_value=True), + pytest.raises(HomeAssistantError), + ): + await hass.services.async_call( + DOMAIN, + SERVICE_SNAPSHOT_PAST, + { + ATTR_ENTITY_ID: TEST_CAMERA_ID, + ATTR_TIMESTAMP: "2025-09-29 14:30:00", + CONF_FILENAME: TEST_FILE, + }, + blocking=True, + )