diff --git a/homeassistant/components/motioneye/camera.py b/homeassistant/components/motioneye/camera.py
index f21769b24f21f929dfe8b289976600db7a846776..428e2f31c81b7c302c46b17278fab65b6528e965 100644
--- a/homeassistant/components/motioneye/camera.py
+++ b/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)
diff --git a/homeassistant/components/motioneye/const.py b/homeassistant/components/motioneye/const.py
index 4e30cfb85140ac512814b3bd941fc254ea34ddca..1dbb78f1e03f9d23d262219f1c2b15d17fa9ed51 100644
--- a/homeassistant/components/motioneye/const.py
+++ b/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." "{}"
diff --git a/homeassistant/components/motioneye/services.yaml b/homeassistant/components/motioneye/services.yaml
new file mode 100644
index 0000000000000000000000000000000000000000..2970124c00065263fe04add9cdbb21067e838f77
--- /dev/null
+++ b/homeassistant/components/motioneye/services.yaml
@@ -0,0 +1,110 @@
+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
diff --git a/tests/components/motioneye/test_camera.py b/tests/components/motioneye/test_camera.py
index 6a669adc65f4a096bc125ffdb6475bea4ac6dd5d..b3d192371656052383ad99db8baf7e6d33df1bc0 100644
--- a/tests/components/motioneye/test_camera.py
+++ b/tests/components/motioneye/test_camera.py
@@ -1,7 +1,7 @@
"""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")