evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add linked doorbell event support to HomeKit (#120834)

Browse source
This commit is contained in:
J. Nick Koston 2024-06-29 18:43:20 -05:00 • committed by GitHub
parent 7172d798f8
commit 5280291f98
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 273 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
homeassistant/components/homekit/__init__.py
View file

@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
from collections import defaultdict
from collections.abc import Iterable
from copy import deepcopy
import ipaddress
@ -29,6 +30,7 @@ from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.components.device_automation.trigger import (
async_validate_trigger_config,
)
from homeassistant.components.event import DOMAIN as EVENT_DOMAIN, EventDeviceClass
from homeassistant.components.http import KEY_HASS, HomeAssistantView
from homeassistant.components.humidifier import DOMAIN as HUMIDIFIER_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN, SensorDeviceClass
@ -156,6 +158,17 @@ _HAS_IPV6 = hasattr(socket, "AF_INET6")
_DEFAULT_BIND = ["0.0.0.0", "::"] if _HAS_IPV6 else ["0.0.0.0"]
BATTERY_CHARGING_SENSOR = (
BINARY_SENSOR_DOMAIN,
BinarySensorDeviceClass.BATTERY_CHARGING,
)
BATTERY_SENSOR = (SENSOR_DOMAIN, SensorDeviceClass.BATTERY)
MOTION_SENSOR = (BINARY_SENSOR_DOMAIN, BinarySensorDeviceClass.MOTION)
DOORBELL_EVENT_SENSOR = (EVENT_DOMAIN, EventDeviceClass.DOORBELL)
DOORBELL_BINARY_SENSOR = (BINARY_SENSOR_DOMAIN, BinarySensorDeviceClass.OCCUPANCY)
HUMIDITY_SENSOR = (SENSOR_DOMAIN, SensorDeviceClass.HUMIDITY)
def _has_all_unique_names_and_ports(
bridges: list[dict[str, Any]],
) -> list[dict[str, Any]]:
@ -522,7 +535,7 @@ class HomeKit:
ip_address: str | None,
entity_filter: EntityFilter,
exclude_accessory_mode: bool,
entity_config: dict,
entity_config: dict[str, Any],
homekit_mode: str,
advertise_ips: list[str],
entry_id: str,
@ -535,7 +548,9 @@ class HomeKit:
self._port = port
self._ip_address = ip_address
self._filter = entity_filter
self._config = entity_config
self._config: defaultdict[str, dict[str, Any]] = defaultdict(
dict, entity_config
)
self._exclude_accessory_mode = exclude_accessory_mode
self._advertise_ips = advertise_ips
self._entry_id = entry_id
@ -1074,7 +1089,7 @@ class HomeKit:
def _async_configure_linked_sensors(
self,
ent_reg_ent: er.RegistryEntry,
device_lookup: dict[tuple[str, str | None], str],
lookup: dict[tuple[str, str | None], str],
state: State,
) -> None:
if (ent_reg_ent.device_class or ent_reg_ent.original_device_class) in (
@ -1085,46 +1100,44 @@ class HomeKit:
domain = state.domain
attributes = state.attributes
config = self._config
entity_id = state.entity_id
if ATTR_BATTERY_CHARGING not in attributes and (
battery_charging_binary_sensor_entity_id := device_lookup.get(
(BINARY_SENSOR_DOMAIN, BinarySensorDeviceClass.BATTERY_CHARGING)
battery_charging_binary_sensor_entity_id := lookup.get(
BATTERY_CHARGING_SENSOR
)
):
self._config.setdefault(state.entity_id, {}).setdefault(
config[entity_id].setdefault(
CONF_LINKED_BATTERY_CHARGING_SENSOR,
battery_charging_binary_sensor_entity_id,
)
if ATTR_BATTERY_LEVEL not in attributes and (
battery_sensor_entity_id := device_lookup.get(
(SENSOR_DOMAIN, SensorDeviceClass.BATTERY)
)
battery_sensor_entity_id := lookup.get(BATTERY_SENSOR)
):
self._config.setdefault(state.entity_id, {}).setdefault(
config[entity_id].setdefault(
CONF_LINKED_BATTERY_SENSOR, battery_sensor_entity_id
)
if domain == CAMERA_DOMAIN:
if motion_binary_sensor_entity_id := device_lookup.get(
(BINARY_SENSOR_DOMAIN, BinarySensorDeviceClass.MOTION)
):
self._config.setdefault(state.entity_id, {}).setdefault(
if motion_binary_sensor_entity_id := lookup.get(MOTION_SENSOR):
config[entity_id].setdefault(
CONF_LINKED_MOTION_SENSOR, motion_binary_sensor_entity_id
)
if doorbell_binary_sensor_entity_id := device_lookup.get(
(BINARY_SENSOR_DOMAIN, BinarySensorDeviceClass.OCCUPANCY)
):
self._config.setdefault(state.entity_id, {}).setdefault(
if doorbell_event_entity_id := lookup.get(DOORBELL_EVENT_SENSOR):
config[entity_id].setdefault(
CONF_LINKED_DOORBELL_SENSOR, doorbell_event_entity_id
)
elif doorbell_binary_sensor_entity_id := lookup.get(DOORBELL_BINARY_SENSOR):
config[entity_id].setdefault(
CONF_LINKED_DOORBELL_SENSOR, doorbell_binary_sensor_entity_id
)
if domain == HUMIDIFIER_DOMAIN and (
current_humidity_sensor_entity_id := device_lookup.get(
(SENSOR_DOMAIN, SensorDeviceClass.HUMIDITY)
)
current_humidity_sensor_entity_id := lookup.get(HUMIDITY_SENSOR)
):
self._config.setdefault(state.entity_id, {}).setdefault(
config[entity_id].setdefault(
CONF_LINKED_HUMIDITY_SENSOR, current_humidity_sensor_entity_id
)
@ -1135,7 +1148,7 @@ class HomeKit:
entity_id: str,
) -> None:
"""Set attributes that will be used for homekit device info."""
ent_cfg = self._config.setdefault(entity_id, {})
ent_cfg = self._config[entity_id]
if ent_reg_ent.device_id:
if dev_reg_ent := dev_reg.async_get(ent_reg_ent.device_id):
self._fill_config_from_device_registry_entry(dev_reg_ent, ent_cfg)

73
homeassistant/components/homekit/type_cameras.py
View file

@ -16,7 +16,7 @@ from pyhap.util import callback as pyhap_callback
from homeassistant.components import camera
from homeassistant.components.ffmpeg import get_ffmpeg_manager
from homeassistant.const import STATE_ON
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import (
Event,
EventStateChangedData,
@ -234,30 +234,35 @@ class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
self._char_doorbell_detected = None
self._char_doorbell_detected_switch = None
self.linked_doorbell_sensor = self.config.get(CONF_LINKED_DOORBELL_SENSOR)
if self.linked_doorbell_sensor:
state = self.hass.states.get(self.linked_doorbell_sensor)
if state:
serv_doorbell = self.add_preload_service(SERV_DOORBELL)
self.set_primary_service(serv_doorbell)
self._char_doorbell_detected = serv_doorbell.configure_char(
CHAR_PROGRAMMABLE_SWITCH_EVENT,
value=0,
)
serv_stateless_switch = self.add_preload_service(
SERV_STATELESS_PROGRAMMABLE_SWITCH
)
self._char_doorbell_detected_switch = (
serv_stateless_switch.configure_char(
CHAR_PROGRAMMABLE_SWITCH_EVENT,
value=0,
valid_values={"SinglePress": DOORBELL_SINGLE_PRESS},
)
)
serv_speaker = self.add_preload_service(SERV_SPEAKER)
serv_speaker.configure_char(CHAR_MUTE, value=0)
linked_doorbell_sensor: str | None = self.config.get(
CONF_LINKED_DOORBELL_SENSOR
)
self.linked_doorbell_sensor = linked_doorbell_sensor
self.doorbell_is_event = False
if not linked_doorbell_sensor:
return
self.doorbell_is_event = linked_doorbell_sensor.startswith("event.")
if not (state := self.hass.states.get(linked_doorbell_sensor)):
return
serv_doorbell = self.add_preload_service(SERV_DOORBELL)
self.set_primary_service(serv_doorbell)
self._char_doorbell_detected = serv_doorbell.configure_char(
CHAR_PROGRAMMABLE_SWITCH_EVENT,
value=0,
)
serv_stateless_switch = self.add_preload_service(
SERV_STATELESS_PROGRAMMABLE_SWITCH
)
self._char_doorbell_detected_switch = serv_stateless_switch.configure_char(
CHAR_PROGRAMMABLE_SWITCH_EVENT,
value=0,
valid_values={"SinglePress": DOORBELL_SINGLE_PRESS},
)
serv_speaker = self.add_preload_service(SERV_SPEAKER)
serv_speaker.configure_char(CHAR_MUTE, value=0)
self._async_update_doorbell_state(state)
if not self.doorbell_is_event:
self._async_update_doorbell_state(state)
@pyhap_callback # type: ignore[misc]
@callback
@ -271,7 +276,7 @@ class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
self._subscriptions.append(
async_track_state_change_event(
self.hass,
[self.linked_motion_sensor],
self.linked_motion_sensor,
self._async_update_motion_state_event,
job_type=HassJobType.Callback,
)
@ -282,7 +287,7 @@ class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
self._subscriptions.append(
async_track_state_change_event(
self.hass,
[self.linked_doorbell_sensor],
self.linked_doorbell_sensor,
self._async_update_doorbell_state_event,
job_type=HassJobType.Callback,
)
@ -322,18 +327,20 @@ class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
self, event: Event[EventStateChangedData]
) -> None:
"""Handle state change event listener callback."""
if not state_changed_event_is_same_state(event):
self._async_update_doorbell_state(event.data["new_state"])
if not state_changed_event_is_same_state(event) and (
new_state := event.data["new_state"]
):
self._async_update_doorbell_state(new_state)
@callback
def _async_update_doorbell_state(self, new_state: State | None) -> None:
def _async_update_doorbell_state(self, new_state: State) -> None:
"""Handle link doorbell sensor state change to update HomeKit value."""
if not new_state:
return
assert self._char_doorbell_detected
assert self._char_doorbell_detected_switch
if new_state.state == STATE_ON:
state = new_state.state
if state == STATE_ON or (
self.doorbell_is_event and state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
):
self._char_doorbell_detected.set_value(DOORBELL_SINGLE_PRESS)
self._char_doorbell_detected_switch.set_value(DOORBELL_SINGLE_PRESS)
_LOGGER.debug(

77
tests/components/homekit/test_homekit.py
View file

@ -14,6 +14,7 @@ import pytest
from homeassistant import config as hass_config
from homeassistant.components import homekit as homekit_base, zeroconf
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.event import EventDeviceClass
from homeassistant.components.homekit import (
MAX_DEVICES,
STATUS_READY,
@ -2005,6 +2006,82 @@ async def test_homekit_finds_linked_motion_sensors(
)
@pytest.mark.parametrize(
("domain", "device_class"),
[
("binary_sensor", BinarySensorDeviceClass.OCCUPANCY),
("event", EventDeviceClass.DOORBELL),
],
)
@pytest.mark.usefixtures("mock_async_zeroconf")
async def test_homekit_finds_linked_doorbell_sensors(
hass: HomeAssistant,
hk_driver,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
domain: str,
device_class: EventDeviceClass | BinarySensorDeviceClass,
) -> None:
"""Test homekit can find linked doorbell sensors."""
entry = await async_init_integration(hass)
homekit = _mock_homekit(hass, entry, HOMEKIT_MODE_BRIDGE)
homekit.driver = hk_driver
homekit.bridge = HomeBridge(hass, hk_driver, "mock_bridge")
config_entry = MockConfigEntry(domain="test", data={})
config_entry.add_to_hass(hass)
device_entry = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
sw_version="0.16.0",
model="Camera Server",
manufacturer="Ubq",
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entry = entity_registry.async_get_or_create(
domain,
"camera",
"doorbell_sensor",
device_id=device_entry.id,
original_device_class=device_class,
)
camera = entity_registry.async_get_or_create(
"camera", "camera", "demo", device_id=device_entry.id
)
hass.states.async_set(
entry.entity_id,
STATE_ON,
{ATTR_DEVICE_CLASS: device_class},
)
hass.states.async_set(camera.entity_id, STATE_ON)
with (
patch.object(homekit.bridge, "add_accessory"),
patch(f"{PATH_HOMEKIT}.async_show_setup_message"),
patch(f"{PATH_HOMEKIT}.get_accessory") as mock_get_acc,
patch("pyhap.accessory_driver.AccessoryDriver.async_start"),
):
await homekit.async_start()
await hass.async_block_till_done()
mock_get_acc.assert_called_with(
hass,
ANY,
ANY,
ANY,
{
"manufacturer": "Ubq",
"model": "Camera Server",
"platform": "test",
"sw_version": "0.16.0",
"linked_doorbell_sensor": entry.entity_id,
},
)
@pytest.mark.usefixtures("mock_async_zeroconf")
async def test_homekit_finds_linked_humidity_sensors(
hass: HomeAssistant,

121
tests/components/homekit/test_type_cameras.py
View file

@ -9,6 +9,7 @@ import pytest
from homeassistant.components import camera, ffmpeg
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.camera.img_util import TurboJPEGSingleton
from homeassistant.components.event import EventDeviceClass
from homeassistant.components.homekit.accessories import HomeBridge
from homeassistant.components.homekit.const import (
AUDIO_CODEC_COPY,
@ -30,10 +31,11 @@ from homeassistant.components.homekit.const import (
)
from homeassistant.components.homekit.type_cameras import Camera
from homeassistant.components.homekit.type_switches import Switch
from homeassistant.const import ATTR_DEVICE_CLASS, STATE_OFF, STATE_ON
from homeassistant.const import ATTR_DEVICE_CLASS, STATE_OFF, STATE_ON, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from tests.components.camera.common import mock_turbo_jpeg
@ -941,6 +943,123 @@ async def test_camera_with_linked_doorbell_sensor(
assert char2.value is None
async def test_camera_with_linked_doorbell_event(
hass: HomeAssistant, run_driver, events
) -> None:
"""Test a camera with a linked doorbell event can update."""
await async_setup_component(hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
await async_setup_component(
hass, camera.DOMAIN, {camera.DOMAIN: {"platform": "demo"}}
)
await hass.async_block_till_done()
doorbell_entity_id = "event.doorbell"
hass.states.async_set(
doorbell_entity_id,
dt_util.utcnow().isoformat(),
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
)
await hass.async_block_till_done()
entity_id = "camera.demo_camera"
hass.states.async_set(entity_id, None)
await hass.async_block_till_done()
acc = Camera(
hass,
run_driver,
"Camera",
entity_id,
2,
{
CONF_STREAM_SOURCE: "/dev/null",
CONF_SUPPORT_AUDIO: True,
CONF_VIDEO_CODEC: VIDEO_CODEC_H264_OMX,
CONF_AUDIO_CODEC: AUDIO_CODEC_COPY,
CONF_LINKED_DOORBELL_SENSOR: doorbell_entity_id,
},
)
bridge = HomeBridge("hass", run_driver, "Test Bridge")
bridge.add_accessory(acc)
acc.run()
assert acc.aid == 2
assert acc.category == 17 # Camera
service = acc.get_service(SERV_DOORBELL)
assert service
char = service.get_characteristic(CHAR_PROGRAMMABLE_SWITCH_EVENT)
assert char
assert char.value is None
service2 = acc.get_service(SERV_STATELESS_PROGRAMMABLE_SWITCH)
assert service2
char2 = service.get_characteristic(CHAR_PROGRAMMABLE_SWITCH_EVENT)
assert char2
broker = MagicMock()
char2.broker = broker
assert char2.value is None
hass.states.async_set(
doorbell_entity_id,
STATE_UNKNOWN,
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
)
await hass.async_block_till_done()
assert char.value is None
assert char2.value is None
assert len(broker.mock_calls) == 0
char.set_value(True)
char2.set_value(True)
broker.reset_mock()
original_time = dt_util.utcnow().isoformat()
hass.states.async_set(
doorbell_entity_id,
original_time,
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
)
await hass.async_block_till_done()
assert char.value is None
assert char2.value is None
assert len(broker.mock_calls) == 2
broker.reset_mock()
hass.states.async_set(
doorbell_entity_id,
original_time,
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
force_update=True,
)
await hass.async_block_till_done()
assert char.value is None
assert char2.value is None
assert len(broker.mock_calls) == 0
broker.reset_mock()
hass.states.async_set(
doorbell_entity_id,
original_time,
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL, "other": "attr"},
)
await hass.async_block_till_done()
assert char.value is None
assert char2.value is None
assert len(broker.mock_calls) == 0
broker.reset_mock()
# Ensure we do not throw when the linked
# doorbell sensor is removed
hass.states.async_remove(doorbell_entity_id)
await hass.async_block_till_done()
acc.run()
await hass.async_block_till_done()
assert char.value is None
assert char2.value is None
async def test_camera_with_a_missing_linked_doorbell_sensor(
hass: HomeAssistant, run_driver, events
) -> None: