Add camera platform support to Hikvision integration (#160252)

Co-authored-by: Joostlek <joostlek@outlook.com>

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

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)

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)

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
 
 

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': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'camera',
+    'entity_category': None,
+    'entity_id': 'camera.front_camera',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    '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': <CameraEntityFeature: 2>,
+    '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': <CameraEntityFeature: 2>,
+    }),
+    'context': <ANY>,
+    'entity_id': 'camera.front_camera',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'idle',
+  })
+# ---
+# name: test_nvr_entities[camera.front_camera_channel_1-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    '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.front_camera_channel_1',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    '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': <CameraEntityFeature: 2>,
+    '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': <CameraEntityFeature: 2>,
+    }),
+    'context': <ANY>,
+    'entity_id': 'camera.front_camera_channel_1',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'idle',
+  })
+# ---
+# name: test_nvr_entities[camera.front_camera_channel_2-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    '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.front_camera_channel_2',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    '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': <CameraEntityFeature: 2>,
+    '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': <CameraEntityFeature: 2>,
+    }),
+    'context': <ANY>,
+    'entity_id': 'camera.front_camera_channel_2',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'idle',
+  })
+# ---

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
 
 

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)