mirror of
https://github.com/home-assistant/core.git
synced 2026-09-25 17:04:04 -04:00
Add Hikvision event platform for smart event detection targets (#180841)
This commit is contained in:
Paul Tarjan
@@ -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
|
||||
|
||||
@@ -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()
|
||||
@@ -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": {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
Reference in New Issue
Block a user