mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 10:13:52 -05:00
Add camera platform to Shelly integration (#179072)
This commit is contained in:
Maciej Bieniek
@@ -83,6 +83,7 @@ from .utils import (
|
||||
PLATFORMS: Final = [
|
||||
Platform.BINARY_SENSOR,
|
||||
Platform.BUTTON,
|
||||
Platform.CAMERA,
|
||||
Platform.CLIMATE,
|
||||
Platform.COVER,
|
||||
Platform.EVENT,
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
"""Support for Shelly cameras."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Final, override
|
||||
from urllib.parse import quote
|
||||
|
||||
from homeassistant.components.camera import (
|
||||
Camera,
|
||||
CameraEntityDescription,
|
||||
CameraEntityFeature,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from .coordinator import ShellyConfigEntry, ShellyRpcCoordinator
|
||||
from .entity import (
|
||||
RpcEntityDescription,
|
||||
ShellyRpcAttributeEntity,
|
||||
async_setup_entry_rpc,
|
||||
)
|
||||
from .utils import get_host
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class RpcCameraEntityDescription(RpcEntityDescription, CameraEntityDescription):
|
||||
"""Class to describe a Shelly RPC camera entity."""
|
||||
|
||||
stream: int
|
||||
|
||||
|
||||
RPC_CAMERA_ENTITIES: Final = {
|
||||
"stream_0": RpcCameraEntityDescription(
|
||||
key="camera",
|
||||
stream=0,
|
||||
translation_key="stream",
|
||||
translation_placeholders={"stream_id": "0"},
|
||||
),
|
||||
"stream_1": RpcCameraEntityDescription(
|
||||
key="camera",
|
||||
stream=1,
|
||||
translation_key="stream",
|
||||
translation_placeholders={"stream_id": "1"},
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ShellyConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Shelly camera entities."""
|
||||
if not config_entry.runtime_data.rpc:
|
||||
return
|
||||
|
||||
async_setup_entry_rpc(
|
||||
hass,
|
||||
config_entry,
|
||||
async_add_entities,
|
||||
RPC_CAMERA_ENTITIES,
|
||||
ShellyCameraEntity,
|
||||
)
|
||||
|
||||
|
||||
class ShellyCameraEntity(ShellyRpcAttributeEntity, Camera):
|
||||
"""Shelly camera entity for RPC devices."""
|
||||
|
||||
_attr_brand = "Shelly"
|
||||
_attr_supported_features = CameraEntityFeature.STREAM
|
||||
entity_description: RpcCameraEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: ShellyRpcCoordinator,
|
||||
key: str,
|
||||
attribute: str,
|
||||
description: RpcCameraEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize Shelly camera entity."""
|
||||
super().__init__(coordinator, key, attribute, description)
|
||||
Camera.__init__(self)
|
||||
|
||||
self._attr_model = self.coordinator.model
|
||||
|
||||
@override
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Available."""
|
||||
available = super().available
|
||||
if not available:
|
||||
return False
|
||||
|
||||
config = self.coordinator.device.config[self.key]
|
||||
return not self.status["privacy"] and config["rtsp"]["enable"]
|
||||
|
||||
@override
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
"""Return True if the camera is running."""
|
||||
return (
|
||||
self.coordinator.device.initialized and self.status["streamer"] == "running"
|
||||
)
|
||||
|
||||
@override
|
||||
@property
|
||||
def is_recording(self) -> bool:
|
||||
"""Return True if the camera is currently recording."""
|
||||
return bool(self.status.get("recordings"))
|
||||
|
||||
@override
|
||||
@property
|
||||
def is_streaming(self) -> bool:
|
||||
"""Return True if the camera is currently streaming."""
|
||||
return bool(self.status["streams"] > 0)
|
||||
|
||||
@override
|
||||
async def stream_source(self) -> str | None:
|
||||
"""Return the RTSP stream source for go2rtc."""
|
||||
username = self.coordinator.config_entry.data.get(CONF_USERNAME)
|
||||
password = self.coordinator.config_entry.data.get(CONF_PASSWORD)
|
||||
host = get_host(self.coordinator.config_entry.data[CONF_HOST])
|
||||
|
||||
if username and password:
|
||||
return (
|
||||
f"rtsp://{quote(username, safe='')}:{quote(password, safe='')}@{host}"
|
||||
f"/stream/{self.entity_description.stream}"
|
||||
)
|
||||
|
||||
return f"rtsp://{host}/stream/{self.entity_description.stream}"
|
||||
|
||||
@override
|
||||
@property
|
||||
def use_stream_for_stills(self) -> bool:
|
||||
"""Use the RTSP stream to generate still images."""
|
||||
return True
|
||||
@@ -263,6 +263,11 @@
|
||||
"name": "Unmute alarm"
|
||||
}
|
||||
},
|
||||
"camera": {
|
||||
"stream": {
|
||||
"name": "Stream {stream_id}"
|
||||
}
|
||||
},
|
||||
"climate": {
|
||||
"thermostat": {
|
||||
"state_attributes": {
|
||||
|
||||
@@ -466,6 +466,25 @@ MOCK_STATUS_RPC = {
|
||||
"wifi": {"rssi": -63},
|
||||
}
|
||||
|
||||
MOCK_CAMERA_CONFIG = {
|
||||
"camera:0": {
|
||||
"id": 0,
|
||||
"rtsp": {"enable": True},
|
||||
}
|
||||
}
|
||||
|
||||
MOCK_CAMERA_STATUS = {
|
||||
"camera:0": {
|
||||
"id": 0,
|
||||
"privacy": False,
|
||||
"arm": True,
|
||||
"streamer": "running",
|
||||
"motion": False,
|
||||
"streams": 0,
|
||||
"recordings": None,
|
||||
}
|
||||
}
|
||||
|
||||
MOCK_SCRIPTS = [
|
||||
""""
|
||||
function eventHandler(event, userdata) {
|
||||
@@ -821,3 +840,14 @@ def disable_async_remove_shelly_rpc_entities() -> Generator[None]:
|
||||
"homeassistant.components.shelly.utils.async_remove_shelly_rpc_entities"
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_camera_rpc_device(
|
||||
monkeypatch: pytest.MonkeyPatch, mock_rpc_device: Mock
|
||||
) -> Mock:
|
||||
"""Set up mock RPC device with camera component data."""
|
||||
monkeypatch.setattr(mock_rpc_device, "config", MOCK_CAMERA_CONFIG)
|
||||
monkeypatch.setattr(mock_rpc_device, "status", MOCK_CAMERA_STATUS)
|
||||
|
||||
return mock_rpc_device
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
# serializer version: 1
|
||||
# name: test_camera_entity_setup[camera.test_name_stream_0-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'camera',
|
||||
'entity_category': None,
|
||||
'entity_id': 'camera.test_name_stream_0',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Stream 0',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Stream 0',
|
||||
'platform': 'shelly',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': <CameraEntityFeature: 2>,
|
||||
'translation_key': 'stream',
|
||||
'unique_id': '123456789ABC-camera:0-stream_0',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_camera_entity_setup[camera.test_name_stream_0-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
<CameraEntityStateAttribute.ACCESS_TOKEN: 'access_token'>: '1caab5c3b3',
|
||||
<CameraEntityStateAttribute.BRAND: 'brand'>: 'Shelly',
|
||||
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/camera_proxy/camera.test_name_stream_0?token=1caab5c3b3',
|
||||
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test name Stream 0',
|
||||
<CameraEntityStateAttribute.MODEL_NAME: 'model_name'>: 'S1CM-0DXW00',
|
||||
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <CameraEntityFeature: 2>,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'camera.test_name_stream_0',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'idle',
|
||||
})
|
||||
# ---
|
||||
# name: test_camera_entity_setup[camera.test_name_stream_1-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'camera',
|
||||
'entity_category': None,
|
||||
'entity_id': 'camera.test_name_stream_1',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Stream 1',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Stream 1',
|
||||
'platform': 'shelly',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': <CameraEntityFeature: 2>,
|
||||
'translation_key': 'stream',
|
||||
'unique_id': '123456789ABC-camera:0-stream_1',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_camera_entity_setup[camera.test_name_stream_1-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
<CameraEntityStateAttribute.ACCESS_TOKEN: 'access_token'>: '1caab5c3b3',
|
||||
<CameraEntityStateAttribute.BRAND: 'brand'>: 'Shelly',
|
||||
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/camera_proxy/camera.test_name_stream_1?token=1caab5c3b3',
|
||||
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Test name Stream 1',
|
||||
<CameraEntityStateAttribute.MODEL_NAME: 'model_name'>: 'S1CM-0DXW00',
|
||||
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <CameraEntityFeature: 2>,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'camera.test_name_stream_1',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'idle',
|
||||
})
|
||||
# ---
|
||||
@@ -0,0 +1,193 @@
|
||||
"""Tests for Shelly camera platform."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from copy import deepcopy
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from aioshelly.const import MODEL_CAMERA
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.camera import (
|
||||
DATA_COMPONENT,
|
||||
CameraState,
|
||||
get_camera_from_entity_id,
|
||||
)
|
||||
from homeassistant.components.shelly.const import CONF_SLEEP_PERIOD
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_MODEL,
|
||||
CONF_PASSWORD,
|
||||
CONF_USERNAME,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_registry import EntityRegistry
|
||||
|
||||
from . import MOCK_MAC, init_integration, patch_platforms
|
||||
|
||||
from tests.common import snapshot_platform
|
||||
|
||||
CAMERA_ENTITY_ID = "camera.test_name_stream_0"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fixture_platforms() -> Generator[None]:
|
||||
"""Limit platforms under test."""
|
||||
with patch_platforms([Platform.CAMERA]):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_camera_entity_setup(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
entity_registry: EntityRegistry,
|
||||
snapshot: SnapshotAssertion,
|
||||
) -> None:
|
||||
"""Test camera entity is created with correct unique_id and initial state."""
|
||||
with patch("random.SystemRandom.getrandbits", return_value=123123123123):
|
||||
entry = await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
assert hass.states.get(CAMERA_ENTITY_ID)
|
||||
await snapshot_platform(hass, entity_registry, snapshot, entry.entry_id)
|
||||
|
||||
assert (er_entry := entity_registry.async_get(CAMERA_ENTITY_ID))
|
||||
assert er_entry.unique_id == f"{MOCK_MAC}-camera:0-stream_0"
|
||||
|
||||
|
||||
async def test_camera_state_streaming(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Test camera state is streaming when streams > 0."""
|
||||
await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
new_status = deepcopy(mock_camera_rpc_device.status)
|
||||
new_status["camera:0"]["streams"] = 1
|
||||
monkeypatch.setattr(mock_camera_rpc_device, "status", new_status)
|
||||
mock_camera_rpc_device.mock_update()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert (state := hass.states.get(CAMERA_ENTITY_ID))
|
||||
assert state.state == CameraState.STREAMING
|
||||
|
||||
|
||||
async def test_camera_state_recording(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Test camera state is recording when recordings is set."""
|
||||
await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
new_status = deepcopy(mock_camera_rpc_device.status)
|
||||
new_status["camera:0"]["recordings"] = {"id": 1}
|
||||
monkeypatch.setattr(mock_camera_rpc_device, "status", new_status)
|
||||
mock_camera_rpc_device.mock_update()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert (state := hass.states.get(CAMERA_ENTITY_ID))
|
||||
assert state.state == CameraState.RECORDING
|
||||
|
||||
|
||||
async def test_camera_use_stream_for_stills(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
) -> None:
|
||||
"""Test use_stream_for_stills returns True (still images from the RTSP stream)."""
|
||||
await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID)
|
||||
assert camera.use_stream_for_stills is True
|
||||
|
||||
|
||||
async def test_camera_stream_source(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
) -> None:
|
||||
"""Test stream_source returns the RTSP URL for go2rtc."""
|
||||
await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID)
|
||||
result = await camera.stream_source()
|
||||
assert result == "rtsp://192.168.1.37/stream/0"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_camera_stream_source_stream_1(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
) -> None:
|
||||
"""Test stream_source returns correct RTSP URL for stream 1."""
|
||||
await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
camera = get_camera_from_entity_id(hass, "camera.test_name_stream_1")
|
||||
result = await camera.stream_source()
|
||||
assert result == "rtsp://192.168.1.37/stream/1"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("password", "expected_password"),
|
||||
[
|
||||
("password", "password"),
|
||||
("pass:word@1", "pass%3Aword%401"),
|
||||
],
|
||||
)
|
||||
async def test_camera_stream_source_with_credentials(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
password: str,
|
||||
expected_password: str,
|
||||
) -> None:
|
||||
"""Test stream_source returns the RTSP URL with credentials for go2rtc."""
|
||||
await init_integration(
|
||||
hass,
|
||||
3,
|
||||
model=MODEL_CAMERA,
|
||||
data={
|
||||
CONF_HOST: "192.168.1.37",
|
||||
CONF_MODEL: MODEL_CAMERA,
|
||||
CONF_PASSWORD: password,
|
||||
CONF_SLEEP_PERIOD: 0,
|
||||
CONF_USERNAME: "admin",
|
||||
},
|
||||
)
|
||||
|
||||
camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID)
|
||||
result = await camera.stream_source()
|
||||
assert result == f"rtsp://admin:{expected_password}@192.168.1.37/stream/0"
|
||||
|
||||
|
||||
async def test_camera_off_when_streamer_stopped(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Test camera is off when the streamer is not running."""
|
||||
status = deepcopy(mock_camera_rpc_device.status)
|
||||
status["camera:0"]["streamer"] = "stopped"
|
||||
monkeypatch.setattr(mock_camera_rpc_device, "status", status)
|
||||
|
||||
await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
camera = hass.data[DATA_COMPONENT].get_entity(CAMERA_ENTITY_ID)
|
||||
assert camera is not None
|
||||
assert camera.is_on is False
|
||||
|
||||
|
||||
async def test_camera_properties_when_device_not_initialized(
|
||||
hass: HomeAssistant,
|
||||
mock_camera_rpc_device: Mock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Test camera properties return safe values when the device is not initialized."""
|
||||
await init_integration(hass, 3, model=MODEL_CAMERA)
|
||||
|
||||
camera = get_camera_from_entity_id(hass, CAMERA_ENTITY_ID)
|
||||
|
||||
monkeypatch.setattr(mock_camera_rpc_device, "initialized", False)
|
||||
|
||||
assert camera.is_on is False
|
||||
assert camera.available is False
|
||||
Reference in New Issue
Block a user