From 2a2babceabcd1750bd12f717315af493bc41889e Mon Sep 17 00:00:00 2001 From: Paul Tarjan Date: Thu, 10 Sep 2026 12:26:12 -0600 Subject: [PATCH] Add Hikvision event platform for smart event detection targets (#180841) --- .../components/hikvision/__init__.py | 2 +- homeassistant/components/hikvision/event.py | 169 ++++++++++ .../components/hikvision/strings.json | 41 +++ tests/components/hikvision/conftest.py | 1 + .../hikvision/test_binary_sensor.py | 2 + tests/components/hikvision/test_event.py | 294 ++++++++++++++++++ 6 files changed, 508 insertions(+), 1 deletion(-) create mode 100644 homeassistant/components/hikvision/event.py create mode 100644 tests/components/hikvision/test_event.py diff --git a/homeassistant/components/hikvision/__init__.py b/homeassistant/components/hikvision/__init__.py index cc45552d5dc6..34525036cc11 100644 --- a/homeassistant/components/hikvision/__init__.py +++ b/homeassistant/components/hikvision/__init__.py @@ -26,7 +26,7 @@ from .const import DOMAIN _LOGGER = logging.getLogger(__name__) -PLATFORMS = [Platform.BINARY_SENSOR, Platform.CAMERA] +PLATFORMS = [Platform.BINARY_SENSOR, Platform.CAMERA, Platform.EVENT] @dataclass diff --git a/homeassistant/components/hikvision/event.py b/homeassistant/components/hikvision/event.py new file mode 100644 index 000000000000..2a0a9125480f --- /dev/null +++ b/homeassistant/components/hikvision/event.py @@ -0,0 +1,169 @@ +"""Support for Hikvision smart events represented as event entities.""" + +import logging +from typing import Any, override + +from pyhik.constants import SENSOR_MAP + +from homeassistant.components.event import ( + EventDeviceClass, + EventEntity, + EventEntityDescription, +) +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from . import HikvisionConfigEntry +from .entity import HikvisionEntity + +_LOGGER = logging.getLogger(__name__) + +PARALLEL_UPDATES = 0 + +# Smart events classify what tripped them. Firmware that does not classify, and +# every trip of a non-smart event, reports no target at all. +DETECTION_TARGETS = ("human", "pet", "vehicle") +EVENT_TYPE_TRIGGERED = "triggered" +EVENT_TYPES = [EVENT_TYPE_TRIGGERED, *DETECTION_TARGETS] + +# Keyed by the friendly names pyhik emits in `current_event_states`, like the +# binary sensor descriptions, so both platforms name the same event the same +# way. Only the events that can carry a detection target are listed; the rest +# are already fully described by their binary sensor. +EVENT_DESCRIPTIONS: dict[str, EventEntityDescription] = { + SENSOR_MAP["vmd"]: EventEntityDescription( + key="motion", + translation_key="motion", + device_class=EventDeviceClass.MOTION, + event_types=EVENT_TYPES, + ), + SENSOR_MAP["linedetection"]: EventEntityDescription( + key="line_crossing", + translation_key="line_crossing", + device_class=EventDeviceClass.MOTION, + event_types=EVENT_TYPES, + ), + SENSOR_MAP["fielddetection"]: EventEntityDescription( + key="field_detection", + translation_key="field_detection", + device_class=EventDeviceClass.MOTION, + event_types=EVENT_TYPES, + ), +} + + +def event_type_for_target(detection_target: str | None) -> str: + """Return the event type for a pyhik detection target.""" + if detection_target is None: + return EVENT_TYPE_TRIGGERED + if detection_target not in DETECTION_TARGETS: + _LOGGER.warning( + "Unknown Hikvision detection target '%s', please report this at " + "https://github.com/home-assistant/core/issues", + detection_target, + ) + return EVENT_TYPE_TRIGGERED + return detection_target + + +async def async_setup_entry( + hass: HomeAssistant, + entry: HikvisionConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up Hikvision events from a config entry.""" + sensors = entry.runtime_data.camera.current_event_states + if not sensors: + return + + entities: list[HikvisionEvent] = [] + for sensor_type, channel_list in sensors.items(): + description = EVENT_DESCRIPTIONS.get(sensor_type) + if description is None: + continue + # pyhik can report the same channel more than once for a sensor type + # (e.g. when a channel has several notification methods enabled), so + # deduplicate on the channel to avoid colliding unique IDs. + seen_channels: set[int] = set() + for channel_info in channel_list: + channel = channel_info[1] + if channel in seen_channels: + continue + seen_channels.add(channel) + entities.append( + HikvisionEvent( + hass=hass, + entry=entry, + description=description, + sensor_type=sensor_type, + channel=channel, + ) + ) + + async_add_entities(entities) + + +class HikvisionEvent(HikvisionEntity, EventEntity): + """Representation of a Hikvision event.""" + + _attr_should_poll = False + entity_description: EventEntityDescription + + def __init__( + self, + hass: HomeAssistant, + entry: HikvisionConfigEntry, + description: EventEntityDescription, + sensor_type: str, + channel: int, + ) -> None: + """Initialize the event entity.""" + super().__init__(hass, entry, channel) + self.entity_description = description + self._sensor_type = sensor_type + + self._attr_unique_id = f"{self._data.device_id}_{sensor_type}_{channel}" + + # pyhik routes an update to the callbacks registered under this exact + # identifier and passes it back as the callback's message. + self._callback_id = f"{self._data.device_id}.{sensor_type}.{channel}" + + # An event already active at startup must not be replayed as new. + self._is_on = self._get_sensor_attributes()[0] + + def _get_sensor_attributes(self) -> tuple[bool, Any, Any, Any, str | None]: + """Get sensor attributes from camera.""" + return self._camera.fetch_attributes(self._sensor_type, self._channel) + + @property + @override + def available(self) -> bool: + """Return true if the device's event stream is connected.""" + return self._camera.stream_connected + + @override + async def async_added_to_hass(self) -> None: + """Register callback when entity is added.""" + await super().async_added_to_hass() + + self._camera.add_update_callback(self._update_callback, self._callback_id) + + def _update_callback(self, msg: str) -> None: + """Handle an update from pyhik's event stream thread.""" + # Read the state on the callback thread: a trip that has already ended + # by the time the event loop runs would otherwise be read as inactive + # by both handlers and never fire. + self.hass.loop.call_soon_threadsafe( + self._async_handle_update, self._get_sensor_attributes() + ) + + @callback + def _async_handle_update( + self, attributes: tuple[bool, Any, Any, Any, str | None] + ) -> None: + """Trigger an event when the underlying event has turned on.""" + is_on = attributes[0] + if is_on and not self._is_on: + self._trigger_event(event_type_for_target(attributes[4])) + self._is_on = is_on + self.async_write_ha_state() diff --git a/homeassistant/components/hikvision/strings.json b/homeassistant/components/hikvision/strings.json index 673c9d5c7612..f1a5aa05cd10 100644 --- a/homeassistant/components/hikvision/strings.json +++ b/homeassistant/components/hikvision/strings.json @@ -90,6 +90,47 @@ "video_mismatch": { "name": "Video mismatch" } + }, + "event": { + "field_detection": { + "name": "[%key:component::hikvision::entity::binary_sensor::field_detection::name%]", + "state_attributes": { + "event_type": { + "state": { + "human": "Human", + "pet": "Pet", + "triggered": "Triggered", + "vehicle": "Vehicle" + } + } + } + }, + "line_crossing": { + "name": "[%key:component::hikvision::entity::binary_sensor::line_crossing::name%]", + "state_attributes": { + "event_type": { + "state": { + "human": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::human%]", + "pet": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::pet%]", + "triggered": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::triggered%]", + "vehicle": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::vehicle%]" + } + } + } + }, + "motion": { + "name": "[%key:component::event::entity_component::motion::name%]", + "state_attributes": { + "event_type": { + "state": { + "human": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::human%]", + "pet": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::pet%]", + "triggered": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::triggered%]", + "vehicle": "[%key:component::hikvision::entity::event::field_detection::state_attributes::event_type::state::vehicle%]" + } + } + } + } } }, "issues": { diff --git a/tests/components/hikvision/conftest.py b/tests/components/hikvision/conftest.py index b0c090451106..ba34c682e13b 100644 --- a/tests/components/hikvision/conftest.py +++ b/tests/components/hikvision/conftest.py @@ -121,6 +121,7 @@ def mock_hikcamera(mock_hik_get_channels: MagicMock) -> Generator[MagicMock]: None, None, "2024-01-01T00:00:00Z", + None, ) camera.get_event_triggers.return_value = {} camera.stream_connected = True diff --git a/tests/components/hikvision/test_binary_sensor.py b/tests/components/hikvision/test_binary_sensor.py index 74b9cf19a22e..ba9537a9e6e4 100644 --- a/tests/components/hikvision/test_binary_sensor.py +++ b/tests/components/hikvision/test_binary_sensor.py @@ -220,6 +220,7 @@ async def test_binary_sensor_state_on( None, None, "2024-01-01T12:00:00Z", + None, ) await setup_integration(hass, mock_config_entry) @@ -378,6 +379,7 @@ async def test_binary_sensor_update_callback( None, None, "2024-01-01T12:00:00Z", + None, ) # Get the registered callback and call it diff --git a/tests/components/hikvision/test_event.py b/tests/components/hikvision/test_event.py new file mode 100644 index 000000000000..b1e82fd42712 --- /dev/null +++ b/tests/components/hikvision/test_event.py @@ -0,0 +1,294 @@ +"""Test Hikvision events.""" + +from collections.abc import Callable +from unittest.mock import MagicMock + +from pyhik.constants import SENSOR_MAP +import pytest + +from homeassistant.components.event import ( + ATTR_EVENT_TYPE, + ATTR_EVENT_TYPES, + DOMAIN as EVENT_DOMAIN, + EventDeviceClass, +) +from homeassistant.const import ( + ATTR_DEVICE_CLASS, + STATE_UNAVAILABLE, + STATE_UNKNOWN, + Platform, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er + +from . import setup_integration +from .conftest import TEST_DEVICE_ID + +from tests.common import MockConfigEntry + +MOTION_ENTITY_ID = "event.front_camera_motion" +LINE_CROSSING_ENTITY_ID = "event.front_camera_line_crossing" +MOTION_CALLBACK_ID = f"{TEST_DEVICE_ID}.{SENSOR_MAP['vmd']}.1" +TEST_TRIP_TIME = "2024-01-01T12:00:00Z" + + +@pytest.fixture +def platforms() -> list[Platform]: + """Platforms, which should be loaded during the test.""" + return [Platform.EVENT] + + +def get_callbacks(mock_hikcamera: MagicMock) -> dict[str, Callable[[str], None]]: + """Return the update callbacks pyhik was handed, keyed by their ID.""" + return { + call.args[1]: call.args[0] + for call in mock_hikcamera.return_value.add_update_callback.call_args_list + } + + +def set_event_state( + mock_hikcamera: MagicMock, is_on: bool, detection_target: str | None +) -> None: + """Set the attribute tuple pyhik reports for the event.""" + mock_hikcamera.return_value.fetch_attributes.return_value = ( + is_on, + 1, + 1, + TEST_TRIP_TIME, + detection_target, + ) + + +async def test_events_created( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test an event entity is created for each smart event.""" + await setup_integration(hass, mock_config_entry) + + assert set(hass.states.async_entity_ids(EVENT_DOMAIN)) == { + MOTION_ENTITY_ID, + LINE_CROSSING_ENTITY_ID, + } + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.state == STATE_UNKNOWN + assert state.attributes[ATTR_DEVICE_CLASS] == EventDeviceClass.MOTION + assert state.attributes[ATTR_EVENT_TYPES] == [ + "triggered", + "human", + "pet", + "vehicle", + ] + + +async def test_events_not_created_for_non_smart_events( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test events that never carry a detection target get no event entity.""" + mock_hikcamera.return_value.current_event_states = { + SENSOR_MAP["vmd"]: [(False, 1)], + SENSOR_MAP["diskfull"]: [(False, 1)], + SENSOR_MAP["tamperdetection"]: [(False, 1)], + } + + await setup_integration(hass, mock_config_entry) + + assert hass.states.async_entity_ids(EVENT_DOMAIN) == [MOTION_ENTITY_ID] + + +async def test_events_no_sensors( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test setup when the device reports no events.""" + mock_hikcamera.return_value.current_event_states = None + + await setup_integration(hass, mock_config_entry) + + assert hass.states.async_entity_ids(EVENT_DOMAIN) == [] + + +@pytest.mark.parametrize( + ("detection_target", "expected_event_type"), + [ + pytest.param("human", "human", id="human"), + pytest.param("vehicle", "vehicle", id="vehicle"), + pytest.param("pet", "pet", id="pet"), + pytest.param(None, "triggered", id="no_detection_target"), + pytest.param("bicycle", "triggered", id="unrecognized_detection_target"), + ], +) +async def test_event_triggered_with_detection_target( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + detection_target: str | None, + expected_event_type: str, +) -> None: + """Test the detection target is reported as the event type.""" + await setup_integration(hass, mock_config_entry) + + set_event_state(mock_hikcamera, True, detection_target) + get_callbacks(mock_hikcamera)[MOTION_CALLBACK_ID]("motion detected") + await hass.async_block_till_done() + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.state != STATE_UNKNOWN + assert state.attributes[ATTR_EVENT_TYPE] == expected_event_type + # The detection target is the event type, never an extra state attribute. + assert "detection_target" not in state.attributes + + +async def test_unrecognized_detection_target_logs_warning( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test a detection target the integration does not know about is reported.""" + await setup_integration(hass, mock_config_entry) + + set_event_state(mock_hikcamera, True, "bicycle") + get_callbacks(mock_hikcamera)[MOTION_CALLBACK_ID]("motion detected") + await hass.async_block_till_done() + + assert "Unknown Hikvision detection target 'bicycle'" in caplog.text + + +async def test_event_only_triggered_on_a_new_trip( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test updates while the event stays active do not trigger a new event.""" + await setup_integration(hass, mock_config_entry) + callbacks = get_callbacks(mock_hikcamera) + + set_event_state(mock_hikcamera, True, "human") + callbacks[MOTION_CALLBACK_ID]("motion detected") + await hass.async_block_till_done() + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.attributes[ATTR_EVENT_TYPE] == "human" + + # pyhik keeps updating while the event is active; that is the same trip. + set_event_state(mock_hikcamera, True, "vehicle") + callbacks[MOTION_CALLBACK_ID]("motion detected") + await hass.async_block_till_done() + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.attributes[ATTR_EVENT_TYPE] == "human" + + set_event_state(mock_hikcamera, False, None) + callbacks[MOTION_CALLBACK_ID]("motion cleared") + await hass.async_block_till_done() + + set_event_state(mock_hikcamera, True, "vehicle") + callbacks[MOTION_CALLBACK_ID]("motion detected") + await hass.async_block_till_done() + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.attributes[ATTR_EVENT_TYPE] == "vehicle" + + +async def test_event_active_at_setup_is_not_replayed( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test an event already active when the entity is added does not fire.""" + set_event_state(mock_hikcamera, True, "human") + + await setup_integration(hass, mock_config_entry) + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.state == STATE_UNKNOWN + + +async def test_event_unavailable_when_stream_disconnected( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test events go unavailable when the event stream disconnects.""" + await setup_integration(hass, mock_config_entry) + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.state == STATE_UNKNOWN + + # pyhik notifies every registered callback when the stream drops + mock_hikcamera.return_value.stream_connected = False + get_callbacks(mock_hikcamera)[MOTION_CALLBACK_ID]("stream disconnected") + await hass.async_block_till_done() + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.state == STATE_UNAVAILABLE + + +async def test_event_fires_for_a_trip_that_ends_before_the_loop_runs( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test a trip is not lost when it ends before the event loop catches up.""" + await setup_integration(hass, mock_config_entry) + callbacks = get_callbacks(mock_hikcamera) + + # Both callbacks land on pyhik's thread before the loop runs either of them. + set_event_state(mock_hikcamera, True, "human") + callbacks[MOTION_CALLBACK_ID]("motion detected") + set_event_state(mock_hikcamera, False, None) + callbacks[MOTION_CALLBACK_ID]("motion cleared") + await hass.async_block_till_done() + + state = hass.states.get(MOTION_ENTITY_ID) + assert state is not None + assert state.attributes[ATTR_EVENT_TYPE] == "human" + + +@pytest.mark.parametrize("amount_of_channels", [2]) +async def test_event_duplicate_channels_deduplicated( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + entity_registry: er.EntityRegistry, +) -> None: + """Test duplicate channel entries do not create colliding unique IDs.""" + mock_hikcamera.return_value.get_type = "NVR" + mock_hikcamera.return_value.current_event_states = { + SENSOR_MAP["linedetection"]: [ + (False, 1), + (False, 1), + (False, 1), + (False, 2), + (False, 2), + ], + } + + await setup_integration(hass, mock_config_entry) + + assert len(hass.states.async_entity_ids(EVENT_DOMAIN)) == 2 + + unique_ids = { + entry.unique_id + for entry in er.async_entries_for_config_entry( + entity_registry, mock_config_entry.entry_id + ) + } + assert unique_ids == { + f"{TEST_DEVICE_ID}_{SENSOR_MAP['linedetection']}_1", + f"{TEST_DEVICE_ID}_{SENSOR_MAP['linedetection']}_2", + }