Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for `linked_doorbell_sensor` to HomeKit locks (#131660) 

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/131973/head
Andy 2024-11-30 20:30:21 +01:00 committed by GitHub
parent 6da2515d7a
commit bcdac7ed37
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 456 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
homeassistant/components/homekit/__init__.py
View File

@ -33,6 +33,7 @@ from homeassistant.components.device_automation.trigger import (
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.lock import DOMAIN as LOCK_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN, SensorDeviceClass
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
@ -1133,6 +1134,8 @@ class HomeKit:
config[entity_id].setdefault(
CONF_LINKED_MOTION_SENSOR, motion_binary_sensor_entity_id
)
if domain in (CAMERA_DOMAIN, LOCK_DOMAIN):
if doorbell_event_entity_id := lookup.get(DOORBELL_EVENT_SENSOR):
config[entity_id].setdefault(
CONF_LINKED_DOORBELL_SENSOR, doorbell_event_entity_id

121
homeassistant/components/homekit/doorbell.py Normal file
View File

@ -0,0 +1,121 @@
"""Extend the doorbell functions."""
from __future__ import annotations
import logging
from typing import Any
from pyhap.util import callback as pyhap_callback
from homeassistant.const import STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import (
Event,
EventStateChangedData,
HassJobType,
State,
callback as ha_callback,
)
from homeassistant.helpers.event import async_track_state_change_event
from .accessories import HomeAccessory
from .const import (
CHAR_MUTE,
CHAR_PROGRAMMABLE_SWITCH_EVENT,
CONF_LINKED_DOORBELL_SENSOR,
SERV_DOORBELL,
SERV_SPEAKER,
SERV_STATELESS_PROGRAMMABLE_SWITCH,
)
from .util import state_changed_event_is_same_state
_LOGGER = logging.getLogger(__name__)
DOORBELL_SINGLE_PRESS = 0
DOORBELL_DOUBLE_PRESS = 1
DOORBELL_LONG_PRESS = 2
class HomeDoorbellAccessory(HomeAccessory):
"""Accessory with optional doorbell."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize an Accessory object with optional attached doorbell."""
super().__init__(*args, **kwargs)
self._char_doorbell_detected = None
self._char_doorbell_detected_switch = None
linked_doorbell_sensor: str | None
linked_doorbell_sensor = 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(None, state)
@ha_callback
@pyhap_callback # type: ignore[misc]
def run(self) -> None:
"""Handle doorbell event."""
if self._char_doorbell_detected:
assert self.linked_doorbell_sensor
self._subscriptions.append(
async_track_state_change_event(
self.hass,
self.linked_doorbell_sensor,
self.async_update_doorbell_state_event,
job_type=HassJobType.Callback,
)
)
super().run()
@ha_callback
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) and (
new_state := event.data["new_state"]
):
self.async_update_doorbell_state(event.data["old_state"], new_state)
@ha_callback
def async_update_doorbell_state(
self, old_state: State | None, new_state: State
) -> None:
"""Handle link doorbell sensor state change to update HomeKit value."""
assert self._char_doorbell_detected
assert self._char_doorbell_detected_switch
state = new_state.state
if state == STATE_ON or (
self.doorbell_is_event
and old_state is not None
and old_state.state != STATE_UNAVAILABLE
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(
"%s: Set linked doorbell %s sensor to %d",
self.entity_id,
self.linked_doorbell_sensor,
DOORBELL_SINGLE_PRESS,
)

88
homeassistant/components/homekit/type_cameras.py
View File

@ -31,15 +31,12 @@ from homeassistant.helpers.event import (
)
from homeassistant.util.async_ import create_eager_task
from .accessories import TYPES, HomeAccessory, HomeDriver
from .accessories import TYPES, HomeDriver
from .const import (
CHAR_MOTION_DETECTED,
CHAR_MUTE,
CHAR_PROGRAMMABLE_SWITCH_EVENT,
CONF_AUDIO_CODEC,
CONF_AUDIO_MAP,
CONF_AUDIO_PACKET_SIZE,
CONF_LINKED_DOORBELL_SENSOR,
CONF_LINKED_MOTION_SENSOR,
CONF_MAX_FPS,
CONF_MAX_HEIGHT,
@ -64,18 +61,13 @@ from .const import (
DEFAULT_VIDEO_MAP,
DEFAULT_VIDEO_PACKET_SIZE,
DEFAULT_VIDEO_PROFILE_NAMES,
SERV_DOORBELL,
SERV_MOTION_SENSOR,
SERV_SPEAKER,
SERV_STATELESS_PROGRAMMABLE_SWITCH,
)
from .doorbell import HomeDoorbellAccessory
from .util import pid_is_alive, state_changed_event_is_same_state
_LOGGER = logging.getLogger(__name__)
DOORBELL_SINGLE_PRESS = 0
DOORBELL_DOUBLE_PRESS = 1
DOORBELL_LONG_PRESS = 2
VIDEO_OUTPUT = (
"-map {v_map} -an "
@ -149,7 +141,7 @@ CONFIG_DEFAULTS = {
@TYPES.register("Camera")
# False-positive on pylint, not a CameraEntity
# pylint: disable-next=hass-enforce-class-module
class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
class Camera(HomeDoorbellAccessory, PyhapCamera): # type: ignore[misc]
"""Generate a Camera accessory."""
def __init__(
@ -237,36 +229,6 @@ class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
)
self._async_update_motion_state(None, state)
self._char_doorbell_detected = None
self._char_doorbell_detected_switch = None
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(None, state)
@pyhap_callback # type: ignore[misc]
@callback
def run(self) -> None:
@ -285,17 +247,6 @@ class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
)
)
if self._char_doorbell_detected:
assert self.linked_doorbell_sensor
self._subscriptions.append(
async_track_state_change_event(
self.hass,
self.linked_doorbell_sensor,
self._async_update_doorbell_state_event,
job_type=HassJobType.Callback,
)
)
super().run()
@callback
@ -344,39 +295,6 @@ class Camera(HomeAccessory, PyhapCamera): # type: ignore[misc]
detected,
)
@callback
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) and (
new_state := event.data["new_state"]
):
self._async_update_doorbell_state(event.data["old_state"], new_state)
@callback
def _async_update_doorbell_state(
self, old_state: State | None, new_state: State
) -> None:
"""Handle link doorbell sensor state change to update HomeKit value."""
assert self._char_doorbell_detected
assert self._char_doorbell_detected_switch
state = new_state.state
if state == STATE_ON or (
self.doorbell_is_event
and old_state is not None
and old_state.state != STATE_UNAVAILABLE
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(
"%s: Set linked doorbell %s sensor to %d",
self.entity_id,
self.linked_doorbell_sensor,
DOORBELL_SINGLE_PRESS,
)
@callback
def async_update_state(self, new_state: State | None) -> None:
"""Handle state change to update HomeKit value."""

5
homeassistant/components/homekit/type_locks.py
View File

@ -9,8 +9,9 @@ from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN, LockState
from homeassistant.const import ATTR_CODE, ATTR_ENTITY_ID, STATE_UNKNOWN
from homeassistant.core import State, callback
from .accessories import TYPES, HomeAccessory
from .accessories import TYPES
from .const import CHAR_LOCK_CURRENT_STATE, CHAR_LOCK_TARGET_STATE, SERV_LOCK
from .doorbell import HomeDoorbellAccessory
_LOGGER = logging.getLogger(__name__)
@ -53,7 +54,7 @@ STATE_TO_SERVICE = {
@TYPES.register("Lock")
class Lock(HomeAccessory):
class Lock(HomeDoorbellAccessory):
"""Generate a Lock accessory for a lock entity.
The lock entity must support: unlock and lock.

14
homeassistant/components/homekit/util.py
View File

@ -182,7 +182,6 @@ HUMIDIFIER_SCHEMA = BASIC_INFO_SCHEMA.extend(
{vol.Optional(CONF_LINKED_HUMIDITY_SENSOR): cv.entity_domain(sensor.DOMAIN)}
)
COVER_SCHEMA = BASIC_INFO_SCHEMA.extend(
{
vol.Optional(CONF_LINKED_OBSTRUCTION_SENSOR): cv.entity_domain(
@ -195,6 +194,14 @@ CODE_SCHEMA = BASIC_INFO_SCHEMA.extend(
{vol.Optional(ATTR_CODE, default=None): vol.Any(None, cv.string)}
)
LOCK_SCHEMA = CODE_SCHEMA.extend(
{
vol.Optional(CONF_LINKED_DOORBELL_SENSOR): cv.entity_domain(
[binary_sensor.DOMAIN, EVENT_DOMAIN]
),
}
)
MEDIA_PLAYER_SCHEMA = vol.Schema(
{
vol.Required(CONF_FEATURE): vol.All(
@ -284,7 +291,7 @@ def validate_entity_config(values: dict) -> dict[str, dict]:
if not isinstance(config, dict):
raise vol.Invalid(f"The configuration for {entity} must be a dictionary.")
if domain in ("alarm_control_panel", "lock"):
if domain == "alarm_control_panel":
config = CODE_SCHEMA(config)
elif domain == media_player.const.DOMAIN:
@ -301,6 +308,9 @@ def validate_entity_config(values: dict) -> dict[str, dict]:
elif domain == "camera":
config = CAMERA_SCHEMA(config)
elif domain == "lock":
config = LOCK_SCHEMA(config)
elif domain == "switch":
config = SWITCH_TYPE_SCHEMA(config)

301
tests/components/homekit/test_type_locks.py
View File

@ -1,17 +1,34 @@
"""Test different accessory types: Locks."""
from unittest.mock import MagicMock
import pytest
from homeassistant.components.homekit.const import ATTR_VALUE
from homeassistant.components import lock
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.event import EventDeviceClass
from homeassistant.components.homekit.accessories import HomeBridge
from homeassistant.components.homekit.const import (
ATTR_VALUE,
CHAR_PROGRAMMABLE_SWITCH_EVENT,
CONF_LINKED_DOORBELL_SENSOR,
SERV_DOORBELL,
SERV_STATELESS_PROGRAMMABLE_SWITCH,
)
from homeassistant.components.homekit.type_locks import Lock
from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN, LockState
from homeassistant.const import (
ATTR_CODE,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import Event, HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from tests.common import async_mock_service
@ -135,3 +152,285 @@ async def test_no_code(
assert acc.char_target_state.value == 1
assert len(events) == 1
assert events[-1].data[ATTR_VALUE] is None
async def test_lock_with_linked_doorbell_sensor(hass: HomeAssistant, hk_driver) -> None:
"""Test a lock with a linked doorbell sensor can update."""
code = "1234"
await async_setup_component(hass, lock.DOMAIN, {lock.DOMAIN: {"platform": "demo"}})
await hass.async_block_till_done()
doorbell_entity_id = "binary_sensor.doorbell"
hass.states.async_set(
doorbell_entity_id,
STATE_ON,
{ATTR_DEVICE_CLASS: BinarySensorDeviceClass.OCCUPANCY},
)
await hass.async_block_till_done()
entity_id = "lock.demo_lock"
hass.states.async_set(entity_id, None)
await hass.async_block_till_done()
acc = Lock(
hass,
hk_driver,
"Lock",
entity_id,
2,
{
ATTR_CODE: code,
CONF_LINKED_DOORBELL_SENSOR: doorbell_entity_id,
},
)
bridge = HomeBridge("hass", hk_driver, "Test Bridge")
bridge.add_accessory(acc)
acc.run()
assert acc.aid == 2
assert acc.category == 6 # DoorLock
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_OFF,
{ATTR_DEVICE_CLASS: BinarySensorDeviceClass.OCCUPANCY},
)
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()
hass.states.async_set(
doorbell_entity_id,
STATE_ON,
{ATTR_DEVICE_CLASS: BinarySensorDeviceClass.OCCUPANCY},
)
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,
STATE_ON,
{ATTR_DEVICE_CLASS: BinarySensorDeviceClass.OCCUPANCY},
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,
STATE_ON,
{ATTR_DEVICE_CLASS: BinarySensorDeviceClass.OCCUPANCY, "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_lock_with_linked_doorbell_event(hass: HomeAssistant, hk_driver) -> None:
"""Test a lock with a linked doorbell event can update."""
await async_setup_component(hass, lock.DOMAIN, {lock.DOMAIN: {"platform": "demo"}})
await hass.async_block_till_done()
doorbell_entity_id = "event.doorbell"
code = "1234"
hass.states.async_set(
doorbell_entity_id,
dt_util.utcnow().isoformat(),
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
)
await hass.async_block_till_done()
entity_id = "lock.demo_lock"
hass.states.async_set(entity_id, None)
await hass.async_block_till_done()
acc = Lock(
hass,
hk_driver,
"Lock",
entity_id,
2,
{
ATTR_CODE: code,
CONF_LINKED_DOORBELL_SENSOR: doorbell_entity_id,
},
)
bridge = HomeBridge("hass", hk_driver, "Test Bridge")
bridge.add_accessory(acc)
acc.run()
assert acc.aid == 2
assert acc.category == 6 # DoorLock
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
await hass.async_block_till_done()
hass.states.async_set(
doorbell_entity_id,
STATE_UNAVAILABLE,
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
)
await hass.async_block_till_done()
# Ensure re-adding does not fire an event
assert not broker.mock_calls
broker.reset_mock()
# going from unavailable to a state should not fire an event
hass.states.async_set(
doorbell_entity_id,
dt_util.utcnow().isoformat(),
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
)
await hass.async_block_till_done()
assert not broker.mock_calls
# But a second update does
hass.states.async_set(
doorbell_entity_id,
dt_util.utcnow().isoformat(),
{ATTR_DEVICE_CLASS: EventDeviceClass.DOORBELL},
)
await hass.async_block_till_done()
assert broker.mock_calls
async def test_lock_with_a_missing_linked_doorbell_sensor(
hass: HomeAssistant, hk_driver
) -> None:
"""Test a lock with a configured linked doorbell sensor that is missing."""
await async_setup_component(hass, lock.DOMAIN, {lock.DOMAIN: {"platform": "demo"}})
await hass.async_block_till_done()
code = "1234"
doorbell_entity_id = "binary_sensor.doorbell"
entity_id = "lock.demo_lock"
hass.states.async_set(entity_id, None)
await hass.async_block_till_done()
acc = Lock(
hass,
hk_driver,
"Lock",
entity_id,
2,
{
ATTR_CODE: code,
CONF_LINKED_DOORBELL_SENSOR: doorbell_entity_id,
},
)
bridge = HomeBridge("hass", hk_driver, "Test Bridge")
bridge.add_accessory(acc)
acc.run()
assert acc.aid == 2
assert acc.category == 6 # DoorLock
assert not acc.get_service(SERV_DOORBELL)
assert not acc.get_service(SERV_STATELESS_PROGRAMMABLE_SWITCH)

16
tests/components/homekit/test_util.py
View File

@ -159,8 +159,20 @@ def test_validate_entity_config() -> None:
assert vec({"lock.demo": {}}) == {
"lock.demo": {ATTR_CODE: None, CONF_LOW_BATTERY_THRESHOLD: 20}
}
assert vec({"lock.demo": {ATTR_CODE: "1234"}}) == {
"lock.demo": {ATTR_CODE: "1234", CONF_LOW_BATTERY_THRESHOLD: 20}
assert vec(
{
"lock.demo": {
ATTR_CODE: "1234",
CONF_LINKED_DOORBELL_SENSOR: "event.doorbell",
}
}
) == {
"lock.demo": {
ATTR_CODE: "1234",
CONF_LOW_BATTERY_THRESHOLD: 20,
CONF_LINKED_DOORBELL_SENSOR: "event.doorbell",
}
}
assert vec({"media_player.demo": {}}) == {