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

Add events to BTHome (#91691) 

Co-authored-by: J. Nick Koston <nick@koston.org>
pull/91873/head
Ernst Klamer 2023-04-22 21:10:27 +02:00 committed by GitHub
parent ca76285bcf
commit 328b79a4af
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 529 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
homeassistant/components/bthome/__init__.py
View File

@ -7,6 +7,7 @@ from bthome_ble import BTHomeBluetoothDeviceData, SensorUpdate
from bthome_ble.parser import EncryptionScheme
from homeassistant.components.bluetooth import (
DOMAIN as BLUETOOTH_DOMAIN,
BluetoothScanningMode,
BluetoothServiceInfoBleak,
)
@ -16,8 +17,16 @@ from homeassistant.components.bluetooth.passive_update_processor import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceRegistry, async_get
from .const import DOMAIN
from .const import (
BTHOME_BLE_EVENT,
CONF_BINDKEY,
CONF_DISCOVERED_EVENT_CLASSES,
DOMAIN,
BTHomeBleEvent,
)
from .models import BTHomeData
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
@ -29,10 +38,53 @@ def process_service_info(
entry: ConfigEntry,
data: BTHomeBluetoothDeviceData,
service_info: BluetoothServiceInfoBleak,
device_registry: DeviceRegistry,
) -> SensorUpdate:
"""Process a BluetoothServiceInfoBleak, running side effects and returning sensor data."""
update = data.update(service_info)
# If that payload was encrypted and the bindkey was not verified then we need to reauth
domain_data: BTHomeData = hass.data[DOMAIN][entry.entry_id]
if update.events:
address = service_info.device.address
for device_key, event in update.events.items():
sensor_device_info = update.devices[device_key.device_id]
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(BLUETOOTH_DOMAIN, address)},
manufacturer=sensor_device_info.manufacturer,
model=sensor_device_info.model,
name=sensor_device_info.name,
sw_version=sensor_device_info.sw_version,
hw_version=sensor_device_info.hw_version,
)
event_class = event.device_key.key
event_type = event.event_type
if event_class not in domain_data.discovered_event_classes:
domain_data.discovered_event_classes.add(event_class)
hass.config_entries.async_update_entry(
entry,
data=entry.data
| {
CONF_DISCOVERED_EVENT_CLASSES: list(
domain_data.discovered_event_classes
)
},
)
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,
)
),
)
# If payload is encrypted and the bindkey is not verified then we need to reauth
if data.encryption_scheme != EncryptionScheme.NONE and not data.bindkey_verified:
entry.async_start_reauth(hass, data={"device": data})
@ -45,10 +97,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
assert address is not None
kwargs = {}
if bindkey := entry.data.get("bindkey"):
kwargs["bindkey"] = bytes.fromhex(bindkey)
if bindkey := entry.data.get(CONF_BINDKEY):
kwargs[CONF_BINDKEY] = bytes.fromhex(bindkey)
data = BTHomeBluetoothDeviceData(**kwargs)
device_registry = async_get(hass)
coordinator = hass.data.setdefault(DOMAIN, {})[
entry.entry_id
] = PassiveBluetoothProcessorCoordinator(
@ -57,11 +110,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
address=address,
mode=BluetoothScanningMode.PASSIVE,
update_method=lambda service_info: process_service_info(
hass, entry, data, service_info
hass, entry, data, service_info, device_registry
),
connectable=False,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
domain_data = BTHomeData(set(entry.data.get(CONF_DISCOVERED_EVENT_CLASSES, [])))
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = domain_data
entry.async_on_unload(
coordinator.async_start()
) # only start after all platforms have had a chance to subscribe

29
homeassistant/components/bthome/const.py
View File

@ -1,3 +1,32 @@
"""Constants for the BTHome Bluetooth integration."""
from __future__ import annotations
from typing import Final, TypedDict
DOMAIN = "bthome"
CONF_BINDKEY: Final = "bindkey"
CONF_DISCOVERED_EVENT_CLASSES: Final = "known_events"
CONF_SUBTYPE: Final = "subtype"
EVENT_TYPE: Final = "event_type"
EVENT_CLASS: Final = "event_class"
EVENT_PROPERTIES: Final = "event_properties"
BTHOME_BLE_EVENT: Final = "bthome_ble_event"
EVENT_CLASS_BUTTON: Final = "button"
EVENT_CLASS_DIMMER: Final = "dimmer"
CONF_EVENT_CLASS: Final = "event_class"
CONF_EVENT_PROPERTIES: Final = "event_properties"
class BTHomeBleEvent(TypedDict):
"""BTHome BLE event data."""
device_id: str
address: str
event_class: str # ie 'button'
event_type: str # ie 'press'
event_properties: dict[str, str | int | float | None] | None

130
homeassistant/components/bthome/device_trigger.py Normal file
View File

@ -0,0 +1,130 @@
"""Provides device triggers for BTHome BLE."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
from homeassistant.components.homeassistant.triggers import event as event_trigger
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_EVENT,
CONF_PLATFORM,
CONF_TYPE,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
from .const import (
BTHOME_BLE_EVENT,
CONF_DISCOVERED_EVENT_CLASSES,
CONF_SUBTYPE,
DOMAIN,
EVENT_CLASS,
EVENT_CLASS_BUTTON,
EVENT_CLASS_DIMMER,
EVENT_TYPE,
)
TRIGGERS_BY_EVENT_CLASS = {
EVENT_CLASS_BUTTON: {
"press",
"double_press",
"triple_press",
"long_press",
"long_double_press",
"long_triple_press",
},
EVENT_CLASS_DIMMER: {"rotate_left", "rotate_right"},
}
SCHEMA_BY_EVENT_CLASS = {
EVENT_CLASS_BUTTON: DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In([EVENT_CLASS_BUTTON]),
vol.Required(CONF_SUBTYPE): vol.In(
TRIGGERS_BY_EVENT_CLASS[EVENT_CLASS_BUTTON]
),
}
),
EVENT_CLASS_DIMMER: DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In([EVENT_CLASS_DIMMER]),
vol.Required(CONF_SUBTYPE): vol.In(
TRIGGERS_BY_EVENT_CLASS[EVENT_CLASS_DIMMER]
),
}
),
}
async def async_validate_trigger_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate trigger config."""
return SCHEMA_BY_EVENT_CLASS.get(config[CONF_TYPE], DEVICE_TRIGGER_BASE_SCHEMA)(
config
)
async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, Any]]:
"""Return a list of triggers for BTHome BLE devices."""
device_registry = dr.async_get(hass)
device = device_registry.async_get(device_id)
assert device is not None
config_entries = [
hass.config_entries.async_get_entry(entry_id)
for entry_id in device.config_entries
]
bthome_config_entry = next(
iter(entry for entry in config_entries if entry and entry.domain == DOMAIN),
None,
)
assert bthome_config_entry is not None
return [
{
# Required fields of TRIGGER_BASE_SCHEMA
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
# Required fields of TRIGGER_SCHEMA
CONF_TYPE: event_class,
CONF_SUBTYPE: event_type,
}
for event_class in bthome_config_entry.data.get(
CONF_DISCOVERED_EVENT_CLASSES, []
)
for event_type in TRIGGERS_BY_EVENT_CLASS.get(event_class, [])
]
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: TriggerActionType,
trigger_info: TriggerInfo,
) -> CALLBACK_TYPE:
"""Attach a trigger."""
return await event_trigger.async_attach_trigger(
hass,
event_trigger.TRIGGER_SCHEMA(
{
event_trigger.CONF_PLATFORM: CONF_EVENT,
event_trigger.CONF_EVENT_TYPE: BTHOME_BLE_EVENT,
event_trigger.CONF_EVENT_DATA: {
CONF_DEVICE_ID: config[CONF_DEVICE_ID],
EVENT_CLASS: config[CONF_TYPE],
EVENT_TYPE: config[CONF_SUBTYPE],
},
}
),
action,
trigger_info,
platform_type="device",
)

