diff --git a/homeassistant/components/hikvision/__init__.py b/homeassistant/components/hikvision/__init__.py index da3cdcdf4de..4c044451659 100644 --- a/homeassistant/components/hikvision/__init__.py +++ b/homeassistant/components/hikvision/__init__.py @@ -20,10 +20,13 @@ from homeassistant.const import ( ) from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers import device_registry as dr + +from .const import DOMAIN _LOGGER = logging.getLogger(__name__) -PLATFORMS = [Platform.BINARY_SENSOR] +PLATFORMS = [Platform.BINARY_SENSOR, Platform.CAMERA] @dataclass @@ -104,6 +107,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) -> # Start the event stream await hass.async_add_executor_job(camera.start_stream) + # Register the main device before platforms that use via_device + device_registry = dr.async_get(hass) + device_registry.async_get_or_create( + config_entry_id=entry.entry_id, + identifiers={(DOMAIN, device_id)}, + name=device_name, + manufacturer="Hikvision", + model=device_type, + ) + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True diff --git a/homeassistant/components/hikvision/binary_sensor.py b/homeassistant/components/hikvision/binary_sensor.py index 6a354458ed3..192248e3aac 100644 --- a/homeassistant/components/hikvision/binary_sensor.py +++ b/homeassistant/components/hikvision/binary_sensor.py @@ -185,19 +185,26 @@ class HikvisionBinarySensor(BinarySensorEntity): # Build unique ID self._attr_unique_id = f"{self._data.device_id}_{sensor_type}_{channel}" - # Build entity name based on device type - if self._data.device_type == "NVR": - self._attr_name = f"{sensor_type} {channel}" - else: - self._attr_name = sensor_type - # Device info for device registry - self._attr_device_info = DeviceInfo( - identifiers={(DOMAIN, self._data.device_id)}, - name=self._data.device_name, - manufacturer="Hikvision", - model=self._data.device_type, - ) + if self._data.device_type == "NVR": + # NVR channels get their own device linked to the NVR via via_device + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, f"{self._data.device_id}_{channel}")}, + via_device=(DOMAIN, self._data.device_id), + name=f"{self._data.device_name} Channel {channel}", + manufacturer="Hikvision", + model="NVR Channel", + ) + self._attr_name = sensor_type + else: + # Single camera device + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, self._data.device_id)}, + name=self._data.device_name, + manufacturer="Hikvision", + model=self._data.device_type, + ) + self._attr_name = sensor_type # Set device class self._attr_device_class = DEVICE_CLASS_MAP.get(sensor_type) diff --git a/homeassistant/components/hikvision/camera.py b/homeassistant/components/hikvision/camera.py new file mode 100644 index 00000000000..19518e8daa9 --- /dev/null +++ b/homeassistant/components/hikvision/camera.py @@ -0,0 +1,93 @@ +"""Support for Hikvision cameras.""" + +from __future__ import annotations + +from homeassistant.components.camera import Camera, CameraEntityFeature +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from . import HikvisionConfigEntry +from .const import DOMAIN + +PARALLEL_UPDATES = 0 + + +async def async_setup_entry( + hass: HomeAssistant, + entry: HikvisionConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up Hikvision cameras from a config entry.""" + data = entry.runtime_data + camera = data.camera + + # Get available channels from the library + channels = await hass.async_add_executor_job(camera.get_channels) + + if channels: + entities = [HikvisionCamera(entry, channel) for channel in channels] + else: + # Fallback to single camera if no channels detected + entities = [HikvisionCamera(entry, 1)] + + async_add_entities(entities) + + +class HikvisionCamera(Camera): + """Representation of a Hikvision camera.""" + + _attr_has_entity_name = True + _attr_name = None + _attr_supported_features = CameraEntityFeature.STREAM + + def __init__( + self, + entry: HikvisionConfigEntry, + channel: int, + ) -> None: + """Initialize the camera.""" + super().__init__() + self._data = entry.runtime_data + self._channel = channel + self._camera = self._data.camera + + # Build unique ID (unique per platform per integration) + self._attr_unique_id = f"{self._data.device_id}_{channel}" + + # Device info for device registry + if self._data.device_type == "NVR": + # NVR channels get their own device linked to the NVR via via_device + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, f"{self._data.device_id}_{channel}")}, + via_device=(DOMAIN, self._data.device_id), + name=f"{self._data.device_name} Channel {channel}", + manufacturer="Hikvision", + model="NVR Channel", + ) + else: + # Single camera device + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, self._data.device_id)}, + name=self._data.device_name, + manufacturer="Hikvision", + model=self._data.device_type, + ) + + async def async_camera_image( + self, width: int | None = None, height: int | None = None + ) -> bytes | None: + """Return a still image from the camera.""" + try: + return await self.hass.async_add_executor_job( + self._camera.get_snapshot, self._channel + ) + except Exception as err: + raise HomeAssistantError( + f"Error getting image from {self._data.device_name} channel {self._channel}: {err}" + ) from err + + async def stream_source(self) -> str | None: + """Return the stream source URL.""" + return self._camera.get_stream_url(self._channel) diff --git a/tests/components/hikvision/conftest.py b/tests/components/hikvision/conftest.py index 4964b1f8f32..d99e68a62a6 100644 --- a/tests/components/hikvision/conftest.py +++ b/tests/components/hikvision/conftest.py @@ -1,10 +1,11 @@ """Common fixtures for the Hikvision tests.""" -from collections.abc import Generator -from unittest.mock import AsyncMock, MagicMock, patch +from collections.abc import AsyncGenerator, Generator +from unittest.mock import MagicMock, patch import pytest +from homeassistant.components.hikvision import PLATFORMS from homeassistant.components.hikvision.const import DOMAIN from homeassistant.const import ( CONF_HOST, @@ -12,6 +13,7 @@ from homeassistant.const import ( CONF_PORT, CONF_SSL, CONF_USERNAME, + Platform, ) from tests.common import MockConfigEntry @@ -25,7 +27,20 @@ TEST_DEVICE_NAME = "Front Camera" @pytest.fixture -def mock_setup_entry() -> Generator[AsyncMock]: +def platforms() -> list[Platform]: + """Platforms, which should be loaded during the test.""" + return PLATFORMS + + +@pytest.fixture(autouse=True) +async def mock_patch_platforms(platforms: list[Platform]) -> AsyncGenerator[None]: + """Fixture to set up platforms for tests.""" + with patch(f"homeassistant.components.{DOMAIN}.PLATFORMS", platforms): + yield + + +@pytest.fixture +def mock_setup_entry() -> Generator[MagicMock]: """Override async_setup_entry.""" with patch( "homeassistant.components.hikvision.async_setup_entry", return_value=True @@ -58,7 +73,6 @@ def mock_hikcamera() -> Generator[MagicMock]: with ( patch( "homeassistant.components.hikvision.HikCamera", - autospec=True, ) as hikcamera_mock, patch( "homeassistant.components.hikvision.config_flow.HikCamera", @@ -80,6 +94,15 @@ def mock_hikcamera() -> Generator[MagicMock]: "2024-01-01T00:00:00Z", ) camera.get_event_triggers.return_value = {} + + # pyHik 0.4.0 methods + camera.get_channels.return_value = [1] + camera.get_snapshot.return_value = b"fake_image_data" + camera.get_stream_url.return_value = ( + f"rtsp://{TEST_USERNAME}:{TEST_PASSWORD}" + f"@{TEST_HOST}:554/Streaming/Channels/1" + ) + yield hikcamera_mock diff --git a/tests/components/hikvision/snapshots/test_camera.ambr b/tests/components/hikvision/snapshots/test_camera.ambr new file mode 100644 index 00000000000..939331d5c3f --- /dev/null +++ b/tests/components/hikvision/snapshots/test_camera.ambr @@ -0,0 +1,154 @@ +# serializer version: 1 +# name: test_all_entities[camera.front_camera-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'camera', + 'entity_category': None, + 'entity_id': 'camera.front_camera', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': None, + 'platform': 'hikvision', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': , + 'translation_key': None, + 'unique_id': 'DS-2CD2142FWD-I20170101AAAA_1', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[camera.front_camera-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'access_token': '1caab5c3b3', + 'entity_picture': '/api/camera_proxy/camera.front_camera?token=1caab5c3b3', + 'friendly_name': 'Front Camera', + 'supported_features': , + }), + 'context': , + 'entity_id': 'camera.front_camera', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'idle', + }) +# --- +# name: test_nvr_entities[camera.front_camera_channel_1-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'camera', + 'entity_category': None, + 'entity_id': 'camera.front_camera_channel_1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': None, + 'platform': 'hikvision', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': , + 'translation_key': None, + 'unique_id': 'DS-2CD2142FWD-I20170101AAAA_1', + 'unit_of_measurement': None, + }) +# --- +# name: test_nvr_entities[camera.front_camera_channel_1-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'access_token': '1caab5c3b3', + 'entity_picture': '/api/camera_proxy/camera.front_camera_channel_1?token=1caab5c3b3', + 'friendly_name': 'Front Camera Channel 1', + 'supported_features': , + }), + 'context': , + 'entity_id': 'camera.front_camera_channel_1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'idle', + }) +# --- +# name: test_nvr_entities[camera.front_camera_channel_2-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'camera', + 'entity_category': None, + 'entity_id': 'camera.front_camera_channel_2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': None, + 'platform': 'hikvision', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': , + 'translation_key': None, + 'unique_id': 'DS-2CD2142FWD-I20170101AAAA_2', + 'unit_of_measurement': None, + }) +# --- +# name: test_nvr_entities[camera.front_camera_channel_2-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'access_token': '1caab5c3b3', + 'entity_picture': '/api/camera_proxy/camera.front_camera_channel_2?token=1caab5c3b3', + 'friendly_name': 'Front Camera Channel 2', + 'supported_features': , + }), + 'context': , + 'entity_id': 'camera.front_camera_channel_2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'idle', + }) +# --- diff --git a/tests/components/hikvision/test_binary_sensor.py b/tests/components/hikvision/test_binary_sensor.py index 7c659c1c11a..88622889b34 100644 --- a/tests/components/hikvision/test_binary_sensor.py +++ b/tests/components/hikvision/test_binary_sensor.py @@ -17,6 +17,7 @@ from homeassistant.const import ( CONF_SSL, CONF_USERNAME, STATE_OFF, + Platform, ) from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant from homeassistant.helpers import ( @@ -39,6 +40,12 @@ from .conftest import ( from tests.common import MockConfigEntry, snapshot_platform +@pytest.fixture +def platforms() -> list[Platform]: + """Platforms, which should be loaded during the test.""" + return [Platform.BINARY_SENSOR] + + @pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_all_entities( hass: HomeAssistant, @@ -132,11 +139,11 @@ async def test_binary_sensor_nvr_device( await setup_integration(hass, mock_config_entry) - # NVR sensors should include channel number in name - state = hass.states.get("binary_sensor.front_camera_motion_1") + # NVR sensors are on per-channel devices + state = hass.states.get("binary_sensor.front_camera_channel_1_motion") assert state is not None - state = hass.states.get("binary_sensor.front_camera_motion_2") + state = hass.states.get("binary_sensor.front_camera_channel_2_motion") assert state is not None diff --git a/tests/components/hikvision/test_camera.py b/tests/components/hikvision/test_camera.py new file mode 100644 index 00000000000..7ae42aaf23e --- /dev/null +++ b/tests/components/hikvision/test_camera.py @@ -0,0 +1,165 @@ +"""Test Hikvision cameras.""" + +from unittest.mock import MagicMock, patch + +import pytest +from syrupy.assertion import SnapshotAssertion + +from homeassistant.components.camera import async_get_image, async_get_stream_source +from homeassistant.components.hikvision.const import DOMAIN +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import device_registry as dr, entity_registry as er + +from . import setup_integration +from .conftest import TEST_DEVICE_ID, TEST_DEVICE_NAME, TEST_HOST, TEST_PASSWORD + +from tests.common import MockConfigEntry, snapshot_platform + + +@pytest.fixture +def platforms() -> list[Platform]: + """Return platforms to load during test.""" + return [Platform.CAMERA] + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_all_entities( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + entity_registry: er.EntityRegistry, + snapshot: SnapshotAssertion, +) -> None: + """Test all camera entities.""" + with patch("random.SystemRandom.getrandbits", return_value=123123123123): + await setup_integration(hass, mock_config_entry) + + await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id) + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_nvr_entities( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + entity_registry: er.EntityRegistry, + snapshot: SnapshotAssertion, +) -> None: + """Test NVR camera entities with multiple channels.""" + mock_hikcamera.return_value.get_type = "NVR" + mock_hikcamera.return_value.get_channels.return_value = [1, 2] + + with patch("random.SystemRandom.getrandbits", return_value=123123123123): + await setup_integration(hass, mock_config_entry) + + await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id) + + +async def test_camera_device_info( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + device_registry: dr.DeviceRegistry, +) -> None: + """Test camera is linked to device.""" + await setup_integration(hass, mock_config_entry) + + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, TEST_DEVICE_ID)} + ) + assert device_entry is not None + assert device_entry.name == TEST_DEVICE_NAME + assert device_entry.manufacturer == "Hikvision" + assert device_entry.model == "Camera" + + +async def test_camera_no_channels_creates_single_camera( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test camera created when device returns no channels.""" + mock_hikcamera.return_value.get_channels.return_value = [] + + await setup_integration(hass, mock_config_entry) + + # Single camera should be created for channel 1 + states = hass.states.async_entity_ids("camera") + assert len(states) == 1 + + state = hass.states.get("camera.front_camera") + assert state is not None + + +async def test_camera_image( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test getting camera image.""" + await setup_integration(hass, mock_config_entry) + + image = await async_get_image(hass, "camera.front_camera") + assert image.content == b"fake_image_data" + + # Verify get_snapshot was called with channel 1 + mock_hikcamera.return_value.get_snapshot.assert_called_with(1) + + +async def test_camera_image_error( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test camera image error handling.""" + mock_hikcamera.return_value.get_snapshot.side_effect = Exception("Connection error") + + await setup_integration(hass, mock_config_entry) + + with pytest.raises(HomeAssistantError, match="Error getting image"): + await async_get_image(hass, "camera.front_camera") + + +async def test_camera_stream_source( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test camera stream source URL.""" + await setup_integration(hass, mock_config_entry) + + stream_url = await async_get_stream_source(hass, "camera.front_camera") + + # Verify RTSP URL from library + assert stream_url is not None + assert stream_url.startswith("rtsp://") + assert f"@{TEST_HOST}:554/Streaming/Channels/1" in stream_url + + # Verify get_stream_url was called with channel 1 + mock_hikcamera.return_value.get_stream_url.assert_called_with(1) + + +async def test_camera_stream_source_nvr( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test NVR camera stream source URL.""" + mock_hikcamera.return_value.get_type = "NVR" + mock_hikcamera.return_value.get_channels.return_value = [2] + mock_hikcamera.return_value.get_stream_url.return_value = ( + f"rtsp://admin:{TEST_PASSWORD}@{TEST_HOST}:554/Streaming/Channels/201" + ) + + await setup_integration(hass, mock_config_entry) + + stream_url = await async_get_stream_source(hass, "camera.front_camera_channel_2") + + # NVR channel 2 should use stream channel 201 + assert stream_url is not None + assert f"@{TEST_HOST}:554/Streaming/Channels/201" in stream_url + + # Verify get_stream_url was called with channel 2 + mock_hikcamera.return_value.get_stream_url.assert_called_with(2)