github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-07 10:13:38 +02:00
Code Releases Activity

Add motionEye switches (#52491)

Browse Source
This commit is contained in:
Dermot Duffy 2021-07-22 23:17:39 -07:00 committed by GitHub
parent 1d44bfcfb6
commit dee5d8903c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 388 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
homeassistant/components/motioneye/__init__.py
View File

@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
import json
import logging
from types import MappingProxyType
from typing import Any, Callable
from urllib.parse import urlencode, urljoin
@ -28,6 +29,7 @@ from motioneye_client.const import (
)
from homeassistant.components.camera.const import DOMAIN as CAMERA_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.components.webhook import (
async_generate_id,
async_generate_path,
@ -49,8 +51,13 @@ from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.network import get_url
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from .const import (
ATTR_EVENT_TYPE,
@ -78,7 +85,7 @@ from .const import (
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [CAMERA_DOMAIN]
PLATFORMS = [CAMERA_DOMAIN, SWITCH_DOMAIN]
def create_motioneye_client(
@ -420,3 +427,48 @@ async def handle_webhook(
},
)
return None
class MotionEyeEntity(CoordinatorEntity):
"""Base class for motionEye entities."""
def __init__(
self,
config_entry_id: str,
type_name: str,
camera: dict[str, Any],
client: MotionEyeClient,
coordinator: DataUpdateCoordinator,
options: MappingProxyType[str, Any],
enabled_by_default: bool = True,
) -> None:
"""Initialize a motionEye entity."""
self._camera_id = camera[KEY_ID]
self._device_identifier = get_motioneye_device_identifier(
config_entry_id, self._camera_id
)
self._unique_id = get_motioneye_entity_unique_id(
config_entry_id,
self._camera_id,
type_name,
)
self._client = client
self._camera: dict[str, Any] | None = camera
self._options = options
self._enabled_by_default = enabled_by_default
super().__init__(coordinator)
@property
def entity_registry_enabled_default(self) -> bool:
"""Whether or not the entity is enabled by default."""
return self._enabled_by_default
@property
def unique_id(self) -> str:
"""Return a unique id for this instance."""
return self._unique_id
@property
def device_info(self) -> DeviceInfo:
"""Return the device information."""
return {"identifiers": {self._device_identifier}}

74
homeassistant/components/motioneye/camera.py
View File

@ -2,13 +2,13 @@
from __future__ import annotations
import logging
from typing import Any, Dict, Optional
from types import MappingProxyType
from typing import Any
import aiohttp
from motioneye_client.client import MotionEyeClient
from motioneye_client.const import (
DEFAULT_SURVEILLANCE_USERNAME,
KEY_ID,
KEY_MOTION_DETECTION,
KEY_NAME,
KEY_STREAMING_AUTH_MODE,
@ -30,17 +30,12 @@ from homeassistant.const import (
HTTP_DIGEST_AUTHENTICATION,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import (
MotionEyeEntity,
get_camera_from_cameras,
get_motioneye_device_identifier,
get_motioneye_entity_unique_id,
is_acceptable_camera,
listen_for_new_cameras,
)
@ -79,6 +74,7 @@ async def async_setup_entry(
camera,
entry_data[CONF_CLIENT],
entry_data[CONF_COORDINATOR],
entry.options,
)
]
)
@ -86,7 +82,7 @@ async def async_setup_entry(
listen_for_new_cameras(hass, entry, camera_add)
class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity[Optional[Dict[str, Any]]]):
class MotionEyeMjpegCamera(MotionEyeEntity, MjpegCamera):
"""motionEye mjpeg camera."""
def __init__(
@ -96,25 +92,26 @@ class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity[Optional[Dict[str, Any
password: str,
camera: dict[str, Any],
client: MotionEyeClient,
coordinator: DataUpdateCoordinator[dict[str, Any] | None],
coordinator: DataUpdateCoordinator,
options: MappingProxyType[str, str],
) -> None:
"""Initialize a MJPEG camera."""
self._surveillance_username = username
self._surveillance_password = password
self._client = client
self._camera_id = camera[KEY_ID]
self._device_identifier = get_motioneye_device_identifier(
config_entry_id, self._camera_id
)
self._unique_id = get_motioneye_entity_unique_id(
config_entry_id, self._camera_id, TYPE_MOTIONEYE_MJPEG_CAMERA
)
self._motion_detection_enabled: bool = camera.get(KEY_MOTION_DETECTION, False)
self._available = self._is_acceptable_streaming_camera(camera)
# motionEye cameras are always streaming or unavailable.
self.is_streaming = True
MotionEyeEntity.__init__(
self,
config_entry_id,
TYPE_MOTIONEYE_MJPEG_CAMERA,
camera,
client,
coordinator,
options,
)
MjpegCamera.__init__(
self,
{
@ -122,7 +119,6 @@ class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity[Optional[Dict[str, Any
**self._get_mjpeg_camera_properties_for_camera(camera),
},
)
CoordinatorEntity.__init__(self, coordinator)
@callback
def _get_mjpeg_camera_properties_for_camera(
@ -162,35 +158,26 @@ class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity[Optional[Dict[str, Any
if self._authentication == HTTP_BASIC_AUTHENTICATION:
self._auth = aiohttp.BasicAuth(self._username, password=self._password)
@property
def unique_id(self) -> str:
"""Return a unique id for this instance."""
return self._unique_id
@classmethod
def _is_acceptable_streaming_camera(cls, camera: dict[str, Any] | None) -> bool:
def _is_acceptable_streaming_camera(self) -> bool:
"""Determine if a camera is streaming/usable."""
return is_acceptable_camera(camera) and MotionEyeClient.is_camera_streaming(
camera
)
return is_acceptable_camera(
self._camera
) and MotionEyeClient.is_camera_streaming(self._camera)
@property
def available(self) -> bool:
"""Return if entity is available."""
return self._available
return super().available and self._is_acceptable_streaming_camera()
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
available = False
if self.coordinator.last_update_success:
camera = get_camera_from_cameras(self._camera_id, self.coordinator.data)
if self._is_acceptable_streaming_camera(camera):
assert camera
self._set_mjpeg_camera_state_for_camera(camera)
self._motion_detection_enabled = camera.get(KEY_MOTION_DETECTION, False)
available = True
self._available = available
self._camera = get_camera_from_cameras(self._camera_id, self.coordinator.data)
if self._camera and self._is_acceptable_streaming_camera():
self._set_mjpeg_camera_state_for_camera(self._camera)
self._motion_detection_enabled = self._camera.get(
KEY_MOTION_DETECTION, False
)
super()._handle_coordinator_update()
@property
@ -202,8 +189,3 @@ class MotionEyeMjpegCamera(MjpegCamera, CoordinatorEntity[Optional[Dict[str, Any
def motion_detection_enabled(self) -> bool:
"""Return the camera motion detection status."""
return self._motion_detection_enabled
@property
def device_info(self) -> DeviceInfo:
"""Return the device information."""
return {"identifiers": {self._device_identifier}}

1
homeassistant/components/motioneye/const.py
View File

@ -84,6 +84,7 @@ SIGNAL_CAMERA_ADD: Final = f"{DOMAIN}_camera_add_signal." "{}"
SIGNAL_CAMERA_REMOVE: Final = f"{DOMAIN}_camera_remove_signal." "{}"
TYPE_MOTIONEYE_MJPEG_CAMERA: Final = "motioneye_mjpeg_camera"
TYPE_MOTIONEYE_SWITCH_BASE: Final = f"{DOMAIN}_switch"
WEB_HOOK_SENTINEL_KEY: Final = "src"
WEB_HOOK_SENTINEL_VALUE: Final = "hass-motioneye"

120
homeassistant/components/motioneye/switch.py Normal file
View File

@ -0,0 +1,120 @@
"""Switch platform for motionEye."""
from __future__ import annotations
from types import MappingProxyType
from typing import Any, Callable
from motioneye_client.client import MotionEyeClient
from motioneye_client.const import (
KEY_MOTION_DETECTION,
KEY_MOVIES,
KEY_NAME,
KEY_STILL_IMAGES,
KEY_TEXT_OVERLAY,
KEY_UPLOAD_ENABLED,
KEY_VIDEO_STREAMING,
)
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import MotionEyeEntity, listen_for_new_cameras
from .const import CONF_CLIENT, CONF_COORDINATOR, DOMAIN, TYPE_MOTIONEYE_SWITCH_BASE
MOTIONEYE_SWITCHES = [
(KEY_MOTION_DETECTION, "Motion Detection", True),
(KEY_TEXT_OVERLAY, "Text Overlay", False),
(KEY_VIDEO_STREAMING, "Video Streaming", False),
(KEY_STILL_IMAGES, "Still Images", True),
(KEY_MOVIES, "Movies", True),
(KEY_UPLOAD_ENABLED, "Upload Enabled", False),
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable
) -> bool:
"""Set up motionEye from a config entry."""
entry_data = hass.data[DOMAIN][entry.entry_id]
@callback
def camera_add(camera: dict[str, Any]) -> None:
"""Add a new motionEye camera."""
async_add_entities(
[
MotionEyeSwitch(
entry.entry_id,
camera,
switch_key,
switch_key_friendly_name,
entry_data[CONF_CLIENT],
entry_data[CONF_COORDINATOR],
entry.options,
enabled,
)
for switch_key, switch_key_friendly_name, enabled in MOTIONEYE_SWITCHES
]
)
listen_for_new_cameras(hass, entry, camera_add)
return True
class MotionEyeSwitch(MotionEyeEntity, SwitchEntity):
"""MotionEyeSwitch switch class."""
def __init__(
self,
config_entry_id: str,
camera: dict[str, Any],
switch_key: str,
switch_key_friendly_name: str,
client: MotionEyeClient,
coordinator: DataUpdateCoordinator,
options: MappingProxyType[str, str],
enabled_by_default: bool,
) -> None:
"""Initialize the switch."""
self._switch_key = switch_key
self._switch_key_friendly_name = switch_key_friendly_name
MotionEyeEntity.__init__(
self,
config_entry_id,
f"{TYPE_MOTIONEYE_SWITCH_BASE}_{switch_key}",
camera,
client,
coordinator,
options,
enabled_by_default,
)
@property
def name(self) -> str:
"""Return the name of the switch."""
camera_name = self._camera[KEY_NAME] if self._camera else ""
return f"{camera_name} {self._switch_key_friendly_name}"
@property
def is_on(self) -> bool:
"""Return true if the switch is on."""
return bool(self._camera and self._camera.get(self._switch_key, False))
async def _async_send_set_camera(self, value: bool) -> None:
"""Set a switch value."""
# 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 camera:
camera[self._switch_key] = value
await self._client.async_set_camera(self._camera_id, camera)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
await self._async_send_set_camera(True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
await self._async_send_set_camera(False)

4
tests/components/motioneye/__init__.py
View File

@ -133,6 +133,10 @@ TEST_CAMERA = {
}
TEST_CAMERAS = {"cameras": [TEST_CAMERA]}
TEST_SURVEILLANCE_USERNAME = "surveillance_username"
TEST_SWITCH_ENTITY_ID_BASE = "switch.test_camera"
TEST_SWITCH_MOTION_DETECTION_ENTITY_ID = (
f"{TEST_SWITCH_ENTITY_ID_BASE}_motion_detection"
)
def create_mock_motioneye_client() -> AsyncMock:

3
tests/components/motioneye/test_camera.py
View File

@ -298,8 +298,7 @@ async def test_state_attributes(hass: HomeAssistant) -> None:
async def test_device_info(hass: HomeAssistant) -> None:
"""Verify device information includes expected details."""
client = create_mock_motioneye_client()
entry = await setup_mock_motioneye_config_entry(hass, client=client)
entry = await setup_mock_motioneye_config_entry(hass)
device_identifier = get_motioneye_device_identifier(entry.entry_id, TEST_CAMERA_ID)
device_registry = dr.async_get(hass)

180
tests/components/motioneye/test_switch.py Normal file
View File

@ -0,0 +1,180 @@
"""Tests for the motionEye switch platform."""
import copy
from datetime import timedelta
from unittest.mock import AsyncMock, call, patch
from motioneye_client.const import (
KEY_MOTION_DETECTION,
KEY_MOVIES,
KEY_STILL_IMAGES,
KEY_TEXT_OVERLAY,
KEY_UPLOAD_ENABLED,
KEY_VIDEO_STREAMING,
)
from homeassistant.components.motioneye import get_motioneye_device_identifier
from homeassistant.components.motioneye.const import DEFAULT_SCAN_INTERVAL
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import RELOAD_AFTER_UPDATE_DELAY
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, SERVICE_TURN_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.util.dt as dt_util
from . import (
TEST_CAMERA,
TEST_CAMERA_ID,
TEST_SWITCH_ENTITY_ID_BASE,
TEST_SWITCH_MOTION_DETECTION_ENTITY_ID,
create_mock_motioneye_client,
setup_mock_motioneye_config_entry,
)
from tests.common import async_fire_time_changed
async def test_switch_turn_on_off(hass: HomeAssistant) -> None:
"""Test turning the switch on and off."""
client = create_mock_motioneye_client()
await setup_mock_motioneye_config_entry(hass, client=client)
# Verify switch is on (as per TEST_COMPONENTS above).
entity_state = hass.states.get(TEST_SWITCH_MOTION_DETECTION_ENTITY_ID)
assert entity_state
assert entity_state.state == "on"
client.async_get_camera = AsyncMock(return_value=TEST_CAMERA)
expected_camera = copy.deepcopy(TEST_CAMERA)
expected_camera[KEY_MOTION_DETECTION] = False
# When the next refresh is called return the updated values.
client.async_get_cameras = AsyncMock(return_value={"cameras": [expected_camera]})
# Turn switch off.
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: TEST_SWITCH_MOTION_DETECTION_ENTITY_ID},
blocking=True,
)
async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
await hass.async_block_till_done()
# Verify correct parameters are passed to the library.
assert client.async_set_camera.call_args == call(TEST_CAMERA_ID, expected_camera)
# Verify the switch turns off.
entity_state = hass.states.get(TEST_SWITCH_MOTION_DETECTION_ENTITY_ID)
assert entity_state
assert entity_state.state == "off"
# When the next refresh is called return the updated values.
client.async_get_cameras = AsyncMock(return_value={"cameras": [TEST_CAMERA]})
# Turn switch on.
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: TEST_SWITCH_MOTION_DETECTION_ENTITY_ID},
blocking=True,
)
# Verify correct parameters are passed to the library.
assert client.async_set_camera.call_args == call(TEST_CAMERA_ID, TEST_CAMERA)
async_fire_time_changed(hass, dt_util.utcnow() + DEFAULT_SCAN_INTERVAL)
await hass.async_block_till_done()
# Verify the switch turns on.
entity_state = hass.states.get(TEST_SWITCH_MOTION_DETECTION_ENTITY_ID)
assert entity_state
assert entity_state.state == "on"
async def test_switch_has_correct_entities(hass: HomeAssistant) -> None:
"""Test that the correct switch entities are created."""
client = create_mock_motioneye_client()
await setup_mock_motioneye_config_entry(hass, client=client)
enabled_switch_keys = [
KEY_MOTION_DETECTION,
KEY_STILL_IMAGES,
KEY_MOVIES,
]
disabled_switch_keys = [
KEY_TEXT_OVERLAY,
KEY_UPLOAD_ENABLED,
KEY_VIDEO_STREAMING,
]
for switch_key in enabled_switch_keys:
entity_id = f"{TEST_SWITCH_ENTITY_ID_BASE}_{switch_key}"
entity_state = hass.states.get(entity_id)
assert entity_state, f"Couldn't find entity: {entity_id}"
for switch_key in disabled_switch_keys:
entity_id = f"{TEST_SWITCH_ENTITY_ID_BASE}_{switch_key}"
entity_state = hass.states.get(entity_id)
assert not entity_state
async def test_disabled_switches_can_be_enabled(hass: HomeAssistant) -> None:
"""Verify disabled switches can be enabled."""
client = create_mock_motioneye_client()
await setup_mock_motioneye_config_entry(hass, client=client)
disabled_switch_keys = [
KEY_TEXT_OVERLAY,
KEY_UPLOAD_ENABLED,
]
for switch_key in disabled_switch_keys:
entity_id = f"{TEST_SWITCH_ENTITY_ID_BASE}_{switch_key}"
entity_registry = er.async_get(hass)
entry = entity_registry.async_get(entity_id)
assert entry
assert entry.disabled
assert entry.disabled_by == er.DISABLED_INTEGRATION
entity_state = hass.states.get(entity_id)
assert not entity_state
with patch(
"homeassistant.components.motioneye.MotionEyeClient",
return_value=client,
):
updated_entry = entity_registry.async_update_entity(
entity_id, disabled_by=None
)
assert not updated_entry.disabled
await hass.async_block_till_done()
async_fire_time_changed(
hass,
dt_util.utcnow() + timedelta(seconds=RELOAD_AFTER_UPDATE_DELAY + 1),
)
await hass.async_block_till_done()
entity_state = hass.states.get(entity_id)
assert entity_state
async def test_switch_device_info(hass: HomeAssistant) -> None:
"""Verify device information includes expected details."""
config_entry = await setup_mock_motioneye_config_entry(hass)
device_identifer = get_motioneye_device_identifier(
config_entry.entry_id, TEST_CAMERA_ID
)
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({device_identifer})
assert device
entity_registry = await er.async_get_registry(hass)
entities_from_device = [
entry.entity_id
for entry in er.async_entries_for_device(entity_registry, device.id)
]
assert TEST_SWITCH_MOTION_DETECTION_ENTITY_ID in entities_from_device