11
homeassistant/components/bthome/models.py Normal file
View File

@ -0,0 +1,11 @@
"""The bthome integration models."""
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class BTHomeData:
"""Data for the bthome integration."""
discovered_event_classes: set[str]

16
homeassistant/components/bthome/strings.json
View File

@ -28,5 +28,21 @@
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"device_automation": {
"trigger_subtype": {
"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",
"rotate_right": "Rotate Right",
"rotate_left": "Rotate Left"
},
"trigger_type": {
"button": "Button \"{subtype}\"",
"dimmer": "Dimmer \"{subtype}\""
}
}
}

282
tests/components/bthome/test_device_trigger.py Normal file
View File

@ -0,0 +1,282 @@
"""Test BTHome BLE events."""
import pytest
from homeassistant.components import automation
from homeassistant.components.bluetooth.const import DOMAIN as BLUETOOTH_DOMAIN
from homeassistant.components.bthome.const import CONF_SUBTYPE, DOMAIN
from homeassistant.components.device_automation import DeviceAutomationType
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import (
CONNECTION_NETWORK_MAC,
async_get as async_get_dev_reg,
)
from homeassistant.setup import async_setup_component
from . import make_bthome_v2_adv
from tests.common import (
MockConfigEntry,
async_capture_events,
async_get_device_automations,
async_mock_service,
)
from tests.components.bluetooth import inject_bluetooth_service_info_bleak
@callback
def get_device_id(mac: str) -> tuple[str, str]:
"""Get device registry identifier for bthome_ble."""
return (BLUETOOTH_DOMAIN, mac)
@pytest.fixture
def calls(hass):
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def _async_setup_bthome_device(hass, mac: str):
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=mac,
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
async def test_event_long_press(hass: HomeAssistant) -> None:
"""Make sure that a long press event is fired."""
mac = "A4:C1:38:8D:18:B2"
entry = await _async_setup_bthome_device(hass, mac)
events = async_capture_events(hass, "bthome_ble_event")
# Emit long press event
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"\x40\x3A\x04"),
)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["address"] == "A4:C1:38:8D:18:B2"
assert events[0].data["event_type"] == "long_press"
assert events[0].data["event_properties"] is None
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_event_rotate_dimmer(hass: HomeAssistant) -> None:
"""Make sure that a rotate dimmer event is fired."""
mac = "A4:C1:38:8D:18:B2"
entry = await _async_setup_bthome_device(hass, mac)
events = async_capture_events(hass, "bthome_ble_event")
# Emit rotate dimmer 3 steps left event
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"\x40\x3C\x01\x03"),
)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["address"] == "A4:C1:38:8D:18:B2"
assert events[0].data["event_type"] == "rotate_left"
assert events[0].data["event_properties"] == {"steps": 3}
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_get_triggers_button(hass: HomeAssistant) -> None:
"""Test that we get the expected triggers from a BTHome BLE sensor."""
mac = "A4:C1:38:8D:18:B2"
entry = await _async_setup_bthome_device(hass, mac)
events = async_capture_events(hass, "bthome_ble_event")
# Emit long press event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"\x40\x3A\x04"),
)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 1
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device({get_device_id(mac)})
assert device
expected_trigger = {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "button",
CONF_SUBTYPE: "long_press",
"metadata": {},
}
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, device.id
)
assert expected_trigger in triggers
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_get_triggers_dimmer(hass: HomeAssistant) -> None:
"""Test that we get the expected triggers from a BTHome BLE sensor."""
mac = "A4:C1:38:8D:18:B2"
entry = await _async_setup_bthome_device(hass, mac)
events = async_capture_events(hass, "bthome_ble_event")
# Emit rotate left with 3 steps event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"\x40\x3C\x01\x03"),
)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 1
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device({get_device_id(mac)})
assert device
expected_trigger = {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "dimmer",
CONF_SUBTYPE: "rotate_left",
"metadata": {},
}
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, device.id
)
assert expected_trigger in triggers
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_get_triggers_for_invalid_bthome_ble_device(hass: HomeAssistant) -> None:
"""Test that we don't get triggers for an invalid device."""
mac = "A4:C1:38:8D:18:B2"
entry = await _async_setup_bthome_device(hass, mac)
events = async_capture_events(hass, "bthome_ble_event")
# Creates the device in the registry but no events
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"\x40\x02\xca\x09\x03\xbf\x13"),
)
# wait to make sure there are no events
await hass.async_block_till_done()
assert len(events) == 0
dev_reg = async_get_dev_reg(hass)
invalid_device = dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, "invdevmac")},
)
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, invalid_device.id
)
assert triggers == []
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_get_triggers_for_invalid_device_id(hass: HomeAssistant) -> None:
"""Test that we don't get triggers when using an invalid device_id."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_bthome_device(hass, mac)
# Emit motion detected event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"@0\xdd\x03$\x03\x00\x01\x01"),
)
# wait for the event
await hass.async_block_till_done()
dev_reg = async_get_dev_reg(hass)
invalid_device = dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
connections={(CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
assert invalid_device
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, invalid_device.id
)
assert triggers == []
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_if_fires_on_motion_detected(hass: HomeAssistant, calls) -> None:
"""Test for motion event trigger firing."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_bthome_device(hass, mac)
# Emit a button event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"\x40\x3A\x03"),
)
# # wait for the event
await hass.async_block_till_done()
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device({get_device_id(mac)})
device_id = device.id
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
CONF_TYPE: "button",
CONF_SUBTYPE: "long_press",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_button_long_press"},
},
},
]
},
)
# Emit long press event
inject_bluetooth_service_info_bleak(
hass,
make_bthome_v2_adv(mac, b"\x40\x3A\x04"),
)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_button_long_press"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()