Add motionEye services (#53411)

homeassistant/components/motioneye/camera.py

@@ -9,10 +9,20 @@ from jinja2 import Template
 from motioneye_client.client import MotionEyeClient, MotionEyeClientURLParseError
 from motioneye_client.const import (
     DEFAULT_SURVEILLANCE_USERNAME,
+    KEY_ACTION_SNAPSHOT,
     KEY_MOTION_DETECTION,
     KEY_NAME,
     KEY_STREAMING_AUTH_MODE,
+    KEY_TEXT_OVERLAY_CAMERA_NAME,
+    KEY_TEXT_OVERLAY_CUSTOM_TEXT,
+    KEY_TEXT_OVERLAY_CUSTOM_TEXT_LEFT,
+    KEY_TEXT_OVERLAY_CUSTOM_TEXT_RIGHT,
+    KEY_TEXT_OVERLAY_DISABLED,
+    KEY_TEXT_OVERLAY_LEFT,
+    KEY_TEXT_OVERLAY_RIGHT,
+    KEY_TEXT_OVERLAY_TIMESTAMP,
 )
+import voluptuous as vol
 
 from homeassistant.components.mjpeg.camera import (
     CONF_MJPEG_URL,
@@ -30,6 +40,7 @@ from homeassistant.const import (
     HTTP_DIGEST_AUTHENTICATION,
 )
 from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import config_validation as cv, entity_platform
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
@@ -40,6 +51,7 @@ from . import (
     listen_for_new_cameras,
 )
 from .const import (
+    CONF_ACTION,
     CONF_CLIENT,
     CONF_COORDINATOR,
     CONF_STREAM_URL_TEMPLATE,
@@ -47,11 +59,40 @@ from .const import (
     CONF_SURVEILLANCE_USERNAME,
     DOMAIN,
     MOTIONEYE_MANUFACTURER,
+    SERVICE_ACTION,
+    SERVICE_SET_TEXT_OVERLAY,
+    SERVICE_SNAPSHOT,
     TYPE_MOTIONEYE_MJPEG_CAMERA,
 )
 
 PLATFORMS = ["camera"]
 
+SCHEMA_TEXT_OVERLAY = vol.In(
+    [
+        KEY_TEXT_OVERLAY_DISABLED,
+        KEY_TEXT_OVERLAY_TIMESTAMP,
+        KEY_TEXT_OVERLAY_CUSTOM_TEXT,
+        KEY_TEXT_OVERLAY_CAMERA_NAME,
+    ]
+)
+SCHEMA_SERVICE_SET_TEXT = vol.Schema(
+    vol.All(
+        {
+            vol.Optional(KEY_TEXT_OVERLAY_LEFT): SCHEMA_TEXT_OVERLAY,
+            vol.Optional(KEY_TEXT_OVERLAY_CUSTOM_TEXT_LEFT): cv.string,
+            vol.Optional(KEY_TEXT_OVERLAY_RIGHT): SCHEMA_TEXT_OVERLAY,
+            vol.Optional(KEY_TEXT_OVERLAY_CUSTOM_TEXT_RIGHT): cv.string,
+        },
+        cv.has_at_least_one_key(
+            KEY_TEXT_OVERLAY_LEFT,
+            KEY_TEXT_OVERLAY_CUSTOM_TEXT_LEFT,
+            KEY_TEXT_OVERLAY_RIGHT,
+            KEY_TEXT_OVERLAY_CUSTOM_TEXT_RIGHT,
+        ),
+    ),
+    extra=vol.ALLOW_EXTRA,
+)
+
 
 async def async_setup_entry(
     hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
@@ -80,6 +121,23 @@ async def async_setup_entry(
 
     listen_for_new_cameras(hass, entry, camera_add)
 
+    platform = entity_platform.async_get_current_platform()
+    platform.async_register_entity_service(
+        SERVICE_SET_TEXT_OVERLAY,
+        SCHEMA_SERVICE_SET_TEXT,
+        "async_set_text_overlay",
+    )
+    platform.async_register_entity_service(
+        SERVICE_ACTION,
+        {vol.Required(CONF_ACTION): cv.string},
+        "async_request_action",
+    )
+    platform.async_register_entity_service(
+        SERVICE_SNAPSHOT,
+        {},
+        "async_request_snapshot",
+    )
+
 
 class MotionEyeMjpegCamera(MotionEyeEntity, MjpegCamera):
     """motionEye mjpeg camera."""
@@ -201,3 +259,38 @@ class MotionEyeMjpegCamera(MotionEyeEntity, MjpegCamera):
     def motion_detection_enabled(self) -> bool:
         """Return the camera motion detection status."""
         return self._motion_detection_enabled
+
+    async def async_set_text_overlay(
+        self,
+        left_text: str = None,
+        right_text: str = None,
+        custom_left_text: str = None,
+        custom_right_text: str = None,
+    ) -> None:
+        """Set text overlay for a camera."""
+        # Fetch the very latest camera config to reduce the risk of updating with a
+        # stale configuration.
+        camera = await self._client.async_get_camera(self._camera_id)
+        if not camera:
+            return
+        if left_text is not None:
+            camera[KEY_TEXT_OVERLAY_LEFT] = left_text
+        if right_text is not None:
+            camera[KEY_TEXT_OVERLAY_RIGHT] = right_text
+        if custom_left_text is not None:
+            camera[KEY_TEXT_OVERLAY_CUSTOM_TEXT_LEFT] = custom_left_text.encode(
+                "unicode_escape"
+            ).decode("UTF-8")
+        if custom_right_text is not None:
+            camera[KEY_TEXT_OVERLAY_CUSTOM_TEXT_RIGHT] = custom_right_text.encode(
+                "unicode_escape"
+            ).decode("UTF-8")
+        await self._client.async_set_camera(self._camera_id, camera)
+
+    async def async_request_action(self, action: str) -> None:
+        """Call a motionEye action on a camera."""
+        await self._client.async_action(self._camera_id, action)
+
+    async def async_request_snapshot(self) -> None:
+        """Request a motionEye snapshot be saved."""
+        await self.async_request_action(KEY_ACTION_SNAPSHOT)

homeassistant/components/motioneye/const.py

@@ -28,6 +28,7 @@ DOMAIN: Final = "motioneye"
 ATTR_EVENT_TYPE: Final = "event_type"
 ATTR_WEBHOOK_ID: Final = "webhook_id"
 
+CONF_ACTION: Final = "action"
 CONF_CLIENT: Final = "client"
 CONF_COORDINATOR: Final = "coordinator"
 CONF_ADMIN_PASSWORD: Final = "admin_password"
@@ -81,6 +82,10 @@ EVENT_FILE_STORED_KEYS: Final = [
 
 MOTIONEYE_MANUFACTURER: Final = "motionEye"
 
+SERVICE_SET_TEXT_OVERLAY: Final = "set_text_overlay"
+SERVICE_ACTION: Final = "action"
+SERVICE_SNAPSHOT: Final = "snapshot"
+
 SIGNAL_CAMERA_ADD: Final = f"{DOMAIN}_camera_add_signal." "{}"
 SIGNAL_CAMERA_REMOVE: Final = f"{DOMAIN}_camera_remove_signal." "{}"
 

homeassistant/components/motioneye/services.yaml

+set_text_overlay:
+  name: Set Text Overlay
+  description: Sets the text overlay for a camera.
+  target:
+    device:
+      integration: motioneye
+    entity:
+      integration: motioneye
+  fields:
+    left_text:
+      name: Left Text Overlay
+      description: Text to display on the left
+      required: false
+      advanced: false
+      example: "timestamp"
+      default: ""
+      selector:
+        select:
+          options:
+            - "disabled"
+            - "camera-name"
+            - "timestamp"
+            - "custom-text"
+    custom_left_text:
+      name: Left Custom Text
+      description: Custom text to display on the left
+      required: false
+      advanced: false
+      example: "Hello on the left!"
+      default: ""
+      selector:
+        text:
+          multiline: true
+    right_text:
+      name: Right Text Overlay
+      description: Text to display on the right
+      required: false
+      advanced: false
+      example: "timestamp"
+      default: ""
+      selector:
+        select:
+          options:
+            - "disabled"
+            - "camera-name"
+            - "timestamp"
+            - "custom-text"
+    custom_right_text:
+      name: Right Custom Text
+      description: Custom text to display on the right
+      required: false
+      advanced: false
+      example: "Hello on the right!"
+      default: ""
+      selector:
+        text:
+          multiline: true
+
+action:
+  name: Action
+  description: Trigger a motionEye action
+  target:
+    device:
+      integration: motioneye
+    entity:
+      integration: motioneye
+  fields:
+    action:
+      name: Action
+      description: Action to trigger
+      required: true
+      advanced: false
+      example: "snapshot"
+      default: ""
+      selector:
+        select:
+          options:
+            - "snapshot"
+            - "record_start"
+            - "record_stop"
+            - "lock"
+            - "unlock"
+            - "light_on"
+            - "light_off"
+            - "alarm_on"
+            - "alarm_off"
+            - "up"
+            - "right"
+            - "down"
+            - "left"
+            - "zoom_in"
+            - "zoom_out"
+            - "preset1"
+            - "preset2"
+            - "preset3"
+            - "preset4"
+            - "preset5"
+            - "preset6"
+            - "preset7"
+            - "preset8"
+            - "preset9"
+
+snapshot:
+  name: Snapshot
+  description: Trigger a motionEye still snapshot
+  target:
+    device:
+      integration: motioneye
+    entity:
+      integration: motioneye

tests/components/motioneye/test_camera.py

 """Test the motionEye camera."""
 import copy
 from typing import Any, cast
-from unittest.mock import AsyncMock, Mock
+from unittest.mock import AsyncMock, Mock, call
 
 from aiohttp import web
 from aiohttp.web_exceptions import HTTPBadGateway
@@ -14,20 +14,31 @@ from motioneye_client.const import (
     KEY_CAMERAS,
     KEY_MOTION_DETECTION,
     KEY_NAME,
+    KEY_TEXT_OVERLAY_CUSTOM_TEXT,
+    KEY_TEXT_OVERLAY_CUSTOM_TEXT_LEFT,
+    KEY_TEXT_OVERLAY_CUSTOM_TEXT_RIGHT,
+    KEY_TEXT_OVERLAY_LEFT,
+    KEY_TEXT_OVERLAY_RIGHT,
+    KEY_TEXT_OVERLAY_TIMESTAMP,
     KEY_VIDEO_STREAMING,
 )
 import pytest
+import voluptuous as vol
 
 from homeassistant.components.camera import async_get_image, async_get_mjpeg_stream
 from homeassistant.components.motioneye import get_motioneye_device_identifier
 from homeassistant.components.motioneye.const import (
+    CONF_ACTION,
     CONF_STREAM_URL_TEMPLATE,
     CONF_SURVEILLANCE_USERNAME,
     DEFAULT_SCAN_INTERVAL,
     DOMAIN,
     MOTIONEYE_MANUFACTURER,
+    SERVICE_ACTION,
+    SERVICE_SET_TEXT_OVERLAY,
+    SERVICE_SNAPSHOT,
 )
-from homeassistant.const import CONF_URL
+from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID, CONF_URL
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import device_registry as dr, entity_registry as er
@@ -35,6 +46,7 @@ from homeassistant.helpers.device_registry import async_get_registry
 import homeassistant.util.dt as dt_util
 
 from . import (
+    TEST_CAMERA,
     TEST_CAMERA_DEVICE_IDENTIFIER,
     TEST_CAMERA_ENTITY_ID,
     TEST_CAMERA_ID,
@@ -379,3 +391,155 @@ async def test_get_stream_from_camera_with_broken_host(
     await hass.async_block_till_done()
     with pytest.raises(HTTPBadGateway):
         await async_get_mjpeg_stream(hass, Mock(), TEST_CAMERA_ENTITY_ID)
+
+
+async def test_set_text_overlay_bad_extra_key(hass: HomeAssistant) -> None:
+    """Test text overlay with incorrect input data."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {ATTR_ENTITY_ID: TEST_CAMERA_ENTITY_ID, "extra_key": "foo"}
+    with pytest.raises(vol.error.MultipleInvalid):
+        await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, data)
+
+
+async def test_set_text_overlay_bad_entity_identifier(hass: HomeAssistant) -> None:
+    """Test text overlay with bad entity identifier."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {
+        ATTR_ENTITY_ID: "some random string",
+        KEY_TEXT_OVERLAY_LEFT: KEY_TEXT_OVERLAY_TIMESTAMP,
+    }
+
+    client.reset_mock()
+    await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, data)
+    await hass.async_block_till_done()
+    assert not client.async_set_camera.called
+
+
+async def test_set_text_overlay_bad_empty(hass: HomeAssistant) -> None:
+    """Test text overlay with incorrect input data."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+    with pytest.raises(vol.error.MultipleInvalid):
+        await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, {})
+        await hass.async_block_till_done()
+
+
+async def test_set_text_overlay_bad_no_left_or_right(hass: HomeAssistant) -> None:
+    """Test text overlay with incorrect input data."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {ATTR_ENTITY_ID: TEST_CAMERA_ENTITY_ID}
+    with pytest.raises(vol.error.MultipleInvalid):
+        await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, data)
+        await hass.async_block_till_done()
+
+
+async def test_set_text_overlay_good(hass: HomeAssistant) -> None:
+    """Test a working text overlay."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    custom_left_text = "one\ntwo\nthree"
+    custom_right_text = "four\nfive\nsix"
+    data = {
+        ATTR_ENTITY_ID: TEST_CAMERA_ENTITY_ID,
+        KEY_TEXT_OVERLAY_LEFT: KEY_TEXT_OVERLAY_CUSTOM_TEXT,
+        KEY_TEXT_OVERLAY_RIGHT: KEY_TEXT_OVERLAY_CUSTOM_TEXT,
+        KEY_TEXT_OVERLAY_CUSTOM_TEXT_LEFT: custom_left_text,
+        KEY_TEXT_OVERLAY_CUSTOM_TEXT_RIGHT: custom_right_text,
+    }
+    client.async_get_camera = AsyncMock(return_value=copy.deepcopy(TEST_CAMERA))
+
+    await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, data)
+    await hass.async_block_till_done()
+    assert client.async_get_camera.called
+
+    expected_camera = copy.deepcopy(TEST_CAMERA)
+    expected_camera[KEY_TEXT_OVERLAY_LEFT] = KEY_TEXT_OVERLAY_CUSTOM_TEXT
+    expected_camera[KEY_TEXT_OVERLAY_RIGHT] = KEY_TEXT_OVERLAY_CUSTOM_TEXT
+    expected_camera[KEY_TEXT_OVERLAY_CUSTOM_TEXT_LEFT] = "one\\ntwo\\nthree"
+    expected_camera[KEY_TEXT_OVERLAY_CUSTOM_TEXT_RIGHT] = "four\\nfive\\nsix"
+    assert client.async_set_camera.call_args == call(TEST_CAMERA_ID, expected_camera)
+
+
+async def test_set_text_overlay_good_entity_id(hass: HomeAssistant) -> None:
+    """Test a working text overlay with entity_id."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {
+        ATTR_ENTITY_ID: TEST_CAMERA_ENTITY_ID,
+        KEY_TEXT_OVERLAY_LEFT: KEY_TEXT_OVERLAY_TIMESTAMP,
+    }
+    client.async_get_camera = AsyncMock(return_value=copy.deepcopy(TEST_CAMERA))
+    await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, data)
+    await hass.async_block_till_done()
+    assert client.async_get_camera.called
+
+    expected_camera = copy.deepcopy(TEST_CAMERA)
+    expected_camera[KEY_TEXT_OVERLAY_LEFT] = KEY_TEXT_OVERLAY_TIMESTAMP
+    assert client.async_set_camera.call_args == call(TEST_CAMERA_ID, expected_camera)
+
+
+async def test_set_text_overlay_bad_device(hass: HomeAssistant) -> None:
+    """Test a working text overlay."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {
+        ATTR_DEVICE_ID: "not a device",
+        KEY_TEXT_OVERLAY_LEFT: KEY_TEXT_OVERLAY_TIMESTAMP,
+    }
+    client.reset_mock()
+    client.async_get_camera = AsyncMock(return_value=copy.deepcopy(TEST_CAMERA))
+    await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, data)
+    await hass.async_block_till_done()
+    assert not client.async_get_camera.called
+    assert not client.async_set_camera.called
+
+
+async def test_set_text_overlay_no_such_camera(hass: HomeAssistant) -> None:
+    """Test a working text overlay."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {
+        ATTR_ENTITY_ID: TEST_CAMERA_ENTITY_ID,
+        KEY_TEXT_OVERLAY_LEFT: KEY_TEXT_OVERLAY_TIMESTAMP,
+    }
+    client.reset_mock()
+    client.async_get_camera = AsyncMock(return_value={})
+    await hass.services.async_call(DOMAIN, SERVICE_SET_TEXT_OVERLAY, data)
+    await hass.async_block_till_done()
+    assert not client.async_set_camera.called
+
+
+async def test_request_action(hass: HomeAssistant) -> None:
+    """Test requesting an action."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {
+        ATTR_ENTITY_ID: TEST_CAMERA_ENTITY_ID,
+        CONF_ACTION: "foo",
+    }
+    await hass.services.async_call(DOMAIN, SERVICE_ACTION, data)
+    await hass.async_block_till_done()
+    assert client.async_action.call_args == call(TEST_CAMERA_ID, data[CONF_ACTION])
+
+
+async def test_request_snapshot(hass: HomeAssistant) -> None:
+    """Test requesting a snapshot."""
+    client = create_mock_motioneye_client()
+    await setup_mock_motioneye_config_entry(hass, client=client)
+
+    data = {ATTR_ENTITY_ID: TEST_CAMERA_ENTITY_ID}
+
+    await hass.services.async_call(DOMAIN, SERVICE_SNAPSHOT, data)
+    await hass.async_block_till_done()
+    assert client.async_action.call_args == call(TEST_CAMERA_ID, "snapshot")