From 4e56217b895d71f1149bea832ac55ce5cdcea997 Mon Sep 17 00:00:00 2001 From: Christopher Bailey Date: Wed, 5 Jan 2022 16:59:21 -0500 Subject: [PATCH] Add UniFi Protect binary_sensor platform (#63489) Co-authored-by: J. Nick Koston --- .../components/unifiprotect/binary_sensor.py | 273 +++++++++++++++ .../components/unifiprotect/const.py | 5 + .../components/unifiprotect/entity.py | 35 ++ tests/components/unifiprotect/conftest.py | 6 + .../unifiprotect/test_binary_sensor.py | 327 ++++++++++++++++++ 5 files changed, 646 insertions(+) create mode 100644 homeassistant/components/unifiprotect/binary_sensor.py create mode 100644 tests/components/unifiprotect/test_binary_sensor.py diff --git a/homeassistant/components/unifiprotect/binary_sensor.py b/homeassistant/components/unifiprotect/binary_sensor.py new file mode 100644 index 00000000000..2bad7d509d4 --- /dev/null +++ b/homeassistant/components/unifiprotect/binary_sensor.py @@ -0,0 +1,273 @@ +"""This component provides binary sensors for UniFi Protect.""" +from __future__ import annotations + +from copy import copy +from dataclasses import dataclass +from datetime import datetime +import logging +from typing import Any, Final + +from pyunifiprotect.data import NVR, Camera, Light, Sensor + +from homeassistant.components.binary_sensor import ( + DEVICE_CLASS_BATTERY, + DEVICE_CLASS_DOOR, + DEVICE_CLASS_MOTION, + DEVICE_CLASS_PROBLEM, + BinarySensorEntity, + BinarySensorEntityDescription, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ATTR_LAST_TRIP_TIME, ATTR_MODEL +from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback +from homeassistant.helpers.entity import EntityCategory +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.event import async_call_later +from homeassistant.util.dt import utcnow + +from .const import DOMAIN, RING_INTERVAL +from .data import ProtectData +from .entity import ProtectDeviceEntity, ProtectNVREntity, async_all_device_entities +from .models import ProtectRequiredKeysMixin +from .utils import get_nested_attr + +_LOGGER = logging.getLogger(__name__) + + +@dataclass +class ProtectBinaryEntityDescription( + ProtectRequiredKeysMixin, BinarySensorEntityDescription +): + """Describes UniFi Protect Binary Sensor entity.""" + + +_KEY_DOORBELL = "doorbell" +_KEY_MOTION = "motion" +_KEY_DOOR = "door" +_KEY_DARK = "dark" +_KEY_BATTERY_LOW = "battery_low" +_KEY_DISK_HEALTH = "disk_health" + +DEVICE_CLASS_RING: Final = "unifiprotect__ring" + + +CAMERA_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = ( + ProtectBinaryEntityDescription( + key=_KEY_DOORBELL, + name="Doorbell Chime", + device_class=DEVICE_CLASS_RING, + icon="mdi:doorbell-video", + ufp_required_field="feature_flags.has_chime", + ufp_value="last_ring", + ), + ProtectBinaryEntityDescription( + key=_KEY_DARK, + name="Is Dark", + icon="mdi:brightness-6", + ufp_value="is_dark", + ), +) + +LIGHT_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = ( + ProtectBinaryEntityDescription( + key=_KEY_DARK, + name="Is Dark", + icon="mdi:brightness-6", + ufp_value="is_dark", + ), + ProtectBinaryEntityDescription( + key=_KEY_MOTION, + name="Motion Detected", + device_class=DEVICE_CLASS_MOTION, + ufp_value="is_pir_motion_detected", + ), +) + +SENSE_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = ( + ProtectBinaryEntityDescription( + key=_KEY_DOOR, + name="Door", + device_class=DEVICE_CLASS_DOOR, + ufp_value="is_opened", + ), + ProtectBinaryEntityDescription( + key=_KEY_BATTERY_LOW, + name="Battery low", + device_class=DEVICE_CLASS_BATTERY, + entity_category=EntityCategory.DIAGNOSTIC, + ufp_value="battery_status.is_low", + ), + ProtectBinaryEntityDescription( + key=_KEY_MOTION, + name="Motion Detected", + device_class=DEVICE_CLASS_MOTION, + ufp_value="is_motion_detected", + ), +) + +DISK_SENSORS: tuple[ProtectBinaryEntityDescription, ...] = ( + ProtectBinaryEntityDescription( + key=_KEY_DISK_HEALTH, + name="Disk {index} Health", + device_class=DEVICE_CLASS_PROBLEM, + entity_category=EntityCategory.DIAGNOSTIC, + ), +) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up binary sensors for UniFi Protect integration.""" + data: ProtectData = hass.data[DOMAIN][entry.entry_id] + entities: list[ProtectDeviceEntity] = async_all_device_entities( + data, + ProtectDeviceBinarySensor, + camera_descs=CAMERA_SENSORS, + light_descs=LIGHT_SENSORS, + sense_descs=SENSE_SENSORS, + ) + entities += _async_nvr_entities(data) + + async_add_entities(entities) + + +@callback +def _async_nvr_entities( + data: ProtectData, +) -> list[ProtectDeviceEntity]: + entities: list[ProtectDeviceEntity] = [] + device = data.api.bootstrap.nvr + for index, _ in enumerate(device.system_info.storage.devices): + for description in DISK_SENSORS: + entities.append( + ProtectDiskBinarySensor(data, device, description, index=index) + ) + _LOGGER.debug( + "Adding binary sensor entity %s", + (description.name or "{index}").format(index=index), + ) + + return entities + + +class ProtectDeviceBinarySensor(ProtectDeviceEntity, BinarySensorEntity): + """A UniFi Protect Device Binary Sensor.""" + + def __init__( + self, + data: ProtectData, + description: ProtectBinaryEntityDescription, + device: Camera | Light | Sensor | None = None, + ) -> None: + """Initialize the Binary Sensor.""" + + if device and not hasattr(self, "device"): + self.device: Camera | Light | Sensor = device + self.entity_description: ProtectBinaryEntityDescription = description + super().__init__(data) + self._doorbell_callback: CALLBACK_TYPE | None = None + + @callback + def _async_update_extra_attrs_from_protect(self) -> dict[str, Any]: + attrs: dict[str, Any] = {} + key = self.entity_description.key + + if key == _KEY_DARK: + return attrs + + if key == _KEY_DOORBELL: + assert isinstance(self.device, Camera) + attrs[ATTR_LAST_TRIP_TIME] = self.device.last_ring + elif isinstance(self.device, Sensor): + if key in (_KEY_MOTION, _KEY_DOOR): + if key == _KEY_MOTION: + last_trip = self.device.motion_detected_at + else: + last_trip = self.device.open_status_changed_at + + attrs[ATTR_LAST_TRIP_TIME] = last_trip + elif isinstance(self.device, Light): + if key == _KEY_MOTION: + attrs[ATTR_LAST_TRIP_TIME] = self.device.last_motion + + return attrs + + @callback + def _async_update_device_from_protect(self) -> None: + super()._async_update_device_from_protect() + + assert self.entity_description.ufp_value is not None + + self._attr_extra_state_attributes = ( + self._async_update_extra_attrs_from_protect() + ) + + if self.entity_description.key == _KEY_DOORBELL: + last_ring = get_nested_attr(self.device, self.entity_description.ufp_value) + now = utcnow() + + is_ringing = ( + False if last_ring is None else (now - last_ring) < RING_INTERVAL + ) + _LOGGER.warning("%s, %s, %s", last_ring, now, is_ringing) + if is_ringing: + self._async_cancel_doorbell_callback() + self._doorbell_callback = async_call_later( + self.hass, RING_INTERVAL, self._async_reset_doorbell + ) + self._attr_is_on = is_ringing + else: + self._attr_is_on = get_nested_attr( + self.device, self.entity_description.ufp_value + ) + + @callback + def _async_cancel_doorbell_callback(self) -> None: + if self._doorbell_callback is not None: + _LOGGER.debug("Canceling doorbell ring callback") + self._doorbell_callback() + self._doorbell_callback = None + + async def _async_reset_doorbell(self, now: datetime) -> None: + _LOGGER.debug("Doorbell ring ended") + self._doorbell_callback = None + self._async_updated_event() + + async def async_will_remove_from_hass(self) -> None: + """Run when entity will be removed from hass.""" + self._async_cancel_doorbell_callback() + return await super().async_will_remove_from_hass() + + +class ProtectDiskBinarySensor(ProtectNVREntity, BinarySensorEntity): + """A UniFi Protect NVR Disk Binary Sensor.""" + + def __init__( + self, + data: ProtectData, + device: NVR, + description: ProtectBinaryEntityDescription, + index: int, + ) -> None: + """Initialize the Binary Sensor.""" + description = copy(description) + description.key = f"{description.key}_{index}" + description.name = (description.name or "{index}").format(index=index) + self._index = index + self.entity_description: ProtectBinaryEntityDescription = description + super().__init__(data, device) + + @callback + def _async_update_device_from_protect(self) -> None: + super()._async_update_device_from_protect() + + disks = self.device.system_info.storage.devices + disk_available = len(disks) > self._index + self._attr_available = self._attr_available and disk_available + if disk_available: + disk = disks[self._index] + self._attr_is_on = not disk.healthy + self._attr_extra_state_attributes = {ATTR_MODEL: disk.model} diff --git a/homeassistant/components/unifiprotect/const.py b/homeassistant/components/unifiprotect/const.py index aa960d2696f..1ca2612a10a 100644 --- a/homeassistant/components/unifiprotect/const.py +++ b/homeassistant/components/unifiprotect/const.py @@ -1,5 +1,7 @@ """Constant definitions for UniFi Protect Integration.""" +from datetime import timedelta + from pyunifiprotect.data.types import ModelType, Version import voluptuous as vol @@ -32,6 +34,8 @@ DEFAULT_BRAND = "Ubiquiti" DEFAULT_SCAN_INTERVAL = 5 DEFAULT_VERIFY_SSL = False +RING_INTERVAL = timedelta(seconds=3) + DEVICE_TYPE_CAMERA = "camera" DEVICES_THAT_ADOPT = { ModelType.CAMERA, @@ -50,6 +54,7 @@ SERVICE_SET_DOORBELL_MESSAGE = "set_doorbell_message" TYPE_EMPTY_VALUE = "" PLATFORMS = [ + Platform.BINARY_SENSOR, Platform.BUTTON, Platform.CAMERA, Platform.LIGHT, diff --git a/homeassistant/components/unifiprotect/entity.py b/homeassistant/components/unifiprotect/entity.py index 44e5846a51e..cc142949ff7 100644 --- a/homeassistant/components/unifiprotect/entity.py +++ b/homeassistant/components/unifiprotect/entity.py @@ -13,6 +13,7 @@ from pyunifiprotect.data import ( StateType, Viewer, ) +from pyunifiprotect.data.nvr import NVR from homeassistant.core import callback import homeassistant.helpers.device_registry as dr @@ -168,3 +169,37 @@ class ProtectDeviceEntity(Entity): self.device.id, self._async_updated_event ) ) + + +class ProtectNVREntity(ProtectDeviceEntity): + """Base class for unifi protect entities.""" + + def __init__( + self, + entry: ProtectData, + device: NVR, + description: EntityDescription | None = None, + ) -> None: + """Initialize the entity.""" + # ProtectNVREntity is intentionally a separate base class + self.device: NVR = device # type: ignore + super().__init__(entry, description=description) + + @callback + def _async_set_device_info(self) -> None: + self._attr_device_info = DeviceInfo( + connections={(dr.CONNECTION_NETWORK_MAC, self.device.mac)}, + identifiers={(DOMAIN, self.device.mac)}, + manufacturer=DEFAULT_BRAND, + name=self.device.name, + model=self.device.type, + sw_version=str(self.device.version), + configuration_url=self.device.api.base_url, + ) + + @callback + def _async_update_device_from_protect(self) -> None: + if self.data.last_update_success: + self.device = self.data.api.bootstrap.nvr + + self._attr_available = self.data.last_update_success diff --git a/tests/components/unifiprotect/conftest.py b/tests/components/unifiprotect/conftest.py index 5931ede0e24..7cd9a95dcb2 100644 --- a/tests/components/unifiprotect/conftest.py +++ b/tests/components/unifiprotect/conftest.py @@ -177,6 +177,12 @@ def mock_sensor(): return Sensor.from_unifi_dict(**data) +@pytest.fixture +def now(): + """Return datetime object that will be consistent throughout test.""" + return dt_util.utcnow() + + async def time_changed(hass: HomeAssistant, seconds: int) -> None: """Trigger time changed.""" next_update = dt_util.utcnow() + timedelta(seconds) diff --git a/tests/components/unifiprotect/test_binary_sensor.py b/tests/components/unifiprotect/test_binary_sensor.py new file mode 100644 index 00000000000..e6a720d0c94 --- /dev/null +++ b/tests/components/unifiprotect/test_binary_sensor.py @@ -0,0 +1,327 @@ +"""Test the UniFi Protect binary_sensor platform.""" +# pylint: disable=protected-access +from __future__ import annotations + +from copy import copy +from datetime import datetime, timedelta +from unittest.mock import Mock + +import pytest +from pyunifiprotect.data import Camera, Light +from pyunifiprotect.data.devices import Sensor + +from homeassistant.components.unifiprotect.binary_sensor import ( + CAMERA_SENSORS, + LIGHT_SENSORS, + SENSE_SENSORS, +) +from homeassistant.components.unifiprotect.const import ( + DEFAULT_ATTRIBUTION, + RING_INTERVAL, +) +from homeassistant.const import ( + ATTR_ATTRIBUTION, + ATTR_LAST_TRIP_TIME, + STATE_OFF, + STATE_ON, + Platform, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er +from homeassistant.util.dt import utcnow + +from .conftest import ( + MockEntityFixture, + assert_entity_counts, + ids_from_device_description, + time_changed, +) + + +@pytest.fixture(name="camera") +async def camera_fixture( + hass: HomeAssistant, + mock_entry: MockEntityFixture, + mock_camera: Camera, + now: datetime, +): + """Fixture for a single camera for testing the binary_sensor platform.""" + + # disable pydantic validation so mocking can happen + Camera.__config__.validate_assignment = False + + camera_obj = mock_camera.copy(deep=True) + camera_obj._api = mock_entry.api + camera_obj.channels[0]._api = mock_entry.api + camera_obj.channels[1]._api = mock_entry.api + camera_obj.channels[2]._api = mock_entry.api + camera_obj.name = "Test Camera" + camera_obj.feature_flags.has_chime = True + camera_obj.last_ring = now - timedelta(hours=1) + camera_obj.is_dark = False + + mock_entry.api.bootstrap.reset_objects() + mock_entry.api.bootstrap.nvr.system_info.storage.devices = [] + mock_entry.api.bootstrap.cameras = { + camera_obj.id: camera_obj, + } + + await hass.config_entries.async_setup(mock_entry.entry.entry_id) + await hass.async_block_till_done() + + assert_entity_counts(hass, Platform.BINARY_SENSOR, 2, 2) + + yield camera_obj + + Camera.__config__.validate_assignment = True + + +@pytest.fixture(name="light") +async def light_fixture( + hass: HomeAssistant, mock_entry: MockEntityFixture, mock_light: Light, now: datetime +): + """Fixture for a single light for testing the binary_sensor platform.""" + + # disable pydantic validation so mocking can happen + Light.__config__.validate_assignment = False + + light_obj = mock_light.copy(deep=True) + light_obj._api = mock_entry.api + light_obj.name = "Test Light" + light_obj.is_dark = False + light_obj.is_pir_motion_detected = False + light_obj.last_motion = now - timedelta(hours=1) + + mock_entry.api.bootstrap.reset_objects() + mock_entry.api.bootstrap.nvr.system_info.storage.devices = [] + mock_entry.api.bootstrap.lights = { + light_obj.id: light_obj, + } + + await hass.config_entries.async_setup(mock_entry.entry.entry_id) + await hass.async_block_till_done() + + assert_entity_counts(hass, Platform.BINARY_SENSOR, 2, 2) + + yield light_obj + + Light.__config__.validate_assignment = True + + +@pytest.fixture(name="camera_none") +async def camera_none_fixture( + hass: HomeAssistant, mock_entry: MockEntityFixture, mock_camera: Camera +): + """Fixture for a single camera for testing the binary_sensor platform.""" + + # disable pydantic validation so mocking can happen + Camera.__config__.validate_assignment = False + + camera_obj = mock_camera.copy(deep=True) + camera_obj._api = mock_entry.api + camera_obj.channels[0]._api = mock_entry.api + camera_obj.channels[1]._api = mock_entry.api + camera_obj.channels[2]._api = mock_entry.api + camera_obj.name = "Test Camera" + camera_obj.feature_flags.has_chime = False + camera_obj.is_dark = False + + mock_entry.api.bootstrap.reset_objects() + mock_entry.api.bootstrap.nvr.system_info.storage.devices = [] + mock_entry.api.bootstrap.cameras = { + camera_obj.id: camera_obj, + } + + await hass.config_entries.async_setup(mock_entry.entry.entry_id) + await hass.async_block_till_done() + + assert_entity_counts(hass, Platform.BINARY_SENSOR, 1, 1) + + yield camera_obj + + Camera.__config__.validate_assignment = True + + +@pytest.fixture(name="sensor") +async def sensor_fixture( + hass: HomeAssistant, + mock_entry: MockEntityFixture, + mock_sensor: Sensor, + now: datetime, +): + """Fixture for a single sensor for testing the binary_sensor platform.""" + + # disable pydantic validation so mocking can happen + Sensor.__config__.validate_assignment = False + + sensor_obj = mock_sensor.copy(deep=True) + sensor_obj._api = mock_entry.api + sensor_obj.name = "Test Sensor" + sensor_obj.is_opened = False + sensor_obj.battery_status.is_low = False + sensor_obj.is_motion_detected = False + sensor_obj.motion_detected_at = now - timedelta(hours=1) + sensor_obj.open_status_changed_at = now - timedelta(hours=1) + + mock_entry.api.bootstrap.reset_objects() + mock_entry.api.bootstrap.nvr.system_info.storage.devices = [] + mock_entry.api.bootstrap.sensors = { + sensor_obj.id: sensor_obj, + } + + await hass.config_entries.async_setup(mock_entry.entry.entry_id) + await hass.async_block_till_done() + + assert_entity_counts(hass, Platform.BINARY_SENSOR, 3, 3) + + yield sensor_obj + + Sensor.__config__.validate_assignment = True + + +async def test_binary_sensor_setup_light( + hass: HomeAssistant, light: Light, now: datetime +): + """Test binary_sensor entity setup for light devices.""" + + entity_registry = er.async_get(hass) + + for index, description in enumerate(LIGHT_SENSORS): + unique_id, entity_id = ids_from_device_description( + Platform.BINARY_SENSOR, light, description + ) + + entity = entity_registry.async_get(entity_id) + assert entity + assert entity.unique_id == unique_id + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_OFF + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + + if index == 1: + assert state.attributes[ATTR_LAST_TRIP_TIME] == now - timedelta(hours=1) + + +async def test_binary_sensor_setup_camera_all( + hass: HomeAssistant, camera: Camera, now: datetime +): + """Test binary_sensor entity setup for camera devices (all features).""" + + entity_registry = er.async_get(hass) + + for index, description in enumerate(CAMERA_SENSORS): + unique_id, entity_id = ids_from_device_description( + Platform.BINARY_SENSOR, camera, description + ) + + entity = entity_registry.async_get(entity_id) + assert entity + assert entity.unique_id == unique_id + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_OFF + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + + if index == 0: + assert state.attributes[ATTR_LAST_TRIP_TIME] == now - timedelta(hours=1) + + +async def test_binary_sensor_setup_camera_none( + hass: HomeAssistant, + camera_none: Camera, +): + """Test binary_sensor entity setup for camera devices (no features).""" + + entity_registry = er.async_get(hass) + description = CAMERA_SENSORS[1] + + unique_id, entity_id = ids_from_device_description( + Platform.BINARY_SENSOR, camera_none, description + ) + + entity = entity_registry.async_get(entity_id) + assert entity + assert entity.unique_id == unique_id + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_OFF + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + + +async def test_binary_sensor_setup_sensor( + hass: HomeAssistant, sensor: Sensor, now: datetime +): + """Test binary_sensor entity setup for sensor devices.""" + + entity_registry = er.async_get(hass) + + expected_trip_time = now - timedelta(hours=1) + for index, description in enumerate(SENSE_SENSORS): + unique_id, entity_id = ids_from_device_description( + Platform.BINARY_SENSOR, sensor, description + ) + + entity = entity_registry.async_get(entity_id) + assert entity + assert entity.unique_id == unique_id + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_OFF + assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION + + if index != 1: + assert state.attributes[ATTR_LAST_TRIP_TIME] == expected_trip_time + + +async def test_binary_sensor_update_doorbell( + hass: HomeAssistant, + mock_entry: MockEntityFixture, + camera: Camera, +): + """Test select entity update (change doorbell message).""" + + _, entity_id = ids_from_device_description( + Platform.BINARY_SENSOR, camera, CAMERA_SENSORS[0] + ) + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_OFF + + new_bootstrap = copy(mock_entry.api.bootstrap) + new_camera = camera.copy() + new_camera.last_ring = utcnow() + + mock_msg = Mock() + mock_msg.changed_data = {} + mock_msg.new_obj = new_camera + + new_bootstrap.cameras = {new_camera.id: new_camera} + mock_entry.api.bootstrap = new_bootstrap + mock_entry.api.ws_subscription(mock_msg) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_ON + + # fire event a second time for code coverage (cancel existing) + mock_entry.api.ws_subscription(mock_msg) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_ON + + # since time is not really changing, switch the last ring back to allow turn off + new_camera.last_ring = utcnow() - RING_INTERVAL + await time_changed(hass, RING_INTERVAL.total_seconds()) + await hass.async_block_till_done() + state = hass.states.get(entity_id) + assert state + assert state.state == STATE_OFF