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

Add event entities to homekit_controller (#97140)

Browse source 
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
Jc2k 2023-07-26 07:20:41 +01:00 • committed by GitHub
parent d0512d5b26
commit c0debaf26e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 401 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
homeassistant/components/homekit_controller/connection.py
View file

@ -95,6 +95,13 @@ class HKDevice:
# A list of callbacks that turn HK service metadata into entities
self.listeners: list[AddServiceCb] = []
# A list of callbacks that turn HK service metadata into triggers
self.trigger_factories: list[AddServiceCb] = []
# Track aid/iid pairs so we know if we already handle triggers for a HK
# service.
self._triggers: list[tuple[int, int]] = []
# A list of callbacks that turn HK characteristics into entities
self.char_factories: list[AddCharacteristicCb] = []
@ -637,11 +644,33 @@ class HKDevice:
self.listeners.append(add_entities_cb)
self._add_new_entities([add_entities_cb])
def add_trigger_factory(self, add_triggers_cb: AddServiceCb) -> None:
"""Add a callback to run when discovering new triggers for services."""
self.trigger_factories.append(add_triggers_cb)
self._add_new_triggers([add_triggers_cb])
def _add_new_triggers(self, callbacks: list[AddServiceCb]) -> None:
for accessory in self.entity_map.accessories:
aid = accessory.aid
for service in accessory.services:
iid = service.iid
entity_key = (aid, iid)
if entity_key in self._triggers:
# Don't add the same trigger again
continue
for add_trigger_cb in callbacks:
if add_trigger_cb(service):
self._triggers.append(entity_key)
break
def add_entities(self) -> None:
"""Process the entity map and create HA entities."""
self._add_new_entities(self.listeners)
self._add_new_entities_for_accessory(self.accessory_factories)
self._add_new_entities_for_char(self.char_factories)
self._add_new_triggers(self.trigger_factories)
def _add_new_entities(self, callbacks) -> None:
for accessory in self.entity_map.accessories:

3
homeassistant/components/homekit_controller/const.py
View file

@ -53,6 +53,9 @@ HOMEKIT_ACCESSORY_DISPATCH = {
ServicesTypes.TELEVISION: "media_player",
ServicesTypes.VALVE: "switch",
ServicesTypes.CAMERA_RTP_STREAM_MANAGEMENT: "camera",
ServicesTypes.DOORBELL: "event",
ServicesTypes.STATELESS_PROGRAMMABLE_SWITCH: "event",
ServicesTypes.SERVICE_LABEL: "event",
}
CHARACTERISTIC_PLATFORMS = {

4
homeassistant/components/homekit_controller/device_trigger.py
View file

@ -211,7 +211,7 @@ async def async_setup_triggers_for_entry(
conn: HKDevice = hass.data[KNOWN_DEVICES][hkid]
@callback
def async_add_service(service):
def async_add_characteristic(service: Service):
aid = service.accessory.aid
service_type = service.type
@ -238,7 +238,7 @@ async def async_setup_triggers_for_entry(
return True
conn.add_listener(async_add_service)
conn.add_trigger_factory(async_add_characteristic)
@callback

160
homeassistant/components/homekit_controller/event.py Normal file
View file

@ -0,0 +1,160 @@
"""Support for Homekit motion sensors."""
from __future__ import annotations
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics.const import InputEventValues
from aiohomekit.model.services import Service, ServicesTypes
from aiohomekit.utils import clamp_enum_to_char
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import KNOWN_DEVICES
from .connection import HKDevice
from .entity import HomeKitEntity
INPUT_EVENT_VALUES = {
InputEventValues.SINGLE_PRESS: "single_press",
InputEventValues.DOUBLE_PRESS: "double_press",
InputEventValues.LONG_PRESS: "long_press",
}
class HomeKitEventEntity(HomeKitEntity, EventEntity):
"""Representation of a Homekit event entity."""
_attr_should_poll = False
def __init__(
self,
connection: HKDevice,
service: Service,
entity_description: EventEntityDescription,
) -> None:
"""Initialise a generic HomeKit event entity."""
super().__init__(
connection,
{
"aid": service.accessory.aid,
"iid": service.iid,
},
)
self._characteristic = service.characteristics_by_type[
CharacteristicsTypes.INPUT_EVENT
]
self.entity_description = entity_description
# An INPUT_EVENT may support single_press, long_press and double_press. All are optional. So we have to
# clamp InputEventValues for this exact device
self._attr_event_types = [
INPUT_EVENT_VALUES[v]
for v in clamp_enum_to_char(InputEventValues, self._characteristic)
]
def get_characteristic_types(self) -> list[str]:
"""Define the homekit characteristics the entity cares about."""
return [CharacteristicsTypes.INPUT_EVENT]
async def async_added_to_hass(self) -> None:
"""Entity added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
self._accessory.async_subscribe(
[(self._aid, self._characteristic.iid)],
self._handle_event,
)
)
@callback
def _handle_event(self):
if self._characteristic.value is None:
# For IP backed devices the characteristic is marked as
# pollable, but always returns None when polled
# Make sure we don't explode if we see that edge case.
return
self._trigger_event(INPUT_EVENT_VALUES[self._characteristic.value])
self.async_write_ha_state()
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Homekit event."""
hkid: str = config_entry.data["AccessoryPairingID"]
conn: HKDevice = hass.data[KNOWN_DEVICES][hkid]
@callback
def async_add_service(service: Service) -> bool:
entities = []
if service.type == ServicesTypes.DOORBELL:
entities.append(
HomeKitEventEntity(
conn,
service,
EventEntityDescription(
key=f"{service.accessory.aid}_{service.iid}",
device_class=EventDeviceClass.DOORBELL,
translation_key="doorbell",
),
)
)
elif service.type == ServicesTypes.SERVICE_LABEL:
switches = list(
service.accessory.services.filter(
service_type=ServicesTypes.STATELESS_PROGRAMMABLE_SWITCH,
child_service=service,
order_by=[CharacteristicsTypes.SERVICE_LABEL_INDEX],
)
)
for switch in switches:
# The Apple docs say that if we number the buttons ourselves
# We do it in service label index order. `switches` is already in
# that order.
entities.append(
HomeKitEventEntity(
conn,
switch,
EventEntityDescription(
key=f"{service.accessory.aid}_{service.iid}",
device_class=EventDeviceClass.BUTTON,
translation_key="button",
),
)
)
elif service.type == ServicesTypes.STATELESS_PROGRAMMABLE_SWITCH:
# A stateless switch that has a SERVICE_LABEL_INDEX is part of a group
# And is handled separately
if not service.has(CharacteristicsTypes.SERVICE_LABEL_INDEX):
entities.append(
HomeKitEventEntity(
conn,
service,
EventEntityDescription(
key=f"{service.accessory.aid}_{service.iid}",
device_class=EventDeviceClass.BUTTON,
translation_key="button",
),
)
)
if entities:
async_add_entities(entities)
return True
return False
conn.add_listener(async_add_service)

24
homeassistant/components/homekit_controller/strings.json
View file

@ -71,6 +71,30 @@
}
},
"entity": {
"event": {
"doorbell": {
"state_attributes": {
"event_type": {
"state": {
"double_press": "Double press",
"long_press": "Long press",
"single_press": "Single press"
}
}
}
},
"button": {
"state_attributes": {
"event_type": {
"state": {
"double_press": "[%key:component::homekit_controller::entity::event::doorbell::state_attributes::event_type::state::double_press%]",
"long_press": "[%key:component::homekit_controller::entity::event::doorbell::state_attributes::event_type::state::long_press%]",
"single_press": "[%key:component::homekit_controller::entity::event::doorbell::state_attributes::event_type::state::single_press%]"
}
}
}
}
},
"select": {
"ecobee_mode": {
"state": {

183
tests/components/homekit_controller/test_event.py Normal file
View file

@ -0,0 +1,183 @@
"""Test homekit_controller stateless triggers."""
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes
from homeassistant.components.event import EventDeviceClass
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .common import setup_test_component
def create_remote(accessory):
"""Define characteristics for a button (that is inn a group)."""
service_label = accessory.add_service(ServicesTypes.SERVICE_LABEL)
char = service_label.add_char(CharacteristicsTypes.SERVICE_LABEL_NAMESPACE)
char.value = 1
for i in range(4):
button = accessory.add_service(ServicesTypes.STATELESS_PROGRAMMABLE_SWITCH)
button.linked.append(service_label)
char = button.add_char(CharacteristicsTypes.INPUT_EVENT)
char.value = 0
char.perms = ["pw", "pr", "ev"]
char = button.add_char(CharacteristicsTypes.NAME)
char.value = f"Button {i + 1}"
char = button.add_char(CharacteristicsTypes.SERVICE_LABEL_INDEX)
char.value = i
battery = accessory.add_service(ServicesTypes.BATTERY_SERVICE)
battery.add_char(CharacteristicsTypes.BATTERY_LEVEL)
def create_button(accessory):
"""Define a button (that is not in a group)."""
button = accessory.add_service(ServicesTypes.STATELESS_PROGRAMMABLE_SWITCH)
char = button.add_char(CharacteristicsTypes.INPUT_EVENT)
char.value = 0
char.perms = ["pw", "pr", "ev"]
char = button.add_char(CharacteristicsTypes.NAME)
char.value = "Button 1"
battery = accessory.add_service(ServicesTypes.BATTERY_SERVICE)
battery.add_char(CharacteristicsTypes.BATTERY_LEVEL)
def create_doorbell(accessory):
"""Define a button (that is not in a group)."""
button = accessory.add_service(ServicesTypes.DOORBELL)
char = button.add_char(CharacteristicsTypes.INPUT_EVENT)
char.value = 0
char.perms = ["pw", "pr", "ev"]
char = button.add_char(CharacteristicsTypes.NAME)
char.value = "Doorbell"
battery = accessory.add_service(ServicesTypes.BATTERY_SERVICE)
battery.add_char(CharacteristicsTypes.BATTERY_LEVEL)
async def test_remote(hass: HomeAssistant, utcnow) -> None:
"""Test that remote is supported."""
helper = await setup_test_component(hass, create_remote)
entities = [
("event.testdevice_button_1", "Button 1"),
("event.testdevice_button_2", "Button 2"),
("event.testdevice_button_3", "Button 3"),
("event.testdevice_button_4", "Button 4"),
]
entity_registry = er.async_get(hass)
for entity_id, service in entities:
button = entity_registry.async_get(entity_id)
assert button.original_device_class == EventDeviceClass.BUTTON
assert button.capabilities["event_types"] == [
"single_press",
"double_press",
"long_press",
]
helper.pairing.testing.update_named_service(
service, {CharacteristicsTypes.INPUT_EVENT: 0}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "single_press"
helper.pairing.testing.update_named_service(
service, {CharacteristicsTypes.INPUT_EVENT: 1}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "double_press"
helper.pairing.testing.update_named_service(
service, {CharacteristicsTypes.INPUT_EVENT: 2}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "long_press"
async def test_button(hass: HomeAssistant, utcnow) -> None:
"""Test that a button is correctly enumerated."""
helper = await setup_test_component(hass, create_button)
entity_id = "event.testdevice_button_1"
entity_registry = er.async_get(hass)
button = entity_registry.async_get(entity_id)
assert button.original_device_class == EventDeviceClass.BUTTON
assert button.capabilities["event_types"] == [
"single_press",
"double_press",
"long_press",
]
helper.pairing.testing.update_named_service(
"Button 1", {CharacteristicsTypes.INPUT_EVENT: 0}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "single_press"
helper.pairing.testing.update_named_service(
"Button 1", {CharacteristicsTypes.INPUT_EVENT: 1}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "double_press"
helper.pairing.testing.update_named_service(
"Button 1", {CharacteristicsTypes.INPUT_EVENT: 2}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "long_press"
async def test_doorbell(hass: HomeAssistant, utcnow) -> None:
"""Test that doorbell service is handled."""
helper = await setup_test_component(hass, create_doorbell)
entity_id = "event.testdevice_doorbell"
entity_registry = er.async_get(hass)
doorbell = entity_registry.async_get(entity_id)
assert doorbell.original_device_class == EventDeviceClass.DOORBELL
assert doorbell.capabilities["event_types"] == [
"single_press",
"double_press",
"long_press",
]
helper.pairing.testing.update_named_service(
"Doorbell", {CharacteristicsTypes.INPUT_EVENT: 0}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "single_press"
helper.pairing.testing.update_named_service(
"Doorbell", {CharacteristicsTypes.INPUT_EVENT: 1}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "double_press"
helper.pairing.testing.update_named_service(
"Doorbell", {CharacteristicsTypes.INPUT_EVENT: 2}
)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.attributes["event_type"] == "long_press"