Skip to content

Commit

Permalink
Add linked doorbell event support to HomeKit (#120834)
Browse files Browse the repository at this point in the history
  • Loading branch information
@bdraco
bdraco authored Jun 29, 2024
1 parent 7172d79 commit 5280291
Show file tree
Hide file tree
Showing 4 changed files with 273 additions and 57 deletions.
59 changes: 36 additions & 23 deletions homeassistant/components/homekit/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -29,6 +30,7 @@
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
Expand Down Expand Up @@ -156,6 +158,17 @@
_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]]:
Expand Down Expand Up @@ -522,7 +535,7 @@ def __init__(
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,
Expand All @@ -535,7 +548,9 @@ def __init__(
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
Expand Down Expand Up @@ -1074,7 +1089,7 @@ async def async_stop(self, *args: Any) -> None:
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 (
Expand All @@ -1085,46 +1100,44 @@ def _async_configure_linked_sensors(

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
)

Expand All @@ -1135,7 +1148,7 @@ async def _async_set_device_info_attributes(
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)
Expand Down
73 changes: 40 additions & 33 deletions homeassistant/components/homekit/type_cameras.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@

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,
Expand Down Expand Up @@ -234,30 +234,35 @@ def __init__(

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
Expand All @@ -271,7 +276,7 @@ def run(self) -> None:
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,
)
Expand All @@ -282,7 +287,7 @@ def run(self) -> None:
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,
)
Expand Down Expand Up @@ -322,18 +327,20 @@ def _async_update_doorbell_state_event(
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(
Expand Down
77 changes: 77 additions & 0 deletions tests/components/homekit/test_homekit.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
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,
Expand Down Expand Up @@ -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,
Expand Down
Loading

0 comments on commit 5280291

Please sign in to comment.