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

Add bthome event platform (#108268)

Browse source 
Co-authored-by: Ernst Klamer <e.klamer@gmail.com>
This commit is contained in:
J. Nick Koston 2024-01-21 08:17:55 -10:00 • committed by GitHub
parent b4ab1bac56
commit 6525dad57a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 326 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

55
homeassistant/components/bthome/__init__.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import cast
from bthome_ble import BTHomeBluetoothDeviceData, SensorUpdate
from bthome_ble.parser import EncryptionScheme
@ -19,6 +20,7 @@ from homeassistant.helpers.device_registry import (
DeviceRegistry,
async_get,
)
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import (
BTHOME_BLE_EVENT,
@ -30,7 +32,7 @@ from .const import (
)
from .coordinator import BTHomePassiveBluetoothProcessorCoordinator
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.EVENT, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
@ -47,7 +49,7 @@ def process_service_info(
coordinator: BTHomePassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
discovered_device_classes = coordinator.discovered_device_classes
discovered_event_classes = coordinator.discovered_event_classes
if entry.data.get(CONF_SLEEPY_DEVICE, False) != data.sleepy_device:
hass.config_entries.async_update_entry(
entry,
@ -67,28 +69,35 @@ def process_service_info(
sw_version=sensor_device_info.sw_version,
hw_version=sensor_device_info.hw_version,
)
# event_class may be postfixed with a number, ie 'button_2'
# but if there is only one button then it will be 'button'
event_class = event.device_key.key
event_type = event.event_type
if event_class not in discovered_device_classes:
discovered_device_classes.add(event_class)
ble_event = BTHomeBleEvent(
device_id=device.id,
address=address,
event_class=event_class, # ie 'button'
event_type=event_type, # ie 'press'
event_properties=event.event_properties,
)
if event_class not in discovered_event_classes:
discovered_event_classes.add(event_class)
hass.config_entries.async_update_entry(
entry,
data=entry.data
| {CONF_DISCOVERED_EVENT_CLASSES: list(discovered_device_classes)},
| {CONF_DISCOVERED_EVENT_CLASSES: list(discovered_event_classes)},
)
async_dispatcher_send(
hass, format_discovered_event_class(address), event_class, ble_event
)
hass.bus.async_fire(
BTHOME_BLE_EVENT,
dict(
BTHomeBleEvent(
device_id=device.id,
address=address,
event_class=event_class, # ie 'button'
event_type=event_type, # ie 'press'
event_properties=event.event_properties,
)
),
hass.bus.async_fire(BTHOME_BLE_EVENT, cast(dict, ble_event))
async_dispatcher_send(
hass,
format_event_dispatcher_name(address, event_class),
ble_event,
)
# If payload is encrypted and the bindkey is not verified then we need to reauth
@ -98,6 +107,16 @@ def process_service_info(
return update
def format_event_dispatcher_name(address: str, event_class: str) -> str:
"""Format an event dispatcher name."""
return f"{DOMAIN}_event_{address}_{event_class}"
def format_discovered_event_class(address: str) -> str:
"""Format a discovered event class."""
return f"{DOMAIN}_discovered_event_class_{address}"
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up BTHome Bluetooth from a config entry."""
address = entry.unique_id
@ -120,9 +139,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass, entry, data, service_info, device_registry
),
device_data=data,
discovered_device_classes=set(
entry.data.get(CONF_DISCOVERED_EVENT_CLASSES, [])
),
discovered_event_classes=set(entry.data.get(CONF_DISCOVERED_EVENT_CLASSES, [])),
connectable=False,
entry=entry,
)

4
homeassistant/components/bthome/coordinator.py
View file

@ -30,13 +30,13 @@ class BTHomePassiveBluetoothProcessorCoordinator(PassiveBluetoothProcessorCoordi
mode: BluetoothScanningMode,
update_method: Callable[[BluetoothServiceInfoBleak], Any],
device_data: BTHomeBluetoothDeviceData,
discovered_device_classes: set[str],
discovered_event_classes: set[str],
entry: ConfigEntry,
connectable: bool = False,
) -> None:
"""Initialize the BTHome Bluetooth Passive Update Processor Coordinator."""
super().__init__(hass, logger, address, mode, update_method, connectable)
self.discovered_device_classes = discovered_device_classes
self.discovered_event_classes = discovered_event_classes
self.device_data = device_data
self.entry = entry

14
homeassistant/components/bthome/device_trigger.py
View file

@ -87,6 +87,9 @@ async def async_get_triggers(
None,
)
assert bthome_config_entry is not None
event_classes: list[str] = bthome_config_entry.data.get(
CONF_DISCOVERED_EVENT_CLASSES, []
)
return [
{
# Required fields of TRIGGER_BASE_SCHEMA
@ -97,10 +100,15 @@ async def async_get_triggers(
CONF_TYPE: event_class,
CONF_SUBTYPE: event_type,
}
for event_class in bthome_config_entry.data.get(
CONF_DISCOVERED_EVENT_CLASSES, []
for event_class in event_classes
for event_type in TRIGGERS_BY_EVENT_CLASS.get(
event_class.split("_")[0],
# If the device has multiple buttons they will have
# event classes like button_1 button_2, button_3, etc
# but if there is only one button then it will be
# button without a number postfix.
(),
)
for event_type in TRIGGERS_BY_EVENT_CLASS.get(event_class, [])
]

133
homeassistant/components/bthome/event.py Normal file
View file

@ -0,0 +1,133 @@
"""Support for bthome event entities."""
from __future__ import annotations
from dataclasses import replace
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import format_discovered_event_class, format_event_dispatcher_name
from .const import (
DOMAIN,
EVENT_CLASS_BUTTON,
EVENT_CLASS_DIMMER,
EVENT_PROPERTIES,
EVENT_TYPE,
BTHomeBleEvent,
)
from .coordinator import BTHomePassiveBluetoothProcessorCoordinator
DESCRIPTIONS_BY_EVENT_CLASS = {
EVENT_CLASS_BUTTON: EventEntityDescription(
key=EVENT_CLASS_BUTTON,
translation_key="button",
event_types=[
"press",
"double_press",
"triple_press",
"long_press",
"long_double_press",
"long_triple_press",
],
device_class=EventDeviceClass.BUTTON,
),
EVENT_CLASS_DIMMER: EventEntityDescription(
key=EVENT_CLASS_DIMMER,
translation_key="dimmer",
event_types=["rotate_left", "rotate_right"],
),
}
class BTHomeEventEntity(EventEntity):
"""Representation of a BTHome event entity."""
_attr_should_poll = False
_attr_has_entity_name = True
def __init__(
self,
address: str,
event_class: str,
event: BTHomeBleEvent | None,
) -> None:
"""Initialise a BTHome event entity."""
self._update_signal = format_event_dispatcher_name(address, event_class)
# event_class is something like "button" or "dimmer"
# and it maybe postfixed with "_1", "_2", "_3", etc
# If there is only one button then it will be "button"
base_event_class, _, postfix = event_class.partition("_")
base_description = DESCRIPTIONS_BY_EVENT_CLASS[base_event_class]
self.entity_description = replace(base_description, key=event_class)
postfix_name = f" {postfix}" if postfix else ""
self._attr_name = f"{base_event_class.title()}{postfix_name}"
# Matches logic in PassiveBluetoothProcessorEntity
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, address)},
connections={(dr.CONNECTION_BLUETOOTH, address)},
)
self._attr_unique_id = f"{address}-{event_class}"
# If the event is provided then we can set the initial state
# since the event itself is likely what triggered the creation
# of this entity. We have to do this at creation time since
# entities are created dynamically and would otherwise miss
# the initial state.
if event:
self._trigger_event(event[EVENT_TYPE], event[EVENT_PROPERTIES])
async def async_added_to_hass(self) -> None:
"""Entity added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
async_dispatcher_connect(
self.hass,
self._update_signal,
self._async_handle_event,
)
)
@callback
def _async_handle_event(self, event: BTHomeBleEvent) -> None:
self._trigger_event(event[EVENT_TYPE], event[EVENT_PROPERTIES])
self.async_write_ha_state()
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up BTHome event."""
coordinator: BTHomePassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
address = coordinator.address
ent_reg = er.async_get(hass)
async_add_entities(
# Matches logic in PassiveBluetoothProcessorEntity
BTHomeEventEntity(address_event_class[0], address_event_class[2], None)
for ent_reg_entry in er.async_entries_for_config_entry(ent_reg, entry.entry_id)
if ent_reg_entry.domain == "event"
and (address_event_class := ent_reg_entry.unique_id.partition("-"))
)
@callback
def _async_discovered_event_class(event_class: str, event: BTHomeBleEvent) -> None:
"""Handle a newly discovered event class with or without a postfix."""
async_add_entities([BTHomeEventEntity(address, event_class, event)])
entry.async_on_unload(
async_dispatcher_connect(
hass,
format_discovered_event_class(address),
_async_discovered_event_class,
)
)

28
homeassistant/components/bthome/strings.json
View file

@ -44,5 +44,33 @@
"button": "Button \"{subtype}\"",
"dimmer": "Dimmer \"{subtype}\""
}
},
"entity": {
"event": {
"button": {
"state_attributes": {
"event_type": {
"state": {
"press": "Press",
"double_press": "Double press",
"triple_press": "Triple press",
"long_press": "Long press",
"long_double_press": "Long double press",
"long_triple_press": "Long triple press"
}
}
}
},
"dimmer": {
"state_attributes": {
"event_type": {
"state": {
"rotate_left": "Rotate left",
"rotate_right": "Rotate right"
}
}
}
}
}
}
}

116
tests/components/bthome/test_event.py Normal file
View file

@ -0,0 +1,116 @@
"""Test the BTHome events."""
import pytest
from homeassistant.components.bthome.const import DOMAIN
from homeassistant.components.event import ATTR_EVENT_TYPE
from homeassistant.const import ATTR_FRIENDLY_NAME, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from . import make_bthome_v2_adv
from tests.common import MockConfigEntry
from tests.components.bluetooth import (
BluetoothServiceInfoBleak,
inject_bluetooth_service_info,
)
@pytest.mark.parametrize(
("mac_address", "advertisement", "bind_key", "result"),
[
(
"A4:C1:38:8D:18:B2",
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x40\x3A\x00\x3A\x01\x3A\x03",
),
None,
[
{
"entity": "event.test_device_18b2_button_2",
ATTR_FRIENDLY_NAME: "Test Device 18B2 Button 2",
ATTR_EVENT_TYPE: "press",
},
{
"entity": "event.test_device_18b2_button_3",
ATTR_FRIENDLY_NAME: "Test Device 18B2 Button 3",
ATTR_EVENT_TYPE: "triple_press",
},
],
),
(
"A4:C1:38:8D:18:B2",
make_bthome_v2_adv(
"A4:C1:38:8D:18:B2",
b"\x40\x3A\x04",
),
None,
[
{
"entity": "event.test_device_18b2_button",
ATTR_FRIENDLY_NAME: "Test Device 18B2 Button",
ATTR_EVENT_TYPE: "long_press",
}
],
),
],
)
async def test_v2_events(
hass: HomeAssistant,
mac_address: str,
advertisement: BluetoothServiceInfoBleak,
bind_key: str | None,
result: list[dict[str, str]],
) -> None:
"""Test the different BTHome V2 events."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=mac_address,
data={"bindkey": bind_key},
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
inject_bluetooth_service_info(
hass,
advertisement,
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == len(result)
for meas in result:
state = hass.states.get(meas["entity"])
attributes = state.attributes
assert attributes[ATTR_FRIENDLY_NAME] == meas[ATTR_FRIENDLY_NAME]
assert attributes[ATTR_EVENT_TYPE] == meas[ATTR_EVENT_TYPE]
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Ensure entities are restored
for meas in result:
state = hass.states.get(meas["entity"])
assert state != STATE_UNAVAILABLE
# Now inject again
inject_bluetooth_service_info(
hass,
advertisement,
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == len(result)
for meas in result:
state = hass.states.get(meas["entity"])
attributes = state.attributes
assert attributes[ATTR_FRIENDLY_NAME] == meas[ATTR_FRIENDLY_NAME]
assert attributes[ATTR_EVENT_TYPE] == meas[ATTR_EVENT_TYPE]
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()