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

Add event entity to Xiaomi-BLE integration (#108811)

Browse source
This commit is contained in:
Ernst Klamer 2024-01-30 19:42:56 +01:00 • committed by GitHub
parent 2b534af960
commit 70ee6a16ee
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 651 additions and 136 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import cast
from xiaomi_ble import EncryptionScheme, SensorUpdate, XiaomiBluetoothDeviceData
@ -20,6 +21,7 @@ from homeassistant.helpers.device_registry import (
DeviceRegistry,
async_get,
)
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import (
CONF_DISCOVERED_EVENT_CLASSES,
@ -30,7 +32,7 @@ from .const import (
)
from .coordinator import XiaomiActiveBluetoothProcessorCoordinator
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: XiaomiActiveBluetoothProcessorCoordinator = 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 = XiaomiBleEvent(
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(
XIAOMI_BLE_EVENT,
dict(
XiaomiBleEvent(
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(XIAOMI_BLE_EVENT, cast(dict, ble_event))
async_dispatcher_send(
hass,
format_event_dispatcher_name(address, event_class),
ble_event,
)
# If device isn't pending we know it has seen at least one broadcast with a payload
@ -103,6 +112,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 Xiaomi BLE device from a config entry."""
address = entry.unique_id
@ -160,9 +179,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
),
needs_poll_method=_needs_poll,
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, [])),
poll_method=_async_poll,
# We will take advertisements from non-connectable devices
# since we will trade the BLEDevice for a connectable one

16
homeassistant/components/xiaomi_ble/const.py
View file

@ -7,12 +7,24 @@ DOMAIN = "xiaomi_ble"
CONF_DISCOVERED_EVENT_CLASSES: Final = "known_events"
CONF_SLEEPY_DEVICE: Final = "sleepy_device"
CONF_EVENT_PROPERTIES: Final = "event_properties"
EVENT_PROPERTIES: Final = "event_properties"
CONF_EVENT_CLASS: Final = "event_class"
CONF_SLEEPY_DEVICE: Final = "sleepy_device"
CONF_SUBTYPE: Final = "subtype"
EVENT_CLASS: Final = "event_class"
EVENT_TYPE: Final = "event_type"
EVENT_SUBTYPE: Final = "event_subtype"
EVENT_PROPERTIES: Final = "event_properties"
XIAOMI_BLE_EVENT: Final = "xiaomi_ble_event"
EVENT_CLASS_BUTTON: Final = "button"
EVENT_CLASS_MOTION: Final = "motion"
BUTTON_PRESS: Final = "button_press"
BUTTON_PRESS_DOUBLE_LONG: Final = "button_press_double_long"
MOTION_DEVICE: Final = "motion_device"
class XiaomiBleEvent(TypedDict):
"""Xiaomi BLE event data."""

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

@ -35,7 +35,7 @@ class XiaomiActiveBluetoothProcessorCoordinator(ActiveBluetoothProcessorCoordina
update_method: Callable[[BluetoothServiceInfoBleak], Any],
needs_poll_method: Callable[[BluetoothServiceInfoBleak, float | None], bool],
device_data: XiaomiBluetoothDeviceData,
discovered_device_classes: set[str],
discovered_event_classes: set[str],
poll_method: Callable[
[BluetoothServiceInfoBleak],
Coroutine[Any, Any, Any],
@ -57,7 +57,7 @@ class XiaomiActiveBluetoothProcessorCoordinator(ActiveBluetoothProcessorCoordina
poll_debouncer=poll_debouncer,
connectable=connectable,
)
self.discovered_device_classes = discovered_device_classes
self.discovered_event_classes = discovered_event_classes
self.device_data = device_data
self.entry = entry

106
homeassistant/components/xiaomi_ble/device_trigger.py
View file

@ -21,41 +21,83 @@ from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_EVENT_PROPERTIES,
BUTTON_PRESS,
BUTTON_PRESS_DOUBLE_LONG,
CONF_SUBTYPE,
DOMAIN,
EVENT_PROPERTIES,
EVENT_CLASS,
EVENT_CLASS_BUTTON,
EVENT_CLASS_MOTION,
EVENT_TYPE,
MOTION_DEVICE,
XIAOMI_BLE_EVENT,
)
MOTION_DEVICE_TRIGGERS = [
{CONF_TYPE: "motion_detected", CONF_EVENT_PROPERTIES: None},
]
MOTION_DEVICE_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(
[trigger[CONF_TYPE] for trigger in MOTION_DEVICE_TRIGGERS]
),
vol.Optional(CONF_EVENT_PROPERTIES): vol.In(
[trigger[CONF_EVENT_PROPERTIES] for trigger in MOTION_DEVICE_TRIGGERS]
),
}
)
TRIGGERS_BY_TYPE = {
BUTTON_PRESS: ["press"],
BUTTON_PRESS_DOUBLE_LONG: ["press", "double_press", "long_press"],
MOTION_DEVICE: ["motion_detected"],
}
@dataclass
class TriggerModelData:
"""Data class for trigger model data."""
triggers: list[dict[str, Any]]
schema: vol.Schema
event_class: str
triggers: list[str]
TRIGGER_MODEL_DATA = {
BUTTON_PRESS: TriggerModelData(
schema=DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In([EVENT_CLASS_BUTTON]),
vol.Required(CONF_SUBTYPE): vol.In(TRIGGERS_BY_TYPE[BUTTON_PRESS]),
}
),
event_class=EVENT_CLASS_BUTTON,
triggers=TRIGGERS_BY_TYPE[BUTTON_PRESS],
),
BUTTON_PRESS_DOUBLE_LONG: TriggerModelData(
schema=DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In([EVENT_CLASS_BUTTON]),
vol.Required(CONF_SUBTYPE): vol.In(
TRIGGERS_BY_TYPE[BUTTON_PRESS_DOUBLE_LONG]
),
}
),
event_class=EVENT_CLASS_BUTTON,
triggers=TRIGGERS_BY_TYPE[BUTTON_PRESS_DOUBLE_LONG],
),
MOTION_DEVICE: TriggerModelData(
schema=DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In([EVENT_CLASS_MOTION]),
vol.Required(CONF_SUBTYPE): vol.In(TRIGGERS_BY_TYPE[MOTION_DEVICE]),
}
),
event_class=EVENT_CLASS_MOTION,
triggers=TRIGGERS_BY_TYPE[MOTION_DEVICE],
),
}
MODEL_DATA = {
"MUE4094RT": TriggerModelData(
triggers=MOTION_DEVICE_TRIGGERS, schema=MOTION_DEVICE_SCHEMA
)
"JTYJGD03MI": TRIGGER_MODEL_DATA[BUTTON_PRESS],
"MS1BB(MI)": TRIGGER_MODEL_DATA[BUTTON_PRESS],
"RTCGQ02LM": TRIGGER_MODEL_DATA[BUTTON_PRESS],
"SJWS01LM": TRIGGER_MODEL_DATA[BUTTON_PRESS],
"K9B-1BTN": TRIGGER_MODEL_DATA[BUTTON_PRESS_DOUBLE_LONG],
"K9B-2BTN": TRIGGER_MODEL_DATA[BUTTON_PRESS_DOUBLE_LONG],
"K9B-3BTN": TRIGGER_MODEL_DATA[BUTTON_PRESS_DOUBLE_LONG],
"K9BB-1BTN": TRIGGER_MODEL_DATA[BUTTON_PRESS_DOUBLE_LONG],
"YLAI003": TRIGGER_MODEL_DATA[BUTTON_PRESS_DOUBLE_LONG],
"XMWXKG01LM": TRIGGER_MODEL_DATA[BUTTON_PRESS_DOUBLE_LONG],
"XMWXKG01YL": TRIGGER_MODEL_DATA[BUTTON_PRESS_DOUBLE_LONG],
"MUE4094RT": TRIGGER_MODEL_DATA[MOTION_DEVICE],
}
@ -77,14 +119,20 @@ async def async_get_triggers(
# Check if device is a model supporting device triggers.
if not (model_data := _async_trigger_model_data(hass, device_id)):
return []
event_type = model_data.event_class
event_subtypes = model_data.triggers
return [
{
# Required fields of TRIGGER_BASE_SCHEMA
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
**trigger,
CONF_DOMAIN: DOMAIN,
# Required fields of TRIGGER_SCHEMA
CONF_TYPE: event_type,
CONF_SUBTYPE: event_subtype,
}
for trigger in model_data.triggers
for event_subtype in event_subtypes
]
@ -95,19 +143,17 @@ async def async_attach_trigger(
trigger_info: TriggerInfo,
) -> CALLBACK_TYPE:
"""Attach a trigger."""
event_data = {
CONF_DEVICE_ID: config[CONF_DEVICE_ID],
EVENT_TYPE: config[CONF_TYPE],
EVENT_PROPERTIES: config[CONF_EVENT_PROPERTIES],
}
return await event_trigger.async_attach_trigger(
hass,
event_trigger.TRIGGER_SCHEMA(
{
event_trigger.CONF_PLATFORM: CONF_EVENT,
event_trigger.CONF_EVENT_TYPE: XIAOMI_BLE_EVENT,
event_trigger.CONF_EVENT_DATA: event_data,
event_trigger.CONF_EVENT_DATA: {
CONF_DEVICE_ID: config[CONF_DEVICE_ID],
EVENT_CLASS: config[CONF_TYPE],
EVENT_TYPE: config[CONF_SUBTYPE],
},
}
),
action,

130
homeassistant/components/xiaomi_ble/event.py Normal file
View file

@ -0,0 +1,130 @@
"""Support for Xiaomi 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_MOTION,
EVENT_PROPERTIES,
EVENT_TYPE,
XiaomiBleEvent,
)
from .coordinator import XiaomiActiveBluetoothProcessorCoordinator
DESCRIPTIONS_BY_EVENT_CLASS = {
EVENT_CLASS_BUTTON: EventEntityDescription(
key=EVENT_CLASS_BUTTON,
translation_key="button",
event_types=[
"press",
"double_press",
"long_press",
],
device_class=EventDeviceClass.BUTTON,
),
EVENT_CLASS_MOTION: EventEntityDescription(
key=EVENT_CLASS_MOTION,
translation_key="motion",
event_types=["motion_detected"],
),
}
class XiaomiEventEntity(EventEntity):
"""Representation of a Xiaomi event entity."""
_attr_should_poll = False
_attr_has_entity_name = True
def __init__(
self,
address: str,
event_class: str,
event: XiaomiBleEvent | None,
) -> None:
"""Initialise a Xiaomi event entity."""
self._update_signal = format_event_dispatcher_name(address, event_class)
# event_class is something like "button" or "motion"
# 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: XiaomiBleEvent) -> 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 Xiaomi event."""
coordinator: XiaomiActiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
address = coordinator.address
ent_reg = er.async_get(hass)
async_add_entities(
# Matches logic in PassiveBluetoothProcessorEntity
XiaomiEventEntity(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: XiaomiBleEvent) -> None:
"""Handle a newly discovered event class with or without a postfix."""
async_add_entities([XiaomiEventEntity(address, event_class, event)])
entry.async_on_unload(
async_dispatcher_connect(
hass,
format_discovered_event_class(address),
_async_discovered_event_class,
)
)

2
homeassistant/components/xiaomi_ble/manifest.json
View file

@ -24,5 +24,5 @@
"dependencies": ["bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/xiaomi_ble",
"iot_class": "local_push",
"requirements": ["xiaomi-ble==0.21.2"]
"requirements": ["xiaomi-ble==0.23.1"]
}

33
homeassistant/components/xiaomi_ble/strings.json
View file

@ -40,8 +40,39 @@
}
},
"device_automation": {
"trigger_subtype": {
"press": "Press",
"double_press": "Double Press",
"long_press": "Long Press",
"motion_detected": "Motion Detected"
},
"trigger_type": {
"motion_detected": "Motion detected"
"button": "Button \"{subtype}\"",
"motion": "{subtype}"
}
},
"entity": {
"event": {
"button": {
"state_attributes": {
"event_type": {
"state": {
"press": "Press",
"double_press": "Double press",
"long_press": "Long press"
}
}
}
},
"motion": {
"state_attributes": {
"event_type": {
"state": {
"motion_detected": "Motion Detected"
}
}
}
}
}
}
}

2
requirements_all.txt
View file

@ -2846,7 +2846,7 @@ wyoming==1.5.2
xbox-webapi==2.0.11
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.21.2
xiaomi-ble==0.23.1
# homeassistant.components.knx
xknx==2.11.2

2
requirements_test_all.txt
View file

@ -2172,7 +2172,7 @@ wyoming==1.5.2
xbox-webapi==2.0.11
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.21.2
xiaomi-ble==0.23.1
# homeassistant.components.knx
xknx==2.11.2

311
tests/components/xiaomi_ble/test_device_trigger.py
View file

@ -4,27 +4,19 @@ import pytest
from homeassistant.components import automation
from homeassistant.components.bluetooth.const import DOMAIN as BLUETOOTH_DOMAIN
from homeassistant.components.device_automation import DeviceAutomationType
from homeassistant.components.xiaomi_ble.const import (
CONF_EVENT_PROPERTIES,
DOMAIN,
EVENT_PROPERTIES,
EVENT_TYPE,
XIAOMI_BLE_EVENT,
)
from homeassistant.const import (
CONF_ADDRESS,
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_PLATFORM,
CONF_TYPE,
)
from homeassistant.components.xiaomi_ble.const import CONF_SUBTYPE, DOMAIN
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
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_advertisement
from tests.common import (
Any,
MockConfigEntry,
async_capture_events,
async_get_device_automations,
@ -45,11 +37,8 @@ def calls(hass):
return async_mock_service(hass, "test", "automation")
async def _async_setup_xiaomi_device(hass, mac: str):
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id=mac,
)
async def _async_setup_xiaomi_device(hass, mac: str, data: Any | None = None):
config_entry = MockConfigEntry(domain=DOMAIN, unique_id=mac, data=data)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
@ -58,6 +47,33 @@ async def _async_setup_xiaomi_device(hass, mac: str):
return config_entry
async def test_event_button_press(hass: HomeAssistant) -> None:
"""Make sure that a button press event is fired."""
mac = "54:EF:44:E3:9C:BC"
data = {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
entry = await _async_setup_xiaomi_device(hass, mac, data)
events = async_capture_events(hass, "xiaomi_ble_event")
# Emit button press event
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(
mac,
b'XY\x97\td\xbc\x9c\xe3D\xefT" `' b"\x88\xfd\x00\x00\x00\x00:\x14\x8f\xb3",
),
)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["address"] == "54:EF:44:E3:9C:BC"
assert events[0].data["event_type"] == "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_motion_detected(hass: HomeAssistant) -> None:
"""Make sure that a motion detected event is fired."""
mac = "DE:70:E8:B2:39:0C"
@ -81,9 +97,87 @@ async def test_event_motion_detected(hass: HomeAssistant) -> None:
await hass.async_block_till_done()
async def test_get_triggers(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
async def test_get_triggers_button(hass: HomeAssistant) -> None:
"""Test that we get the expected triggers from a Xiaomi BLE button sensor."""
mac = "54:EF:44:E3:9C:BC"
data = {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
entry = await _async_setup_xiaomi_device(hass, mac, data)
events = async_capture_events(hass, "xiaomi_ble_event")
# Emit button press event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(
mac,
b'XY\x97\td\xbc\x9c\xe3D\xefT" `' b"\x88\xfd\x00\x00\x00\x00:\x14\x8f\xb3",
),
)
# 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(identifiers={get_device_id(mac)})
assert device
expected_trigger = {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "button",
CONF_SUBTYPE: "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_double_button(hass: HomeAssistant) -> None:
"""Test that we get the expected triggers from a Xiaomi BLE switch with 2 buttons."""
mac = "DC:ED:83:87:12:73"
data = {"bindkey": "b93eb3787eabda352edd94b667f5d5a9"}
entry = await _async_setup_xiaomi_device(hass, mac, data)
events = async_capture_events(hass, "xiaomi_ble_event")
# Emit button press event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(
mac,
b"XYI\x19Os\x12\x87\x83\xed\xdc\x0b48\n\x02\x00\x00\x8dI\xae(",
),
)
# 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(identifiers={get_device_id(mac)})
assert device
expected_trigger = {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "button",
CONF_SUBTYPE: "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_motion(hass: HomeAssistant) -> None:
"""Test that we get the expected triggers from a Xiaomi BLE motion sensor."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
@ -99,14 +193,15 @@ async def test_get_triggers(
await hass.async_block_till_done()
assert len(events) == 1
device = device_registry.async_get_device(identifiers={get_device_id(mac)})
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device(identifiers={get_device_id(mac)})
assert device
expected_trigger = {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device.id,
CONF_TYPE: "motion_detected",
CONF_EVENT_PROPERTIES: None,
CONF_TYPE: "motion",
CONF_SUBTYPE: "motion_detected",
"metadata": {},
}
triggers = await async_get_device_automations(
@ -118,25 +213,24 @@ async def test_get_triggers(
await hass.async_block_till_done()
async def test_get_triggers_for_invalid_xiami_ble_device(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
"""Test that we don't get triggers for an invalid device."""
mac = "DE:70:E8:B2:39:0C"
async def test_get_triggers_for_invalid_xiami_ble_device(hass: HomeAssistant) -> None:
"""Test that we don't get triggers for an device that does not emit events."""
mac = "C4:7C:8D:6A:3E:7A"
entry = await _async_setup_xiaomi_device(hass, mac)
events = async_capture_events(hass, "xiaomi_ble_event")
# Emit motion detected event so it creates the device in the registry
# Creates the device in the registry but no events
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(mac, b"@0\xdd\x03$\x03\x00\x01\x01"),
make_advertisement(mac, b"q \x5d\x01iz>j\x8d|\xc4\r\x10\x10\x02\xf4\x00"),
)
# wait for the event
# wait to make sure there are no events
await hass.async_block_till_done()
assert len(events) == 1
assert len(events) == 0
invalid_device = device_registry.async_get_or_create(
dev_reg = async_get_dev_reg(hass)
invalid_device = dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, "invdevmac")},
)
@ -150,9 +244,7 @@ async def test_get_triggers_for_invalid_xiami_ble_device(
await hass.async_block_till_done()
async def test_get_triggers_for_invalid_device_id(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
) -> None:
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_xiaomi_device(hass, mac)
@ -166,9 +258,11 @@ async def test_get_triggers_for_invalid_device_id(
# wait for the event
await hass.async_block_till_done()
invalid_device = device_registry.async_get_or_create(
dev_reg = async_get_dev_reg(hass)
invalid_device = dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
connections={(CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
assert invalid_device
triggers = await async_get_device_automations(
@ -180,23 +274,26 @@ async def test_get_triggers_for_invalid_device_id(
await hass.async_block_till_done()
async def test_if_fires_on_motion_detected(
hass: HomeAssistant, calls, device_registry: dr.DeviceRegistry
) -> None:
"""Test for motion event trigger firing."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
async def test_if_fires_on_button_press(hass: HomeAssistant, calls) -> None:
"""Test for button press event trigger firing."""
mac = "54:EF:44:E3:9C:BC"
data = {"bindkey": "5b51a7c91cde6707c9ef18dfda143a58"}
entry = await _async_setup_xiaomi_device(hass, mac, data)
# Emit motion detected event so it creates the device in the registry
# Creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(mac, b"@0\xdd\x03$\x03\x00\x01\x01"),
make_advertisement(
mac,
b"XY\x97\tf\xbc\x9c\xe3D\xefT\x01" b"\x08\x12\x05\x00\x00\x00q^\xbe\x90",
),
)
# wait for the event
# wait for the device being created
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={get_device_id(mac)})
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device(identifiers={get_device_id(mac)})
device_id = device.id
assert await async_setup_component(
@ -209,8 +306,64 @@ async def test_if_fires_on_motion_detected(
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
CONF_TYPE: "motion_detected",
CONF_EVENT_PROPERTIES: None,
CONF_TYPE: "button",
CONF_SUBTYPE: "press",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_button_press"},
},
},
]
},
)
# Emit button press event
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(
mac,
b'XY\x97\td\xbc\x9c\xe3D\xefT" `' b"\x88\xfd\x00\x00\x00\x00:\x14\x8f\xb3",
),
)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_button_press"
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_xiaomi_device(hass, mac)
# Creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(mac, b"@0\xdd\x03$\x0A\x10\x01\x64"),
)
# wait for the device being created
await hass.async_block_till_done()
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device(identifiers={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: "motion",
CONF_SUBTYPE: "motion_detected",
},
"action": {
"service": "test.automation",
@ -220,15 +373,11 @@ async def test_if_fires_on_motion_detected(
]
},
)
message = {
CONF_DEVICE_ID: device_id,
CONF_ADDRESS: "DE:70:E8:B2:39:0C",
EVENT_TYPE: "motion_detected",
EVENT_PROPERTIES: None,
}
hass.bus.async_fire(XIAOMI_BLE_EVENT, message)
# Emit motion detected event
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(mac, b"@0\xdd\x03$\x03\x00\x01\x01"),
)
await hass.async_block_till_done()
assert len(calls) == 1
@ -241,7 +390,6 @@ async def test_if_fires_on_motion_detected(
async def test_automation_with_invalid_trigger_type(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test for automation with invalid trigger type."""
mac = "DE:70:E8:B2:39:0C"
@ -256,7 +404,8 @@ async def test_automation_with_invalid_trigger_type(
# wait for the event
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={get_device_id(mac)})
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device(identifiers={get_device_id(mac)})
device_id = device.id
assert await async_setup_component(
@ -270,7 +419,7 @@ async def test_automation_with_invalid_trigger_type(
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
CONF_TYPE: "invalid",
CONF_EVENT_PROPERTIES: None,
CONF_SUBTYPE: None,
},
"action": {
"service": "test.automation",
@ -290,7 +439,6 @@ async def test_automation_with_invalid_trigger_type(
async def test_automation_with_invalid_trigger_event_property(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test for automation with invalid trigger event property."""
mac = "DE:70:E8:B2:39:0C"
@ -305,7 +453,8 @@ async def test_automation_with_invalid_trigger_event_property(
# wait for the event
await hass.async_block_till_done()
device = device_registry.async_get_device(identifiers={get_device_id(mac)})
dev_reg = async_get_dev_reg(hass)
device = dev_reg.async_get_device(identifiers={get_device_id(mac)})
device_id = device.id
assert await async_setup_component(
@ -318,27 +467,28 @@ async def test_automation_with_invalid_trigger_event_property(
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
CONF_TYPE: "motion_detected",
CONF_EVENT_PROPERTIES: "invalid_property",
CONF_TYPE: "motion",
CONF_SUBTYPE: "invalid_subtype",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_motion_detected"},
"data_template": {
"some": "test_trigger_motion_motion_detected"
},
},
},
]
},
)
# Logs should return message to make sure event property is of one [None] for motion event
assert str([None]) in caplog.text
await hass.async_block_till_done()
# Logs should return message to make sure subtype is of one 'motion_detected' for motion event
assert "value must be one of ['motion_detected']" in caplog.text
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_triggers_for_invalid__model(
hass: HomeAssistant, calls, device_registry: dr.DeviceRegistry
) -> None:
async def test_triggers_for_invalid__model(hass: HomeAssistant, calls) -> None:
"""Test invalid model doesn't return triggers."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
@ -353,7 +503,8 @@ async def test_triggers_for_invalid__model(
await hass.async_block_till_done()
# modify model to invalid model
invalid_model = device_registry.async_get_or_create(
dev_reg = async_get_dev_reg(hass)
invalid_model = dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, mac)},
model="invalid model",
@ -371,12 +522,14 @@ async def test_triggers_for_invalid__model(
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: invalid_model_id,
CONF_TYPE: "motion_detected",
CONF_EVENT_PROPERTIES: None,
CONF_TYPE: "motion",
CONF_SUBTYPE: "motion_detected",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_motion_detected"},
"data_template": {
"some": "test_trigger_motion_motion_detected"
},
},
},
]

126
tests/components/xiaomi_ble/test_event.py Normal file
View file

@ -0,0 +1,126 @@
"""Test the Xiaomi BLE events."""
import pytest
from homeassistant.components.event import ATTR_EVENT_TYPE
from homeassistant.components.xiaomi_ble.const import DOMAIN
from homeassistant.const import ATTR_FRIENDLY_NAME, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from . import make_advertisement
from tests.common import MockConfigEntry
from tests.components.bluetooth import (
BluetoothServiceInfoBleak,
inject_bluetooth_service_info,
)
@pytest.mark.parametrize(
("mac_address", "advertisement", "bind_key", "result"),
[
(
"54:EF:44:E3:9C:BC",
make_advertisement(
"54:EF:44:E3:9C:BC",
b'XY\x97\td\xbc\x9c\xe3D\xefT" `'
b"\x88\xfd\x00\x00\x00\x00:\x14\x8f\xb3",
),
"5b51a7c91cde6707c9ef18dfda143a58",
[
{
"entity": "event.smoke_detector_9cbc_button",
ATTR_FRIENDLY_NAME: "Smoke Detector 9CBC Button",
ATTR_EVENT_TYPE: "press",
}
],
),
(
"DC:ED:83:87:12:73",
make_advertisement(
"DC:ED:83:87:12:73",
b"XYI\x19Os\x12\x87\x83\xed\xdc\x0b48\n\x02\x00\x00\x8dI\xae(",
),
"b93eb3787eabda352edd94b667f5d5a9",
[
{
"entity": "event.switch_double_button_1273_button_right",
ATTR_FRIENDLY_NAME: "Switch (double button) 1273 Button right",
ATTR_EVENT_TYPE: "press",
}
],
),
(
"DE:70:E8:B2:39:0C",
make_advertisement(
"DE:70:E8:B2:39:0C",
b"@0\xdd\x03$\x03\x00\x01\x01",
),
None,
[
{
"entity": "event.nightlight_390c_motion",
ATTR_FRIENDLY_NAME: "Nightlight 390C Motion",
ATTR_EVENT_TYPE: "motion_detected",
}
],
),
],
)
async def test_events(
hass: HomeAssistant,
mac_address: str,
advertisement: BluetoothServiceInfoBleak,
bind_key: str | None,
result: list[dict[str, str]],
) -> None:
"""Test the different Xiaomi BLE 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()