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

Support polling the MiFlora battery (#76342)

pull/76607/head
Jc2k 2022-08-11 09:10:18 +01:00 committed by GitHub
parent 7fc2a73c88
commit 6ad2708946
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 101 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -11,8 +11,8 @@ from homeassistant.components.bluetooth import (
BluetoothScanningMode,
BluetoothServiceInfoBleak,
)
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothProcessorCoordinator,
from homeassistant.components.bluetooth.active_update_coordinator import (
ActiveBluetoothProcessorCoordinator,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
@ -56,9 +56,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
kwargs["bindkey"] = bytes.fromhex(bindkey)
data = XiaomiBluetoothDeviceData(**kwargs)
def _needs_poll(
service_info: BluetoothServiceInfoBleak, last_poll: float | None
) -> bool:
return data.poll_needed(service_info, last_poll)
async def _async_poll(service_info: BluetoothServiceInfoBleak):
# BluetoothServiceInfoBleak is defined in HA, otherwise would just pass it
# directly to the Xiaomi code
return await data.async_poll(service_info.device)
coordinator = hass.data.setdefault(DOMAIN, {})[
entry.entry_id
] = PassiveBluetoothProcessorCoordinator(
] = ActiveBluetoothProcessorCoordinator(
hass,
_LOGGER,
address=address,
@ -66,6 +76,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
update_method=lambda service_info: process_service_info(
hass, entry, data, service_info
),
needs_poll_method=_needs_poll,
poll_method=_async_poll,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(

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

@ -8,7 +8,7 @@
"service_data_uuid": "0000fe95-0000-1000-8000-00805f9b34fb"
}
],
"requirements": ["xiaomi-ble==0.6.4"],
"requirements": ["xiaomi-ble==0.8.1"],
"dependencies": ["bluetooth"],
"codeowners": ["@Jc2k", "@Ernst79"],
"iot_class": "local_push"

2
requirements_all.txt
View File

@ -2480,7 +2480,7 @@ xbox-webapi==2.0.11
xboxapi==2.0.1
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.6.4
xiaomi-ble==0.8.1
# homeassistant.components.knx
xknx==0.22.1

2
requirements_test_all.txt
View File

@ -1678,7 +1678,7 @@ wolf_smartset==0.1.11
xbox-webapi==2.0.11
# homeassistant.components.xiaomi_ble
xiaomi-ble==0.6.4
xiaomi-ble==0.8.1
# homeassistant.components.knx
xknx==0.22.1

34
tests/components/xiaomi_ble/__init__.py
View File

@ -1,21 +1,26 @@
"""Tests for the SensorPush integration."""
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
NOT_SENSOR_PUSH_SERVICE_INFO = BluetoothServiceInfo(
NOT_SENSOR_PUSH_SERVICE_INFO = BluetoothServiceInfoBleak(
name="Not it",
address="00:00:00:00:00:00",
device=BLEDevice("00:00:00:00:00:00", None),
rssi=-63,
manufacturer_data={3234: b"\x00\x01"},
service_data={},
service_uuids=[],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
)
LYWSDCGQ_SERVICE_INFO = BluetoothServiceInfo(
LYWSDCGQ_SERVICE_INFO = BluetoothServiceInfoBleak(
name="LYWSDCGQ",
address="58:2D:34:35:93:21",
device=BLEDevice("00:00:00:00:00:00", None),
rssi=-63,
manufacturer_data={},
service_data={
@ -23,11 +28,13 @@ LYWSDCGQ_SERVICE_INFO = BluetoothServiceInfo(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
)
MMC_T201_1_SERVICE_INFO = BluetoothServiceInfo(
MMC_T201_1_SERVICE_INFO = BluetoothServiceInfoBleak(
name="MMC_T201_1",
address="00:81:F9:DD:6F:C1",
device=BLEDevice("00:00:00:00:00:00", None),
rssi=-56,
manufacturer_data={},
service_data={
@ -35,11 +42,13 @@ MMC_T201_1_SERVICE_INFO = BluetoothServiceInfo(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
)
JTYJGD03MI_SERVICE_INFO = BluetoothServiceInfo(
JTYJGD03MI_SERVICE_INFO = BluetoothServiceInfoBleak(
name="JTYJGD03MI",
address="54:EF:44:E3:9C:BC",
device=BLEDevice("00:00:00:00:00:00", None),
rssi=-56,
manufacturer_data={},
service_data={
@ -47,11 +56,13 @@ JTYJGD03MI_SERVICE_INFO = BluetoothServiceInfo(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
)
YLKG07YL_SERVICE_INFO = BluetoothServiceInfo(
YLKG07YL_SERVICE_INFO = BluetoothServiceInfoBleak(
name="YLKG07YL",
address="F8:24:41:C5:98:8B",
device=BLEDevice("00:00:00:00:00:00", None),
rssi=-56,
manufacturer_data={},
service_data={
@ -59,11 +70,13 @@ YLKG07YL_SERVICE_INFO = BluetoothServiceInfo(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
)
MISSING_PAYLOAD_ENCRYPTED = BluetoothServiceInfo(
MISSING_PAYLOAD_ENCRYPTED = BluetoothServiceInfoBleak(
name="LYWSD02MMC",
address="A4:C1:38:56:53:84",
device=BLEDevice("00:00:00:00:00:00", None),
rssi=-56,
manufacturer_data={},
service_data={
@ -71,14 +84,16 @@ MISSING_PAYLOAD_ENCRYPTED = BluetoothServiceInfo(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
)
def make_advertisement(address: str, payload: bytes) -> BluetoothServiceInfo:
def make_advertisement(address: str, payload: bytes) -> BluetoothServiceInfoBleak:
"""Make a dummy advertisement."""
return BluetoothServiceInfo(
return BluetoothServiceInfoBleak(
name="Test Device",
address=address,
device=BLEDevice(address, None),
rssi=-56,
manufacturer_data={},
service_data={
@ -86,4 +101,5 @@ def make_advertisement(address: str, payload: bytes) -> BluetoothServiceInfo:
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Test Device"),
)

47
tests/components/xiaomi_ble/conftest.py
View File

@ -1,8 +1,55 @@
"""Session fixtures."""
from unittest import mock
import pytest
class MockServices:
"""Mock GATTServicesCollection."""
def get_characteristic(self, key: str) -> str:
"""Mock GATTServicesCollection.get_characteristic."""
return key
class MockBleakClient:
"""Mock BleakClient."""
services = MockServices()
def __init__(self, *args, **kwargs):
"""Mock BleakClient."""
pass
async def __aenter__(self, *args, **kwargs):
"""Mock BleakClient.__aenter__."""
return self
async def __aexit__(self, *args, **kwargs):
"""Mock BleakClient.__aexit__."""
pass
async def connect(self, *args, **kwargs):
"""Mock BleakClient.connect."""
pass
async def disconnect(self, *args, **kwargs):
"""Mock BleakClient.disconnect."""
pass
class MockBleakClientBattery5(MockBleakClient):
"""Mock BleakClient that returns a battery level of 5."""
async def read_gatt_char(self, *args, **kwargs) -> bytes:
"""Mock BleakClient.read_gatt_char."""
return b"\x05\x001.2.3"
@pytest.fixture(autouse=True)
def mock_bluetooth(enable_bluetooth):
"""Auto mock bluetooth."""
with mock.patch("xiaomi_ble.parser.BleakClient", MockBleakClientBattery5):
yield

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

@ -78,7 +78,7 @@ async def test_xiaomi_formaldeyhde(hass):
# obj type is 0x1010, payload len is 0x2 and payload is 0xf400
saved_callback(
make_advertisement(
"C4:7C:8D:6A:3E:7A", b"q \x98\x00iz>j\x8d|\xc4\r\x10\x10\x02\xf4\x00"
"C4:7C:8D:6A:3E:7A", b"q \x5d\x01iz>j\x8d|\xc4\r\x10\x10\x02\xf4\x00"
),
BluetoothChange.ADVERTISEMENT,
)
@ -125,7 +125,7 @@ async def test_xiaomi_consumable(hass):
# obj type is 0x1310, payload len is 0x2 and payload is 0x6000
saved_callback(
make_advertisement(
"C4:7C:8D:6A:3E:7A", b"q \x98\x00iz>j\x8d|\xc4\r\x13\x10\x02\x60\x00"
"C4:7C:8D:6A:3E:7A", b"q \x5d\x01iz>j\x8d|\xc4\r\x13\x10\x02\x60\x00"
),
BluetoothChange.ADVERTISEMENT,
)
@ -172,7 +172,7 @@ async def test_xiaomi_battery_voltage(hass):
# obj type is 0x0a10, payload len is 0x2 and payload is 0x6400
saved_callback(
make_advertisement(
"C4:7C:8D:6A:3E:7A", b"q \x98\x00iz>j\x8d|\xc4\r\x0a\x10\x02\x64\x00"
"C4:7C:8D:6A:3E:7A", b"q \x5d\x01iz>j\x8d|\xc4\r\x0a\x10\x02\x64\x00"
),
BluetoothChange.ADVERTISEMENT,
)
@ -246,7 +246,7 @@ async def test_xiaomi_HHCCJCY01(hass):
BluetoothChange.ADVERTISEMENT,
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 4
assert len(hass.states.async_all()) == 5
illum_sensor = hass.states.get("sensor.test_device_illuminance")
illum_sensor_attr = illum_sensor.attributes
@ -276,6 +276,13 @@ async def test_xiaomi_HHCCJCY01(hass):
assert temp_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "°C"
assert temp_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
batt_sensor = hass.states.get("sensor.test_device_battery")
batt_sensor_attribtes = batt_sensor.attributes
assert batt_sensor.state == "5"
assert batt_sensor_attribtes[ATTR_FRIENDLY_NAME] == "Test Device Battery"
assert batt_sensor_attribtes[ATTR_UNIT_OF_MEASUREMENT] == "%"
assert batt_sensor_attribtes[ATTR_STATE_CLASS] == "measurement"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()