Add Reolink snapshot past action (#183812)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

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,
+        )

homeassistant/components/reolink/icons.json

@@ -669,6 +669,9 @@
     },
     "ptz_move": {
       "service": "mdi:pan"
+    },
+    "snapshot_past": {
+      "service": "mdi:camera-timer"
     }
   }
 }

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*"

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,
+    )

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:

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"
     }
   }
 }

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,
+        )