github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add events for xiaomi-ble (#85139) 

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Ernst Klamer 2023-01-24 19:48:30 +01:00 committed by GitHub
parent 90fc8dd860
commit 886d2fc3a1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 573 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
homeassistant/components/xiaomi_ble/__init__.py
View File

@ -8,6 +8,7 @@ from xiaomi_ble.parser import EncryptionScheme
from homeassistant import config_entries
from homeassistant.components.bluetooth import (
DOMAIN as BLUETOOTH_DOMAIN,
BluetoothScanningMode,
BluetoothServiceInfoBleak,
async_ble_device_from_address,
@ -18,8 +19,9 @@ from homeassistant.components.bluetooth.active_update_processor import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import CoreState, HomeAssistant
from homeassistant.helpers.device_registry import DeviceRegistry, async_get
from .const import DOMAIN
from .const import DOMAIN, XIAOMI_BLE_EVENT, XiaomiBleEvent
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
@ -31,9 +33,35 @@ def process_service_info(
entry: config_entries.ConfigEntry,
data: XiaomiBluetoothDeviceData,
service_info: BluetoothServiceInfoBleak,
device_registry: DeviceRegistry,
) -> SensorUpdate:
"""Process a BluetoothServiceInfoBleak, running side effects and returning sensor data."""
update = data.update(service_info)
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,
)
hass.bus.async_fire(
XIAOMI_BLE_EVENT,
dict(
XiaomiBleEvent(
device_id=device.id,
address=address,
event_type=event.event_type,
event_properties=event.event_properties,
)
),
)
# If device isn't pending we know it has seen at least one broadcast with a payload
# If that payload was encrypted and the bindkey was not verified then we need to reauth
@ -91,6 +119,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
return await data.async_poll(connectable_device)
device_registry = async_get(hass)
coordinator = hass.data.setdefault(DOMAIN, {})[
entry.entry_id
] = ActiveBluetoothProcessorCoordinator(
@ -99,7 +128,7 @@ 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
),
needs_poll_method=_needs_poll,
poll_method=_async_poll,

18
homeassistant/components/xiaomi_ble/const.py
View File

@ -1,3 +1,21 @@
"""Constants for the Xiaomi Bluetooth integration."""
from __future__ import annotations
from typing import Final, TypedDict
DOMAIN = "xiaomi_ble"
CONF_EVENT_PROPERTIES: Final = "event_properties"
EVENT_PROPERTIES: Final = "event_properties"
EVENT_TYPE: Final = "event_type"
XIAOMI_BLE_EVENT: Final = "xiaomi_ble_event"
class XiaomiBleEvent(TypedDict):
"""Xiaomi BLE event data."""
device_id: str
address: str
event_type: str
event_properties: dict[str, str | int | float | None] | None

127
homeassistant/components/xiaomi_ble/device_trigger.py Normal file
View File

@ -0,0 +1,127 @@
"""Provides device triggers for Xiaomi BLE."""
from __future__ import annotations
from dataclasses import dataclass
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 (
CONF_EVENT_PROPERTIES,
DOMAIN,
EVENT_PROPERTIES,
EVENT_TYPE,
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]
),
}
)
@dataclass
class TriggerModelData:
"""Data class for trigger model data."""
triggers: list[dict[str, Any]]
schema: vol.Schema
MODEL_DATA = {
"MUE4094RT": TriggerModelData(
triggers=MOTION_DEVICE_TRIGGERS, schema=MOTION_DEVICE_SCHEMA
)
}
async def async_validate_trigger_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate trigger config."""
device_id = config[CONF_DEVICE_ID]
if model_data := _async_trigger_model_data(hass, device_id):
return model_data.schema(config)
return config
async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, Any]]:
"""List a list of triggers for Xiaomi BLE devices."""
# Check if device is a model supporting device triggers.
if not (model_data := _async_trigger_model_data(hass, device_id)):
return []
return [
{
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: device_id,
**trigger,
}
for trigger in model_data.triggers
]
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: TriggerActionType,
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,
}
),
action,
trigger_info,
platform_type="device",
)
def _async_trigger_model_data(
hass: HomeAssistant, device_id: str
) -> TriggerModelData | None:
"""Get available triggers for a given model."""
device_registry = dr.async_get(hass)
device = device_registry.async_get(device_id)
if device and device.model and (model_data := MODEL_DATA.get(device.model)):
return model_data
return None

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

@ -13,8 +13,8 @@
"service_data_uuid": "0000fe95-0000-1000-8000-00805f9b34fb"
}
],
"requirements": ["xiaomi-ble==0.14.3"],
"dependencies": ["bluetooth_adapters"],
"requirements": ["xiaomi-ble==0.15.0"],
"codeowners": ["@Jc2k", "@Ernst79"],
"iot_class": "local_push"
}

5
homeassistant/components/xiaomi_ble/strings.json
View File

@ -38,5 +38,10 @@
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"device_automation": {
"trigger_type": {
"motion_detected": "Motion detected"
}
}
}

5
homeassistant/components/xiaomi_ble/translations/en.json
View File

@ -38,5 +38,10 @@
"description": "Choose a device to set up"
}
}
},
"device_automation": {
"trigger_type": {
"motion_detected": "Motion detected"
}
}
}

2
requirements_all.txt
View File

@ -2631,7 +2631,7 @@ xbox-webapi==2.0.11
xboxapi==2.0.1
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.14.3
xiaomi-ble==0.15.0
# homeassistant.components.knx
xknx==2.3.0

2
requirements_test_all.txt
View File

@ -1856,7 +1856,7 @@ wolf_smartset==0.1.11
xbox-webapi==2.0.11
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.14.3
xiaomi-ble==0.15.0
# homeassistant.components.knx
xknx==2.3.0

383
tests/components/xiaomi_ble/test_device_trigger.py Normal file
View File

@ -0,0 +1,383 @@
"""Test Xiaomi BLE events."""
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.core import callback
from homeassistant.helpers import device_registry
from homeassistant.helpers.device_registry import async_get as async_get_dev_reg
from homeassistant.setup import async_setup_component
from . import make_advertisement
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 xiaomi_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_xiaomi_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_motion_detected(hass):
"""Make sure that a motion detected event is fired."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
events = async_capture_events(hass, "xiaomi_ble_event")
# Emit motion detected event
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(mac, b"@0\xdd\x03$\x03\x00\x01\x01"),
)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["address"] == "DE:70:E8:B2:39:0C"
assert events[0].data["event_type"] == "motion_detected"
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_get_triggers(hass):
"""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)
events = async_capture_events(hass, "xiaomi_ble_event")
# Emit motion detected event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(mac, b"@0\xdd\x03$\x03\x00\x01\x01"),
)
# 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: "motion_detected",
CONF_EVENT_PROPERTIES: None,
"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_xiami_ble_device(hass):
"""Test that we don't get triggers for an invalid device."""
mac = "DE:70:E8:B2:39:0C"
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
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(mac, b"@0\xdd\x03$\x03\x00\x01\x01"),
)
# wait for the event
await hass.async_block_till_done()
assert len(events) == 1
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):
"""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)
# Emit motion detected event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(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={(device_registry.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, calls):
"""Test for motion event trigger firing."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
# Emit motion detected event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(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)
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: "motion_detected",
CONF_EVENT_PROPERTIES: None,
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_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)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_motion_detected"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_automation_with_invalid_trigger_type(hass, caplog):
"""Test for automation with invalid trigger type."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
# Emit motion detected event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(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)
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: "invalid",
CONF_EVENT_PROPERTIES: None,
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_motion_detected"},
},
},
]
},
)
# Logs should return message to make sure event type is of one ["motion_detected"]
assert "motion_detected" in caplog.text
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_automation_with_invalid_trigger_event_property(hass, caplog):
"""Test for automation with invalid trigger event property."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
# Emit motion detected event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(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)
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: "motion_detected",
CONF_EVENT_PROPERTIES: "invalid_property",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_motion_detected"},
},
},
]
},
)
# Logs should return message to make sure event property is of one [None] for motion event
assert str([None]) 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, calls):
"""Test invalid model doesn't return triggers."""
mac = "DE:70:E8:B2:39:0C"
entry = await _async_setup_xiaomi_device(hass, mac)
# Emit motion detected event so it creates the device in the registry
inject_bluetooth_service_info_bleak(
hass,
make_advertisement(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)
# modify model to invalid model
invalid_model = dev_reg.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, mac)},
model="invalid model",
)
invalid_model_id = invalid_model.id
# setup automation to validate trigger config
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: invalid_model_id,
CONF_TYPE: "motion_detected",
CONF_EVENT_PROPERTIES: None,
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_motion_detected"},
},
},
]
},
)
triggers = await async_get_device_automations(
hass, DeviceAutomationType.TRIGGER, invalid_model_id
)
assert triggers == []
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()

3
tests/components/xiaomi_ble/test_sensor.py
View File

@ -1,5 +1,4 @@
"""Test the Xiaomi config flow."""
"""Test Xiaomi BLE sensors."""
from homeassistant.components.sensor import ATTR_STATE_CLASS
from homeassistant.components.xiaomi_ble.const import DOMAIN