diff --git a/homeassistant/components/unifiprotect/binary_sensor.py b/homeassistant/components/unifiprotect/binary_sensor.py index 7e6342644237..88ae7da658cc 100644 --- a/homeassistant/components/unifiprotect/binary_sensor.py +++ b/homeassistant/components/unifiprotect/binary_sensor.py @@ -1,12 +1,13 @@ """Component providing binary sensors for UniFi Protect.""" -from collections.abc import Sequence +from collections.abc import Callable, Sequence import dataclasses import operator from typing import cast, override from uiprotect.data import ( NVR, + Fob, ModelType, MountType, ProtectAdoptableDeviceModel, @@ -26,6 +27,7 @@ from homeassistant.components.binary_sensor import ( ) from homeassistant.const import EntityCategory, Platform from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from .data import ProtectData, ProtectDeviceType, UFPConfigEntry @@ -36,6 +38,7 @@ from .entity import ( ProtectDeviceEntity, ProtectEntityDescription, ProtectEventMixin, + ProtectFobEntity, ProtectIsOnEntity, ProtectNVREntity, async_all_device_entities, @@ -727,6 +730,53 @@ def _async_nvr_entities( ] +def _fob_battery_low(fob: Fob) -> bool | None: + """Return whether the key fob battery is low, if it has been reported.""" + if (battery := fob.wireless_connection_state.battery_status) is not None: + return battery.is_low + return None + + +@dataclasses.dataclass(frozen=True, kw_only=True) +class ProtectFobBinaryEntityDescription(BinarySensorEntityDescription): + """Describes a UniFi Protect key fob binary sensor entity.""" + + value_fn: Callable[[Fob], bool | None] + + +FOB_BINARY_SENSORS: tuple[ProtectFobBinaryEntityDescription, ...] = ( + ProtectFobBinaryEntityDescription( + key="battery_low", + device_class=BinarySensorDeviceClass.BATTERY, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=_fob_battery_low, + ), +) + + +class ProtectFobBinarySensor(ProtectFobEntity, BinarySensorEntity): + """A binary sensor entity for a UniFi Protect key fob (Public API).""" + + entity_description: ProtectFobBinaryEntityDescription + _fob_state_attrs = ("_attr_available", "_attr_is_on") + + def __init__( + self, + data: ProtectData, + fob: Fob, + description: ProtectFobBinaryEntityDescription, + ) -> None: + """Initialize the key fob binary sensor.""" + self.entity_description = description + self._attr_unique_id = f"{fob.mac}_{description.key}" + super().__init__(data, fob) + + @callback + @override + def _async_update_from_fob(self, fob: Fob) -> None: + self._attr_is_on = self.entity_description.value_fn(fob) + + async def async_setup_entry( hass: HomeAssistant, entry: UFPConfigEntry, @@ -734,6 +784,34 @@ async def async_setup_entry( ) -> None: """Set up binary sensors for UniFi Protect integration.""" data = entry.runtime_data + + @callback + def _add_new_public_device(device: PublicDeviceModel) -> None: + if isinstance(device, Fob): + async_add_entities( + ProtectFobBinarySensor(data, device, description) + for description in FOB_BINARY_SENSORS + ) + + entry.async_on_unload( + async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device) + ) + + # The public bootstrap is primed only with an API key and supported NVR + # firmware; without it there are no fobs to expose. + api = data.api + if api.has_public_bootstrap: + async_add_entities( + ProtectFobBinarySensor(data, fob, description) + for fob in api.public_bootstrap.fobs.values() + for description in FOB_BINARY_SENSORS + ) + + # Everything below is driven by the private bootstrap, which public-only + # entries do not have. + if api.is_public_only: + return + async_remove_unsupported_sense_entities( hass, Platform.BINARY_SENSOR, data, (*SENSE_SENSORS, *MOUNTABLE_SENSE_SENSORS) ) diff --git a/homeassistant/components/unifiprotect/const.py b/homeassistant/components/unifiprotect/const.py index f8685a6f73f3..03927d100fa9 100644 --- a/homeassistant/components/unifiprotect/const.py +++ b/homeassistant/components/unifiprotect/const.py @@ -85,8 +85,11 @@ PLATFORMS = [ # rest enumerate from the private bootstrap, which is absent in this mode. PUBLIC_ONLY_PLATFORMS = [ Platform.ALARM_CONTROL_PANEL, + Platform.BINARY_SENSOR, Platform.CAMERA, + Platform.EVENT, Platform.LIGHT, + Platform.SENSOR, ] # Stored local-user credentials do not imply the mode: they are kept on a diff --git a/homeassistant/components/unifiprotect/entity.py b/homeassistant/components/unifiprotect/entity.py index c814336aa1e0..8f1bba6124ff 100644 --- a/homeassistant/components/unifiprotect/entity.py +++ b/homeassistant/components/unifiprotect/entity.py @@ -14,6 +14,7 @@ from uiprotect.data import ( NVR, DeviceState, Event, + Fob, ModelType, ProtectAdoptableDeviceModel, PublicDeviceModel, @@ -443,6 +444,91 @@ class ProtectNVREntity(BaseProtectEntity): ) +class ProtectFobEntity(Entity): + """Base class for UniFi Protect key fob (Public API) entities. + + A key fob is a public-only device: it lives in + ``ProtectApiClient.public_bootstrap.fobs`` and is refreshed over the public + devices websocket, so it does not use the private-device machinery in + :class:`BaseProtectEntity`. Availability follows the public websocket health + and the fob's presence in the bootstrap, mirroring the relay switch. + Subclasses fed by the events websocket set ``_ufp_requires_events_ws`` so + they also go unavailable when that stream drops. + """ + + _attr_should_poll = False + _attr_attribution = DEFAULT_ATTRIBUTION + _attr_has_entity_name = True + _ufp_requires_events_ws: bool = False + _fob_state_attrs: tuple[str, ...] = ("_attr_available",) + + def __init__(self, data: ProtectData, fob: Fob) -> None: + """Initialize the fob entity and prime its state from the bootstrap.""" + self.data = data + self._fob_id = fob.id + self._fob_mac = fob.mac + self._attr_device_info = DeviceInfo( + connections={(dr.CONNECTION_NETWORK_MAC, fob.mac)}, + identifiers={(DOMAIN, fob.mac)}, + manufacturer=DEFAULT_BRAND, + # A freshly-paired, unnamed fob reports ``name=None``; fall back to a + # stable default so the device is never registered nameless. + name=fob.name or f"Key Fob {fob.mac}", + model="Key Fob", + via_device_id=data.nvr_device_id, + ) + self._attr_available = self._async_public_available() + self._async_update_from_fob(fob) + + @property + def _fob(self) -> Fob | None: + """Return the cached fob from the public bootstrap, if still present.""" + api = self.data.api + if not api.has_public_bootstrap: + return None + return api.public_bootstrap.fobs.get(self._fob_id) + + @callback + def _async_public_available(self) -> bool: + """Return whether the streams backing this entity are healthy.""" + data = self.data + return data.last_public_update_success and ( + not self._ufp_requires_events_ws or data.last_events_update_success + ) + + @callback + def _async_update_from_fob(self, fob: Fob) -> None: + """Refresh entity state from the fob. Overridden by subclasses.""" + + @callback + def _async_updated(self, _obj: PublicDeviceModel | None) -> None: + """Handle a public devices WS update for this fob. + + The state is always re-read from the public bootstrap: the library + merges WS updates into it before dispatching, and ``None`` (a websocket + state change or a delete) carries no object to read. + """ + prev = [getattr(self, attr, None) for attr in self._fob_state_attrs] + if (fob := self._fob) is None: + self._attr_available = False + else: + self._attr_available = self._async_public_available() + self._async_update_from_fob(fob) + if [getattr(self, attr, None) for attr in self._fob_state_attrs] != prev: + self.async_write_ha_state() + + @override + async def async_added_to_hass(self) -> None: + """Subscribe to public devices WS updates dispatched by ProtectData.""" + await super().async_added_to_hass() + self.async_on_remove( + self.data.async_subscribe_public(self._fob_mac, self._async_updated) + ) + # Refresh from the bootstrap: an update or delete that landed between + # construction and this subscription would otherwise be missed. + self._async_updated(None) + + class EventEntityMixin(ProtectDeviceEntity): """Adds motion event attributes to sensor.""" diff --git a/homeassistant/components/unifiprotect/event.py b/homeassistant/components/unifiprotect/event.py index 6408865a8d04..3f3ccc585bc6 100644 --- a/homeassistant/components/unifiprotect/event.py +++ b/homeassistant/components/unifiprotect/event.py @@ -5,8 +5,9 @@ import re from typing import Any, override from uiprotect import ProtectEvent -from uiprotect.data import ModelType, SmartDetectObjectType +from uiprotect.data import Fob, ModelType, PublicDeviceModel, SmartDetectObjectType from uiprotect.data.nvr import Event, EventDetectedThumbnail +from uiprotect.data.types import EventButtonType from homeassistant.components.event import ( DoorbellEventType, @@ -15,6 +16,7 @@ from homeassistant.components.event import ( EventEntityDescription, ) from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback +from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.event import async_call_at @@ -41,7 +43,12 @@ from .data import ( ProtectDeviceType, UFPConfigEntry, ) -from .entity import EventEntityMixin, ProtectDeviceEntity, ProtectEventMixin +from .entity import ( + EventEntityMixin, + ProtectDeviceEntity, + ProtectEventMixin, + ProtectFobEntity, +) PARALLEL_UPDATES = 0 @@ -102,26 +109,17 @@ class ProtectDetectionEventEntityDescription(ProtectEventEntityDescription): include_event_source: bool = False -class ProtectDevicePublicEventEntity( - EventEntityMixin, ProtectDeviceEntity, EventEntity -): - """Base for entities driven by the public events WS. +class ProtectFireOnceMixin(EventEntity): + """Dedup mixin for entities fired from the public events WS. A detection type can surface at the event start, on a later update, or only - as the event ends, and every non-eviction change is dispatched — so firing is - deduped per ``(event id, object type, event source)``. - - Availability follows the public API (device present and connected) plus the - events websocket, which is the only channel these entities fire from. + as the event ends, and every non-eviction change is dispatched, so firing is + deduped per ``(event id, surfaced type, Protect event type)``. """ - _ufp_uses_public = True - _ufp_requires_events_ws = True - - entity_description: ProtectEventEntityDescription - # A camera can run two overlapping events of the same category whose - # dispatches interleave, so dedup tracks fired object/source pairs per - # recent event id (bounded), not just the current one. + # A device can run two overlapping events of the same category whose + # dispatches interleave, so dedup tracks fired type pairs per recent event + # id (bounded), not just the current one. _fired: dict[str, frozenset[tuple[str, EventType]]] | None = None @callback @@ -148,6 +146,21 @@ class ProtectDevicePublicEventEntity( self.async_write_ha_state() +class ProtectDevicePublicEventEntity( + ProtectFireOnceMixin, EventEntityMixin, ProtectDeviceEntity, EventEntity +): + """Base for entities driven by the public events WS. + + Availability follows the public API (device present and connected) plus the + events websocket, which is the only channel these entities fire from. + """ + + _ufp_uses_public = True + _ufp_requires_events_ws = True + + entity_description: ProtectEventEntityDescription + + class ProtectDeviceRingEventEntity(ProtectDevicePublicEventEntity): """A UniFi Protect doorbell ring event entity driven by the public events WS.""" @@ -565,6 +578,60 @@ class ProtectDeviceMotionEventEntity(ProtectDeviceDetectionEventEntity): self._fire_once(event, EventType.MOTION.value, {ATTR_EVENT_ID: event.id}) +# Real hardware reports an empty ``feature_flags.buttons``, so the whole +# vocabulary is declared. Sourced from the enum matched against +# ``metadata.button`` below so the two cannot drift apart. +_FOB_EVENT_TYPES: list[str] = [ + button.name.lower() + for button in EventButtonType + if button is not EventButtonType.UNKNOWN +] + + +class ProtectFobButtonEventEntity(ProtectFireOnceMixin, ProtectFobEntity, EventEntity): + """A UniFi Protect key fob button-press event entity. + + Each fob exposes one event entity that fires the pressed button (from a + public ``sensorButtonPressed`` event's ``metadata.button``) as its event + type. + """ + + _attr_translation_key = "keyfob" + _attr_event_types = _FOB_EVENT_TYPES + # Presses arrive only on the events websocket, so its health gates + # availability on top of the devices websocket. + _ufp_requires_events_ws = True + + def __init__(self, data: ProtectData, fob: Fob) -> None: + """Initialize the key fob button event entity.""" + self._attr_unique_id = f"{fob.mac}_keyfob" + super().__init__(data, fob) + + @override + async def async_added_to_hass(self) -> None: + """Subscribe to public key-fob button-press events.""" + await super().async_added_to_hass() + # A press arrives as a ``sensorButtonPressed`` event whose ``device`` is + # the fob and whose ``metadata.button`` is the pressed button. + self.async_on_remove( + self.data.async_subscribe_public_event( + self._fob_id, EventType.SENSOR_BUTTON_PRESSED, self._async_button_event + ) + ) + + @callback + def _async_button_event(self, event: ProtectEvent) -> None: + if (metadata := event.metadata) is None or (button := metadata.button) is None: + return + # Skip a button added by newer firmware that coerces to + # ``EventButtonType.UNKNOWN`` (not among the declared event types). + if (button_type := button.name.lower()) not in self.event_types: + return + # A press is dispatched on every non-eviction change to its event, so + # the same press can arrive more than once. + self._fire_once(event, button_type, {ATTR_EVENT_ID: event.id}) + + EVENT_DESCRIPTIONS: tuple[ProtectEventEntityDescription, ...] = ( ProtectEventEntityDescription( key="doorbell", @@ -658,6 +725,29 @@ async def async_setup_entry( """Set up event entities for UniFi Protect integration.""" data = entry.runtime_data + @callback + def _add_new_public_device(device: PublicDeviceModel) -> None: + if isinstance(device, Fob): + async_add_entities([ProtectFobButtonEventEntity(data, device)]) + + entry.async_on_unload( + async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device) + ) + + # The public bootstrap is primed only with an API key and supported NVR + # firmware; without it there are no fobs to expose. + api = data.api + if api.has_public_bootstrap: + async_add_entities( + ProtectFobButtonEventEntity(data, fob) + for fob in api.public_bootstrap.fobs.values() + ) + + # Everything below is driven by the private bootstrap, which public-only + # entries do not have. + if api.is_public_only: + return + @callback def _add_new_device(device: ProtectAdoptableDeviceModel) -> None: # AiPort inherits from Camera but should not create camera-specific entities diff --git a/homeassistant/components/unifiprotect/icons.json b/homeassistant/components/unifiprotect/icons.json index d027fca002c2..6a4bbad7f7da 100644 --- a/homeassistant/components/unifiprotect/icons.json +++ b/homeassistant/components/unifiprotect/icons.json @@ -181,6 +181,9 @@ "fingerprint": { "default": "mdi:fingerprint" }, + "keyfob": { + "default": "mdi:remote" + }, "motion_detection": { "default": "mdi:motion-sensor" }, @@ -274,6 +277,9 @@ "doorbell_text": { "default": "mdi:card-text" }, + "fob_status": { + "default": "mdi:remote" + }, "infrared_mode": { "default": "mdi:circle-opacity" }, diff --git a/homeassistant/components/unifiprotect/sensor.py b/homeassistant/components/unifiprotect/sensor.py index cb3896cbd2c4..c53b7125f71c 100644 --- a/homeassistant/components/unifiprotect/sensor.py +++ b/homeassistant/components/unifiprotect/sensor.py @@ -10,6 +10,8 @@ from typing import Any, cast, override from uiprotect.data import ( NVR, Camera, + Fob, + FobAwayState, Light, ModelType, ProtectAdoptableDeviceModel, @@ -42,6 +44,7 @@ from homeassistant.const import ( UnitOfTime, ) from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from .data import ProtectData, ProtectDeviceType, UFPConfigEntry @@ -52,6 +55,7 @@ from .entity import ( ProtectDeviceEntity, ProtectEntityDescription, ProtectEventMixin, + ProtectFobEntity, ProtectNVREntity, T, async_all_device_entities, @@ -591,6 +595,91 @@ _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = { } +def _fob_battery_level(fob: Fob) -> int | None: + """Return the key fob battery percentage, if it has been reported.""" + if (battery := fob.wireless_connection_state.battery_status) is not None: + return battery.percentage + return None + + +def _fob_signal_strength(fob: Fob) -> int | None: + """Return the key fob Bluetooth signal strength, if it has been reported.""" + if (signal := fob.wireless_connection_state.signal_state) is not None: + return signal.signal_strength + return None + + +def _fob_status(fob: Fob) -> str | None: + """Return the key fob presence state. + + ``FobAwayState`` carries an ``UNKNOWN`` member that the library coerces + unrecognized wire values into; map it to ``None`` (unknown state) rather + than a value the enum sensor's ``options`` do not list. + """ + if (away_state := fob.away_state) is FobAwayState.UNKNOWN: + return None + return away_state.value.lower() + + +@dataclass(frozen=True, kw_only=True) +class ProtectFobSensorEntityDescription(SensorEntityDescription): + """Describes a UniFi Protect key fob sensor entity.""" + + value_fn: Callable[[Fob], int | str | None] + + +FOB_SENSORS: tuple[ProtectFobSensorEntityDescription, ...] = ( + ProtectFobSensorEntityDescription( + key="battery_level", + device_class=SensorDeviceClass.BATTERY, + native_unit_of_measurement=PERCENTAGE, + entity_category=EntityCategory.DIAGNOSTIC, + state_class=SensorStateClass.MEASUREMENT, + value_fn=_fob_battery_level, + ), + ProtectFobSensorEntityDescription( + key="signal_strength", + device_class=SensorDeviceClass.SIGNAL_STRENGTH, + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, + state_class=SensorStateClass.MEASUREMENT, + value_fn=_fob_signal_strength, + ), + ProtectFobSensorEntityDescription( + key="status", + translation_key="fob_status", + device_class=SensorDeviceClass.ENUM, + entity_category=EntityCategory.DIAGNOSTIC, + options=["online", "recently_seen", "no_recent_heartbeat", "device_lost"], + value_fn=_fob_status, + ), +) + + +class ProtectFobSensor(ProtectFobEntity, SensorEntity): + """A sensor entity for a UniFi Protect key fob (Public API).""" + + entity_description: ProtectFobSensorEntityDescription + _fob_state_attrs = ("_attr_available", "_attr_native_value") + + def __init__( + self, + data: ProtectData, + fob: Fob, + description: ProtectFobSensorEntityDescription, + ) -> None: + """Initialize the key fob sensor.""" + self.entity_description = description + self._attr_unique_id = f"{fob.mac}_{description.key}" + super().__init__(data, fob) + + @callback + @override + def _async_update_from_fob(self, fob: Fob) -> None: + self._attr_native_value = self.entity_description.value_fn(fob) + + async def async_setup_entry( hass: HomeAssistant, entry: UFPConfigEntry, @@ -598,6 +687,34 @@ async def async_setup_entry( ) -> None: """Set up sensors for UniFi Protect integration.""" data = entry.runtime_data + + @callback + def _add_new_public_device(device: PublicDeviceModel) -> None: + if isinstance(device, Fob): + async_add_entities( + ProtectFobSensor(data, device, description) + for description in FOB_SENSORS + ) + + entry.async_on_unload( + async_dispatcher_connect(hass, data.public_add_signal, _add_new_public_device) + ) + + # The public bootstrap is primed only with an API key and supported NVR + # firmware; without it there are no fobs to expose. + api = data.api + if api.has_public_bootstrap: + async_add_entities( + ProtectFobSensor(data, fob, description) + for fob in api.public_bootstrap.fobs.values() + for description in FOB_SENSORS + ) + + # Everything below is driven by the private bootstrap, which public-only + # entries do not have. + if api.is_public_only: + return + async_remove_unsupported_sense_entities(hass, Platform.SENSOR, data, SENSE_SENSORS) @callback diff --git a/homeassistant/components/unifiprotect/strings.json b/homeassistant/components/unifiprotect/strings.json index d8776e546a14..49272c1732af 100644 --- a/homeassistant/components/unifiprotect/strings.json +++ b/homeassistant/components/unifiprotect/strings.json @@ -365,6 +365,26 @@ } } }, + "keyfob": { + "name": "Button", + "state_attributes": { + "event_type": { + "state": { + "alarm_hub_button": "Alarm hub button", + "arm": "Arm", + "disarm": "Disarm", + "function": "Function", + "input1": "Input 1", + "input2": "Input 2", + "left": "Left", + "main": "Main", + "night": "Night", + "panic": "Panic", + "right": "Right" + } + } + } + }, "motion_detection": { "name": "Motion detection", "state_attributes": { @@ -586,6 +606,15 @@ "doorbell_text": { "name": "[%key:component::unifiprotect::entity::select::doorbell_text::name%]" }, + "fob_status": { + "name": "Status", + "state": { + "device_lost": "Device lost", + "no_recent_heartbeat": "No recent heartbeat", + "online": "Online", + "recently_seen": "Recently seen" + } + }, "infrared_mode": { "name": "[%key:component::unifiprotect::entity::select::infrared_mode::name%]" }, diff --git a/tests/components/unifiprotect/conftest.py b/tests/components/unifiprotect/conftest.py index 71b5c2ae3979..a265e0d262d5 100644 --- a/tests/components/unifiprotect/conftest.py +++ b/tests/components/unifiprotect/conftest.py @@ -199,6 +199,7 @@ def mock_ufp_client(bootstrap: Bootstrap): client.public_bootstrap.lights = {} client.public_bootstrap.relays = {} client.public_bootstrap.sirens = {} + client.public_bootstrap.fobs = {} client.public_bootstrap.arm_profiles = {} client.public_bootstrap.arm_mode = None client.public_bootstrap.nvr = Mock() @@ -229,6 +230,7 @@ def mock_ufp_client(bootstrap: Bootstrap): yield from pb.lights.values() yield from pb.relays.values() yield from pb.sirens.values() + yield from pb.fobs.values() client.public_bootstrap.all_devices = _public_all_devices @@ -652,7 +654,12 @@ def mock_ufp_public_only_client() -> Mock: # them, so both helpers below read the attribute at call time. pb.cameras = {} pb.lights = {} - device_maps = {ModelType.CAMERA: "cameras", ModelType.LIGHT: "lights"} + pb.fobs = {} + device_maps = { + ModelType.CAMERA: "cameras", + ModelType.LIGHT: "lights", + ModelType.FOB: "fobs", + } def _all_devices(*, include_nvr: bool = False) -> Iterator[Mock]: if include_nvr and pb.nvr is not None: diff --git a/tests/components/unifiprotect/snapshots/test_fob.ambr b/tests/components/unifiprotect/snapshots/test_fob.ambr new file mode 100644 index 000000000000..d9038e1cccac --- /dev/null +++ b/tests/components/unifiprotect/snapshots/test_fob.ambr @@ -0,0 +1,16 @@ +# serializer version: 1 +# name: test_fob_button_event_types + list([ + 'function', + 'alarm_hub_button', + 'arm', + 'disarm', + 'night', + 'panic', + 'left', + 'right', + 'input1', + 'input2', + 'main', + ]) +# --- diff --git a/tests/components/unifiprotect/test_alarm_control_panel.py b/tests/components/unifiprotect/test_alarm_control_panel.py index 5db8d8048644..2c2dc41d8e5d 100644 --- a/tests/components/unifiprotect/test_alarm_control_panel.py +++ b/tests/components/unifiprotect/test_alarm_control_panel.py @@ -46,6 +46,7 @@ def _make_public_bootstrap(arm_mode: Mock | None) -> Mock: pb.arm_profiles = {} pb.relays = {} pb.sirens = {} + pb.fobs = {} return pb diff --git a/tests/components/unifiprotect/test_fob.py b/tests/components/unifiprotect/test_fob.py new file mode 100644 index 000000000000..b8cec638d26d --- /dev/null +++ b/tests/components/unifiprotect/test_fob.py @@ -0,0 +1,636 @@ +"""Tests for the UniFi Protect key fob (Public API) entities.""" + +from datetime import datetime +from unittest.mock import AsyncMock, Mock + +import pytest +from syrupy.assertion import SnapshotAssertion +from uiprotect import EventChange, ProtectEvent, ProtectEventChannel +from uiprotect.data import ( + DeviceState, + EventType, + Fob, + FobAwayState, + FobButton, + ModelType, + PublicBootstrap, + PublicFobFeatureFlags, + WSAction, +) +from uiprotect.data.public_devices import ( + PublicSignalState, + PublicWirelessBatteryStatus, + PublicWirelessConnectionState, +) +from uiprotect.data.public_event import PublicEventMetadata +from uiprotect.data.types import EventButtonType +from uiprotect.websocket import WebsocketState + +from homeassistant.components.unifiprotect.const import ATTR_EVENT_ID +from homeassistant.const import ATTR_ATTRIBUTION, STATE_UNAVAILABLE +from homeassistant.core import Event as HAEvent, HomeAssistant, callback +from homeassistant.helpers import device_registry as dr, entity_registry as er +from homeassistant.helpers.event import async_track_state_change_event + +from .utils import MockUFPFixture, enable_entity, init_entry, public_device_ws_message + +FOB_ID = "fob-id-1" +FOB_MAC = "AA:BB:CC:DD:EE:F0" +FOB_NAME = "Front Door Fob" + +BATTERY_SENSOR = "sensor.front_door_fob_battery" +SIGNAL_SENSOR = "sensor.front_door_fob_signal_strength" +STATUS_SENSOR = "sensor.front_door_fob_status" +BATTERY_LOW_BINARY = "binary_sensor.front_door_fob_battery" +BUTTON_EVENT = "event.front_door_fob_button" + + +def _make_fob( + *, + buttons: list[FobButton] | None = None, + away_state: FobAwayState = FobAwayState.ONLINE, + percentage: int | None = 80, + is_low: bool = False, + signal_strength: int | None = -55, + state: DeviceState = DeviceState.CONNECTED, + name: str | None = FOB_NAME, +) -> Mock: + """Build a mock :class:`Fob` backed by real public sub-models.""" + fob = Mock(spec=Fob) + fob.id = FOB_ID + fob.mac = FOB_MAC + fob.name = name + fob.model = ModelType.FOB + fob.state = state + fob.away_state = away_state + # Real USL-FOB hardware reports an empty featureFlags.buttons. + fob.feature_flags = PublicFobFeatureFlags(buttons=buttons or []) + fob.wireless_connection_state = PublicWirelessConnectionState( + battery_status=PublicWirelessBatteryStatus( + percentage=percentage, is_low=is_low + ), + signal_state=PublicSignalState( + signal_strength=signal_strength, signal_quality=None + ), + ) + return fob + + +def _make_public_bootstrap(fob: Mock | None) -> Mock: + """Build a public bootstrap mock holding the given fob.""" + pb = Mock(spec=PublicBootstrap) + pb.fobs = {fob.id: fob} if fob is not None else {} + pb.cameras = {} + pb.lights = {} + pb.relays = {} + pb.sirens = {} + pb.arm_mode = None + pb.arm_profiles = {} + pb.nvr = Mock() + pb.nvr.mac = "aa:bb:cc:dd:ee:ff" + pb.nvr.name = "Test NVR" + pb.nvr.display_name = "Test NVR" + pb.nvr.device_type = None + pb.nvr.type = None + + # The baseline and reconnect resync enumerate all_devices(); a fob missing + # from it would be redispatched as new on every reconnect. + def _all_devices(*, include_nvr: bool = False) -> list[Mock]: + devices = list(pb.fobs.values()) + return [pb.nvr, *devices] if include_nvr else devices + + pb.all_devices = _all_devices + return pb + + +@pytest.fixture(name="ufp_with_fob") +def _ufp_with_fob(ufp: MockUFPFixture) -> tuple[MockUFPFixture, Mock]: + """Configure the ufp fixture with a single key fob on the public API.""" + fob = _make_fob() + ufp.api.has_public_bootstrap = True + ufp.api.public_bootstrap = _make_public_bootstrap(fob) + return ufp, fob + + +def _button_event( + fob: Mock, + *, + event_id: str = "evt-1", + button: EventButtonType | None = EventButtonType.ARM, + now: datetime | None = None, +) -> ProtectEvent: + """Build a button-press ``ProtectEvent`` matching a real USL-FOB capture. + + A real press is born-closed (``start == end``) and carries the fob itself as + the event ``device`` with the pressed button in ``metadata.button``. Pass + ``button=None`` to omit metadata entirely. + """ + when = now or datetime(2026, 1, 1) + return ProtectEvent( + id=event_id, + type=EventType.SENSOR_BUTTON_PRESSED, + channel=ProtectEventChannel.SENSOR, + device_id=fob.id, + device_mac=fob.mac, + start=when, + end=when, + metadata=None if button is None else PublicEventMetadata(button=button), + ) + + +async def test_fob_not_created_without_public_bootstrap( + hass: HomeAssistant, ufp: MockUFPFixture +) -> None: + """No fob entities are created when the public bootstrap is unavailable.""" + ufp.api.has_public_bootstrap = False + await init_entry(hass, ufp, []) + + assert hass.states.get(BATTERY_SENSOR) is None + assert hass.states.get(BUTTON_EVENT) is None + + +async def test_fob_entities_created( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + device_registry: dr.DeviceRegistry, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """The fob device and its entities are created from the public bootstrap.""" + ufp, _fob = ufp_with_fob + await init_entry(hass, ufp, []) + + device = device_registry.async_get_device_by_connection( + (dr.CONNECTION_NETWORK_MAC, FOB_MAC), ufp.entry.entry_id + ) + assert device is not None + assert device.name == FOB_NAME + + battery = entity_registry.async_get(BATTERY_SENSOR) + assert battery is not None + assert battery.unique_id == f"{FOB_MAC}_battery_level" + state = hass.states.get(BATTERY_SENSOR) + assert state is not None + assert state.state == "80" + assert state.attributes[ATTR_ATTRIBUTION] + + # Signal strength is a diagnostic sensor, disabled by default. + assert hass.states.get(SIGNAL_SENSOR) is None + signal = entity_registry.async_get(SIGNAL_SENSOR) + assert signal is not None + assert signal.disabled + + status = hass.states.get(STATUS_SENSOR) + assert status is not None + assert status.state == "online" + + battery_low = hass.states.get(BATTERY_LOW_BINARY) + assert battery_low is not None + assert battery_low.state == "off" + + +async def test_fob_button_event_types( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], + snapshot: SnapshotAssertion, +) -> None: + """The button event entity declares the full button vocabulary. + + Real hardware reports an empty ``feature_flags.buttons``, so the entity + cannot derive its types from the device and declares them all. + """ + ufp, _fob = ufp_with_fob + await init_entry(hass, ufp, []) + + state = hass.states.get(BUTTON_EVENT) + assert state is not None + assert state.attributes["event_types"] == snapshot + + +@pytest.mark.parametrize( + ("button", "expected_event_type"), + [ + pytest.param(EventButtonType.ARM, "arm", id="arm"), + pytest.param(EventButtonType.DISARM, "disarm", id="disarm"), + pytest.param(EventButtonType.PANIC, "panic", id="panic"), + pytest.param( + EventButtonType.ALARM_HUB_BUTTON, + "alarm_hub_button", + id="camelcase_maps_to_snake_case", + ), + ], +) +async def test_fob_button_press_fires_event( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], + button: EventButtonType, + expected_event_type: str, +) -> None: + """A sensor button-press fires an event of the matching snake_case type.""" + ufp, fob = ufp_with_fob + await init_entry(hass, ufp, []) + + events: list[HAEvent] = [] + + @callback + def _capture(event: HAEvent) -> None: + events.append(event) + + unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture) + + ufp.events_msg(_button_event(fob, button=button), EventChange.STARTED) + await hass.async_block_till_done() + + assert len(events) == 1 + new_state = events[0].data["new_state"] + assert new_state.attributes["event_type"] == expected_event_type + assert new_state.attributes[ATTR_EVENT_ID] == "evt-1" + unsub() + + +@pytest.mark.parametrize( + "button", + [ + pytest.param(EventButtonType.UNKNOWN, id="unknown_button"), + pytest.param(None, id="no_metadata"), + ], +) +async def test_fob_button_press_ignored( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], + button: EventButtonType | None, +) -> None: + """A press for an unknown button, or one with no metadata, does not fire.""" + ufp, fob = ufp_with_fob + await init_entry(hass, ufp, []) + + events: list[HAEvent] = [] + + @callback + def _capture(event: HAEvent) -> None: + events.append(event) + + unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture) + + ufp.events_msg(_button_event(fob, button=button), EventChange.STARTED) + await hass.async_block_till_done() + + assert len(events) == 0 + unsub() + + +async def test_fob_battery_updates_from_public_ws( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """A public devices WS update for the fob refreshes the battery sensor.""" + ufp, fob = ufp_with_fob + await init_entry(hass, ufp, []) + + assert hass.states.get(BATTERY_SENSOR).state == "80" + + fob.wireless_connection_state = PublicWirelessConnectionState( + battery_status=PublicWirelessBatteryStatus(percentage=42, is_low=False), + signal_state=PublicSignalState(signal_strength=-55, signal_quality=None), + ) + + mock_msg = Mock() + mock_msg.changed_data = {} + mock_msg.old_obj = fob + mock_msg.new_obj = fob + assert ufp.devices_ws_subscription is not None + ufp.devices_ws_subscription(mock_msg) + await hass.async_block_till_done() + + assert hass.states.get(BATTERY_SENSOR).state == "42" + + +async def test_fob_unavailable_on_public_ws_disconnect( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """Fob entities go unavailable when the public websocket disconnects.""" + ufp, _fob = ufp_with_fob + await init_entry(hass, ufp, []) + + assert hass.states.get(BATTERY_SENSOR).state == "80" + + assert ufp.devices_ws_state_subscription is not None + ufp.devices_ws_state_subscription(WebsocketState.DISCONNECTED) + await hass.async_block_till_done() + + assert hass.states.get(BATTERY_SENSOR).state == STATE_UNAVAILABLE + + +async def test_fob_present_at_startup_not_added_twice( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + ufp: MockUFPFixture, + caplog: pytest.LogCaptureFixture, +) -> None: + """A fob known at setup is not added a second time by a later add frame. + + The startup baseline is taken from ``all_devices()``; a fob missing from it + is treated as new when any add frame for it arrives (a re-delivered frame, + or a re-pair), which would clash on unique_id. + """ + fob = _make_fob() + ufp.api.is_public_only = True + ufp.api.has_public_bootstrap = True + pb = _make_public_bootstrap(fob) + ufp.api.public_bootstrap = pb + ufp.api.update_public = AsyncMock(return_value=pb) + + await init_entry(hass, ufp, []) + assert hass.states.get(BATTERY_SENSOR).state == "80" + + msg = public_device_ws_message(fob) + msg.action = WSAction.ADD + assert ufp.devices_ws_subscription is not None + ufp.devices_ws_subscription(msg) + await hass.async_block_till_done() + + assert "already exists" not in caplog.text + assert ( + len( + [ + entry + for entry in entity_registry.entities.values() + if entry.unique_id.startswith(FOB_MAC) + ] + ) + == 5 + ) + + +async def test_fob_button_dedup_across_dispatches( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """A press fires once even though every change to its event is dispatched.""" + ufp, fob = ufp_with_fob + await init_entry(hass, ufp, []) + + events: list[HAEvent] = [] + + @callback + def _capture(event: HAEvent) -> None: + events.append(event) + + unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture) + for change in (EventChange.STARTED, EventChange.UPDATED, EventChange.ENDED): + ufp.events_msg(_button_event(fob), change) + await hass.async_block_till_done() + unsub() + + assert len(events) == 1 + + # A genuinely new press on the same button still fires. + unsub = async_track_state_change_event(hass, BUTTON_EVENT, _capture) + ufp.events_msg(_button_event(fob, event_id="evt-2"), EventChange.STARTED) + await hass.async_block_till_done() + unsub() + + assert len(events) == 2 + + +async def test_fob_button_unavailable_on_events_ws_disconnect( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """Only the button entity follows the events websocket. + + Presses arrive solely on the events stream, so the event entity must not + stay available while it is down; the sensors are fed by the devices + stream and are unaffected. + """ + ufp, _fob = ufp_with_fob + await init_entry(hass, ufp, []) + + assert hass.states.get(BUTTON_EVENT).state != STATE_UNAVAILABLE + + assert ufp.events_ws_state_subscription is not None + ufp.events_ws_state_subscription(WebsocketState.DISCONNECTED) + await hass.async_block_till_done() + + assert hass.states.get(BUTTON_EVENT).state == STATE_UNAVAILABLE + assert hass.states.get(BATTERY_SENSOR).state == "80" + + ufp.events_ws_state_subscription(WebsocketState.CONNECTED) + await hass.async_block_till_done() + + assert hass.states.get(BUTTON_EVENT).state != STATE_UNAVAILABLE + + +async def test_fob_battery_none_is_unknown( + hass: HomeAssistant, + ufp: MockUFPFixture, +) -> None: + """A freshly-paired fob reporting null battery yields an unknown state.""" + fob = _make_fob(percentage=None) + ufp.api.has_public_bootstrap = True + ufp.api.public_bootstrap = _make_public_bootstrap(fob) + + await init_entry(hass, ufp, []) + + state = hass.states.get(BATTERY_SENSOR) + assert state is not None + assert state.state == "unknown" + + +async def test_fob_signal_sensor_when_enabled( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """Enabling the signal-strength sensor exposes the fob's signal value.""" + ufp, _fob = ufp_with_fob + await init_entry(hass, ufp, []) + + await enable_entity(hass, ufp.entry.entry_id, SIGNAL_SENSOR) + + state = hass.states.get(SIGNAL_SENSOR) + assert state is not None + assert state.state == "-55" + + +async def test_fob_event_entity_created_with_empty_feature_flags( + hass: HomeAssistant, + ufp: MockUFPFixture, +) -> None: + """A fob with empty feature_flags (real hardware) still gets an event entity.""" + fob = _make_fob(buttons=[]) + ufp.api.has_public_bootstrap = True + ufp.api.public_bootstrap = _make_public_bootstrap(fob) + + await init_entry(hass, ufp, []) + + state = hass.states.get(BUTTON_EVENT) + assert state is not None + assert "arm" in state.attributes["event_types"] + + +async def test_fob_status_reflects_away_state( + hass: HomeAssistant, + ufp: MockUFPFixture, +) -> None: + """The status sensor reflects a lost fob's away state.""" + fob = _make_fob(away_state=FobAwayState.DEVICE_LOST) + ufp.api.has_public_bootstrap = True + ufp.api.public_bootstrap = _make_public_bootstrap(fob) + + await init_entry(hass, ufp, []) + + state = hass.states.get(STATUS_SENSOR) + assert state is not None + assert state.state == "device_lost" + + +async def test_fob_unknown_away_state_is_unknown( + hass: HomeAssistant, + ufp: MockUFPFixture, +) -> None: + """An unrecognized away state maps to an unknown status, not a ValueError.""" + fob = _make_fob(away_state=FobAwayState.UNKNOWN) + ufp.api.has_public_bootstrap = True + ufp.api.public_bootstrap = _make_public_bootstrap(fob) + + await init_entry(hass, ufp, []) + + state = hass.states.get(STATUS_SENSOR) + assert state is not None + assert state.state == "unknown" + + +async def test_fob_unavailable_when_removed_from_bootstrap( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """A fob deleted from the public bootstrap marks its entities unavailable.""" + ufp, fob = ufp_with_fob + await init_entry(hass, ufp, []) + + assert hass.states.get(BATTERY_SENSOR).state == "80" + + # Delete event: the library removes the object before dispatching ``None``. + del ufp.api.public_bootstrap.fobs[fob.id] + mock_msg = Mock() + mock_msg.old_obj = fob + mock_msg.new_obj = None + assert ufp.devices_ws_subscription is not None + ufp.devices_ws_subscription(mock_msg) + await hass.async_block_till_done() + + state = hass.states.get(BATTERY_SENSOR) + assert state is not None + assert state.state == STATE_UNAVAILABLE + + +async def test_fob_without_wireless_data_is_unknown( + hass: HomeAssistant, + ufp: MockUFPFixture, +) -> None: + """A freshly-paired fob with no battery/signal reported reads as unknown.""" + fob = _make_fob() + fob.wireless_connection_state = PublicWirelessConnectionState( + battery_status=None, signal_state=None + ) + ufp.api.has_public_bootstrap = True + ufp.api.public_bootstrap = _make_public_bootstrap(fob) + + await init_entry(hass, ufp, []) + + assert hass.states.get(BATTERY_SENSOR).state == "unknown" + assert hass.states.get(BATTERY_LOW_BINARY).state == "unknown" + + await enable_entity(hass, ufp.entry.entry_id, SIGNAL_SENSOR) + assert hass.states.get(SIGNAL_SENSOR).state == "unknown" + + +async def test_fob_unavailable_when_public_bootstrap_lost( + hass: HomeAssistant, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """Losing the public bootstrap marks fob entities unavailable.""" + ufp, fob = ufp_with_fob + await init_entry(hass, ufp, []) + + assert hass.states.get(BATTERY_SENSOR).state == "80" + + ufp.api.has_public_bootstrap = False + mock_msg = Mock() + mock_msg.changed_data = {} + mock_msg.old_obj = fob + mock_msg.new_obj = fob + assert ufp.devices_ws_subscription is not None + ufp.devices_ws_subscription(mock_msg) + await hass.async_block_till_done() + + state = hass.states.get(BATTERY_SENSOR) + assert state is not None + assert state.state == STATE_UNAVAILABLE + + +async def test_fob_entity_counts( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + ufp_with_fob: tuple[MockUFPFixture, Mock], +) -> None: + """Exactly the expected fob entities are created across platforms.""" + ufp, _fob = ufp_with_fob + await init_entry(hass, ufp, []) + + fob_entities = [ + entry + for entry in entity_registry.entities.values() + if entry.unique_id.startswith(FOB_MAC) + ] + # battery sensor, signal sensor, status sensor, battery-low binary, button event + assert len(fob_entities) == 5 + assert sum(not entry.disabled for entry in fob_entities) == 4 + + +async def test_fob_added_at_runtime( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + ufp: MockUFPFixture, + caplog: pytest.LogCaptureFixture, +) -> None: + """A fob paired after setup is discovered from its public add frame.""" + ufp.api.is_public_only = True + ufp.api.has_public_bootstrap = True + pb = _make_public_bootstrap(None) + ufp.api.public_bootstrap = pb + ufp.api.update_public = AsyncMock(return_value=pb) + + await init_entry(hass, ufp, []) + assert hass.states.get(BATTERY_SENSOR) is None + + fob = _make_fob() + pb.fobs = {fob.id: fob} + msg = public_device_ws_message(fob) + msg.action = WSAction.ADD + assert ufp.devices_ws_subscription is not None + ufp.devices_ws_subscription(msg) + await hass.async_block_till_done() + + assert hass.states.get(BATTERY_SENSOR).state == "80" + assert hass.states.get(BATTERY_LOW_BINARY).state == "off" + assert hass.states.get(BUTTON_EVENT) is not None + + # A re-delivered add frame is deduped before the platforms see it, so no + # duplicate unique_id error is logged. + msg = public_device_ws_message(fob) + msg.action = WSAction.ADD + ufp.devices_ws_subscription(msg) + await hass.async_block_till_done() + + assert "already exists" not in caplog.text + assert ( + len( + [ + entry + for entry in entity_registry.entities.values() + if entry.unique_id.startswith(FOB_MAC) + ] + ) + == 5 + ) diff --git a/tests/components/unifiprotect/test_relay.py b/tests/components/unifiprotect/test_relay.py index 28608fefdac6..f84c892868ba 100644 --- a/tests/components/unifiprotect/test_relay.py +++ b/tests/components/unifiprotect/test_relay.py @@ -80,6 +80,7 @@ def _make_public_bootstrap(relay: Mock | None) -> Mock: pb.arm_mode = None pb.arm_profiles = {} pb.sirens = {} + pb.fobs = {} return pb diff --git a/tests/components/unifiprotect/test_select.py b/tests/components/unifiprotect/test_select.py index 830b494314d5..8436315d6cee 100644 --- a/tests/components/unifiprotect/test_select.py +++ b/tests/components/unifiprotect/test_select.py @@ -1024,6 +1024,7 @@ def _make_public_bootstrap(arm_mode: Mock | None, profiles: dict[str, Mock]) -> pb.arm_profiles = profiles pb.relays = {} pb.sirens = {} + pb.fobs = {} return pb diff --git a/tests/components/unifiprotect/test_siren.py b/tests/components/unifiprotect/test_siren.py index 747972d6764e..aca926ff5e49 100644 --- a/tests/components/unifiprotect/test_siren.py +++ b/tests/components/unifiprotect/test_siren.py @@ -73,6 +73,7 @@ def _make_public_bootstrap(siren: Mock | None) -> Mock: pb.relays = {} pb.arm_mode = None pb.arm_profiles = {} + pb.fobs = {} return pb diff --git a/tests/components/unifiprotect/utils.py b/tests/components/unifiprotect/utils.py index 7bceed289a70..2d1f6dfc2d02 100644 --- a/tests/components/unifiprotect/utils.py +++ b/tests/components/unifiprotect/utils.py @@ -588,6 +588,7 @@ def setup_public_sensor( pb.sensors = public_bootstrap.sensors pb.relays = {} pb.sirens = {} + pb.fobs = {} pb.arm_mode = None pb.arm_profiles = {} @@ -618,6 +619,7 @@ def setup_public_light(ufp: MockUFPFixture) -> None: pb.lights = public_bootstrap.lights pb.relays = {} pb.sirens = {} + pb.fobs = {} pb.arm_mode = None pb.arm_profiles = {} @@ -648,6 +650,7 @@ def setup_public_camera(ufp: MockUFPFixture) -> None: pb.cameras = public_bootstrap.cameras pb.relays = {} pb.sirens = {} + pb.fobs = {} pb.arm_mode = None pb.arm_profiles = {}