jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-22 16:27:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Handle iBeacons that broadcast multiple different uuids (#79011)

Browse Source 
* Handle iBeacons that broadcast multiple different uuids

* fix flip-flopping between uuids

* naming
This commit is contained in:
J. Nick Koston 2022-09-23 14:45:09 -10:00 committed by GitHub
parent fc58d88770
commit 02731efc4c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 180 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
homeassistant/components/ibeacon/coordinator.py
View File

@ -54,7 +54,7 @@ def make_short_address(address: str) -> str:
@callback
def async_name(
service_info: bluetooth.BluetoothServiceInfoBleak,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
unique_address: bool = False,
) -> str:
"""Return a name for the device."""
@ -62,7 +62,7 @@ def async_name(
service_info.name,
service_info.name.replace("_", ":"),
):
base_name = f"{parsed.uuid} {parsed.major}.{parsed.minor}"
base_name = f"{ibeacon_advertisement.uuid} {ibeacon_advertisement.major}.{ibeacon_advertisement.minor}"
else:
base_name = service_info.name
if unique_address:
@ -77,7 +77,7 @@ def _async_dispatch_update(
hass: HomeAssistant,
device_id: str,
service_info: bluetooth.BluetoothServiceInfoBleak,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
new: bool,
unique_address: bool,
) -> None:
@ -87,15 +87,15 @@ def _async_dispatch_update(
hass,
SIGNAL_IBEACON_DEVICE_NEW,
device_id,
async_name(service_info, parsed, unique_address),
parsed,
async_name(service_info, ibeacon_advertisement, unique_address),
ibeacon_advertisement,
)
return
async_dispatcher_send(
hass,
signal_seen(device_id),
parsed,
ibeacon_advertisement,
)
@ -117,7 +117,9 @@ class IBeaconCoordinator:
)
# iBeacons with fixed MAC addresses
self._last_rssi_by_unique_id: dict[str, int] = {}
self._last_ibeacon_advertisement_by_unique_id: dict[
str, iBeaconAdvertisement
] = {}
self._group_ids_by_address: dict[str, set[str]] = {}
self._unique_ids_by_address: dict[str, set[str]] = {}
self._unique_ids_by_group_id: dict[str, set[str]] = {}
@ -162,21 +164,23 @@ class IBeaconCoordinator:
for unique_id in unique_ids:
if device := self._dev_reg.async_get_device({(DOMAIN, unique_id)}):
self._dev_reg.async_remove_device(device.id)
self._last_rssi_by_unique_id.pop(unique_id, None)
self._last_ibeacon_advertisement_by_unique_id.pop(unique_id, None)
@callback
def _async_convert_random_mac_tracking(
self,
group_id: str,
service_info: bluetooth.BluetoothServiceInfoBleak,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Switch to random mac tracking method when a group is using rotating mac addresses."""
self._group_ids_random_macs.add(group_id)
self._async_purge_untrackable_entities(self._unique_ids_by_group_id[group_id])
self._unique_ids_by_group_id.pop(group_id)
self._addresses_by_group_id.pop(group_id)
self._async_update_ibeacon_with_random_mac(group_id, service_info, parsed)
self._async_update_ibeacon_with_random_mac(
group_id, service_info, ibeacon_advertisement
)
def _async_track_ibeacon_with_unique_address(
self, address: str, group_id: str, unique_id: str
@ -197,49 +201,55 @@ class IBeaconCoordinator:
"""Update from a bluetooth callback."""
if service_info.address in self._ignore_addresses:
return
if not (parsed := parse(service_info)):
if not (ibeacon_advertisement := parse(service_info)):
return
group_id = f"{parsed.uuid}_{parsed.major}_{parsed.minor}"
group_id = f"{ibeacon_advertisement.uuid}_{ibeacon_advertisement.major}_{ibeacon_advertisement.minor}"
if group_id in self._group_ids_random_macs:
self._async_update_ibeacon_with_random_mac(group_id, service_info, parsed)
self._async_update_ibeacon_with_random_mac(
group_id, service_info, ibeacon_advertisement
)
return
self._async_update_ibeacon_with_unique_address(group_id, service_info, parsed)
self._async_update_ibeacon_with_unique_address(
group_id, service_info, ibeacon_advertisement
)
@callback
def _async_update_ibeacon_with_random_mac(
self,
group_id: str,
service_info: bluetooth.BluetoothServiceInfoBleak,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Update iBeacons with random mac addresses."""
new = group_id not in self._last_seen_by_group_id
self._last_seen_by_group_id[group_id] = service_info
self._unavailable_group_ids.discard(group_id)
_async_dispatch_update(self.hass, group_id, service_info, parsed, new, False)
_async_dispatch_update(
self.hass, group_id, service_info, ibeacon_advertisement, new, False
)
@callback
def _async_update_ibeacon_with_unique_address(
self,
group_id: str,
service_info: bluetooth.BluetoothServiceInfoBleak,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
# Handle iBeacon with a fixed mac address
# and or detect if the iBeacon is using a rotating mac address
# and switch to random mac tracking method
address = service_info.address
unique_id = f"{group_id}_{address}"
new = unique_id not in self._last_rssi_by_unique_id
new = unique_id not in self._last_ibeacon_advertisement_by_unique_id
# Reject creating new trackers if the name is not set
if new and (
service_info.device.name is None
or service_info.device.name.replace("-", ":") == service_info.device.address
):
return
self._last_rssi_by_unique_id[unique_id] = service_info.rssi
self._last_ibeacon_advertisement_by_unique_id[unique_id] = ibeacon_advertisement
self._async_track_ibeacon_with_unique_address(address, group_id, unique_id)
if address not in self._unavailable_trackers:
self._unavailable_trackers[address] = bluetooth.async_track_unavailable(
@ -259,10 +269,14 @@ class IBeaconCoordinator:
# group_id we remove all the trackers for that group_id
# as it means the addresses are being rotated.
if len(self._addresses_by_group_id[group_id]) >= MAX_IDS:
self._async_convert_random_mac_tracking(group_id, service_info, parsed)
self._async_convert_random_mac_tracking(
group_id, service_info, ibeacon_advertisement
)
return
_async_dispatch_update(self.hass, unique_id, service_info, parsed, new, True)
_async_dispatch_update(
self.hass, unique_id, service_info, ibeacon_advertisement, new, True
)
@callback
def _async_stop(self) -> None:
@ -294,21 +308,21 @@ class IBeaconCoordinator:
here and send them over the dispatcher periodically to
ensure the distance calculation is update.
"""
for unique_id, rssi in self._last_rssi_by_unique_id.items():
for (
unique_id,
ibeacon_advertisement,
) in self._last_ibeacon_advertisement_by_unique_id.items():
address = unique_id.split("_")[-1]
if (
(
service_info := bluetooth.async_last_service_info(
self.hass, address, connectable=False
)
service_info := bluetooth.async_last_service_info(
self.hass, address, connectable=False
)
and service_info.rssi != rssi
and (parsed := parse(service_info))
):
) and service_info.rssi != ibeacon_advertisement.rssi:
ibeacon_advertisement.update_rssi(service_info.rssi)
async_dispatcher_send(
self.hass,
signal_seen(unique_id),
parsed,
ibeacon_advertisement,
)
@callback

14
homeassistant/components/ibeacon/device_tracker.py
View File

@ -26,7 +26,7 @@ async def async_setup_entry(
def _async_device_new(
unique_id: str,
identifier: str,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Signal a new device."""
async_add_entities(
@ -35,7 +35,7 @@ async def async_setup_entry(
coordinator,
identifier,
unique_id,
parsed,
ibeacon_advertisement,
)
]
)
@ -53,10 +53,12 @@ class IBeaconTrackerEntity(IBeaconEntity, BaseTrackerEntity):
coordinator: IBeaconCoordinator,
identifier: str,
device_unique_id: str,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Initialize an iBeacon tracker entity."""
super().__init__(coordinator, identifier, device_unique_id, parsed)
super().__init__(
coordinator, identifier, device_unique_id, ibeacon_advertisement
)
self._attr_unique_id = device_unique_id
self._active = True
@ -78,11 +80,11 @@ class IBeaconTrackerEntity(IBeaconEntity, BaseTrackerEntity):
@callback
def _async_seen(
self,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Update state."""
self._active = True
self._parsed = parsed
self._ibeacon_advertisement = ibeacon_advertisement
self.async_write_ha_state()
@callback

16
homeassistant/components/ibeacon/entity.py
View File

@ -24,12 +24,12 @@ class IBeaconEntity(Entity):
coordinator: IBeaconCoordinator,
identifier: str,
device_unique_id: str,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Initialize an iBeacon sensor entity."""
self._device_unique_id = device_unique_id
self._coordinator = coordinator
self._parsed = parsed
self._ibeacon_advertisement = ibeacon_advertisement
self._attr_device_info = DeviceInfo(
name=identifier,
identifiers={(DOMAIN, device_unique_id)},
@ -40,19 +40,19 @@ class IBeaconEntity(Entity):
self,
) -> dict[str, str | int]:
"""Return the device state attributes."""
parsed = self._parsed
ibeacon_advertisement = self._ibeacon_advertisement
return {
ATTR_UUID: str(parsed.uuid),
ATTR_MAJOR: parsed.major,
ATTR_MINOR: parsed.minor,
ATTR_SOURCE: parsed.source,
ATTR_UUID: str(ibeacon_advertisement.uuid),
ATTR_MAJOR: ibeacon_advertisement.major,
ATTR_MINOR: ibeacon_advertisement.minor,
ATTR_SOURCE: ibeacon_advertisement.source,
}
@abstractmethod
@callback
def _async_seen(
self,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Update state."""

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

@ -4,7 +4,7 @@
"documentation": "https://www.home-assistant.io/integrations/ibeacon",
"dependencies": ["bluetooth"],
"bluetooth": [{ "manufacturer_id": 76, "manufacturer_data_start": [2, 21] }],
"requirements": ["ibeacon_ble==0.6.4"],
"requirements": ["ibeacon_ble==0.7.0"],
"codeowners": ["@bdraco"],
"iot_class": "local_push",
"loggers": ["bleak"],

22
homeassistant/components/ibeacon/sensor.py
View File

@ -42,7 +42,7 @@ SENSOR_DESCRIPTIONS = (
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
entity_registry_enabled_default=False,
value_fn=lambda parsed: parsed.rssi,
value_fn=lambda ibeacon_advertisement: ibeacon_advertisement.rssi,
state_class=SensorStateClass.MEASUREMENT,
),
IBeaconSensorEntityDescription(
@ -51,7 +51,7 @@ SENSOR_DESCRIPTIONS = (
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
entity_registry_enabled_default=False,
value_fn=lambda parsed: parsed.power,
value_fn=lambda ibeacon_advertisement: ibeacon_advertisement.power,
state_class=SensorStateClass.MEASUREMENT,
),
IBeaconSensorEntityDescription(
@ -59,7 +59,7 @@ SENSOR_DESCRIPTIONS = (
name="Estimated Distance",
icon="mdi:signal-distance-variant",
native_unit_of_measurement=LENGTH_METERS,
value_fn=lambda parsed: parsed.distance,
value_fn=lambda ibeacon_advertisement: ibeacon_advertisement.distance,
state_class=SensorStateClass.MEASUREMENT,
),
)
@ -75,7 +75,7 @@ async def async_setup_entry(
def _async_device_new(
unique_id: str,
identifier: str,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Signal a new device."""
async_add_entities(
@ -84,7 +84,7 @@ async def async_setup_entry(
description,
identifier,
unique_id,
parsed,
ibeacon_advertisement,
)
for description in SENSOR_DESCRIPTIONS
)
@ -105,21 +105,23 @@ class IBeaconSensorEntity(IBeaconEntity, SensorEntity):
description: IBeaconSensorEntityDescription,
identifier: str,
device_unique_id: str,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Initialize an iBeacon sensor entity."""
super().__init__(coordinator, identifier, device_unique_id, parsed)
super().__init__(
coordinator, identifier, device_unique_id, ibeacon_advertisement
)
self._attr_unique_id = f"{device_unique_id}_{description.key}"
self.entity_description = description
@callback
def _async_seen(
self,
parsed: iBeaconAdvertisement,
ibeacon_advertisement: iBeaconAdvertisement,
) -> None:
"""Update state."""
self._attr_available = True
self._parsed = parsed
self._ibeacon_advertisement = ibeacon_advertisement
self.async_write_ha_state()
@callback
@ -131,4 +133,4 @@ class IBeaconSensorEntity(IBeaconEntity, SensorEntity):
@property
def native_value(self) -> int | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self._parsed)
return self.entity_description.value_fn(self._ibeacon_advertisement)

2
requirements_all.txt
View File

@ -898,7 +898,7 @@ iammeter==0.1.7
iaqualink==0.4.1
# homeassistant.components.ibeacon
ibeacon_ble==0.6.4
ibeacon_ble==0.7.0
# homeassistant.components.watson_tts
ibm-watson==5.2.2

2
requirements_test_all.txt
View File

@ -663,7 +663,7 @@ hyperion-py==0.7.5
iaqualink==0.4.1
# homeassistant.components.ibeacon
ibeacon_ble==0.6.4
ibeacon_ble==0.7.0
# homeassistant.components.ping
icmplib==3.0

43
tests/components/ibeacon/__init__.py
View File

@ -58,3 +58,46 @@ BEACON_RANDOM_ADDRESS_SERVICE_INFO = BluetoothServiceInfo(
service_uuids=[],
source="local",
)
FEASY_BEACON_BLE_DEVICE = BLEDevice(
address="AA:BB:CC:DD:EE:FF",
name="FSC-BP108",
)
FEASY_BEACON_SERVICE_INFO_1 = BluetoothServiceInfo(
name="FSC-BP108",
address="AA:BB:CC:DD:EE:FF",
rssi=-63,
manufacturer_data={
76: b"\x02\x15\xfd\xa5\x06\x93\xa4\xe2O\xb1\xaf\xcf\xc6\xeb\x07dx%'Qe\xc1\xfd"
},
service_data={
"0000feaa-0000-1000-8000-00805f9b34fb": b' \x00\x0c\x86\x80\x00\x00\x00\x93f\x0b\x7f\x93"',
"0000fff0-0000-1000-8000-00805f9b34fb": b"'\x02\x17\x92\xdc\r0\x0e \xbad",
},
service_uuids=[
"0000feaa-0000-1000-8000-00805f9b34fb",
"0000fef5-0000-1000-8000-00805f9b34fb",
],
source="local",
)
FEASY_BEACON_SERVICE_INFO_2 = BluetoothServiceInfo(
name="FSC-BP108",
address="AA:BB:CC:DD:EE:FF",
rssi=-63,
manufacturer_data={
76: b"\x02\x15\xd5F\xdf\x97GWG\xef\xbe\t>-\xcb\xdd\x0cw\xed\xd1;\xd2\xb5"
},
service_data={
"0000feaa-0000-1000-8000-00805f9b34fb": b' \x00\x0c\x86\x80\x00\x00\x00\x93f\x0b\x7f\x93"',
"0000fff0-0000-1000-8000-00805f9b34fb": b"'\x02\x17\x92\xdc\r0\x0e \xbad",
},
service_uuids=[
"0000feaa-0000-1000-8000-00805f9b34fb",
"0000fef5-0000-1000-8000-00805f9b34fb",
],
source="local",
)

62
tests/components/ibeacon/test_sensor.py
View File

@ -20,6 +20,9 @@ from . import (
BLUECHARM_BEACON_SERVICE_INFO,
BLUECHARM_BEACON_SERVICE_INFO_2,
BLUECHARM_BLE_DEVICE,
FEASY_BEACON_BLE_DEVICE,
FEASY_BEACON_SERVICE_INFO_1,
FEASY_BEACON_SERVICE_INFO_2,
NO_NAME_BEACON_SERVICE_INFO,
)
@ -182,3 +185,62 @@ async def test_can_unload_and_reload(hass):
assert (
hass.states.get("sensor.bluecharm_177999_8105_estimated_distance").state == "2"
)
async def test_multiple_uuids_same_beacon(hass):
"""Test a beacon that broadcasts multiple uuids."""
entry = MockConfigEntry(
domain=DOMAIN,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with patch_all_discovered_devices([FEASY_BEACON_BLE_DEVICE]):
inject_bluetooth_service_info(hass, FEASY_BEACON_SERVICE_INFO_1)
await hass.async_block_till_done()
distance_sensor = hass.states.get("sensor.fsc_bp108_eeff_estimated_distance")
distance_attributes = distance_sensor.attributes
assert distance_sensor.state == "400"
assert (
distance_attributes[ATTR_FRIENDLY_NAME] == "FSC-BP108 EEFF Estimated Distance"
)
assert distance_attributes[ATTR_UNIT_OF_MEASUREMENT] == "m"
assert distance_attributes[ATTR_STATE_CLASS] == "measurement"
with patch_all_discovered_devices([FEASY_BEACON_BLE_DEVICE]):
inject_bluetooth_service_info(hass, FEASY_BEACON_SERVICE_INFO_2)
await hass.async_block_till_done()
distance_sensor = hass.states.get("sensor.fsc_bp108_eeff_estimated_distance_2")
distance_attributes = distance_sensor.attributes
assert distance_sensor.state == "0"
assert (
distance_attributes[ATTR_FRIENDLY_NAME] == "FSC-BP108 EEFF Estimated Distance"
)
assert distance_attributes[ATTR_UNIT_OF_MEASUREMENT] == "m"
assert distance_attributes[ATTR_STATE_CLASS] == "measurement"
with patch_all_discovered_devices([FEASY_BEACON_BLE_DEVICE]):
inject_bluetooth_service_info(hass, FEASY_BEACON_SERVICE_INFO_1)
await hass.async_block_till_done()
distance_sensor = hass.states.get("sensor.fsc_bp108_eeff_estimated_distance")
distance_attributes = distance_sensor.attributes
assert distance_sensor.state == "400"
assert (
distance_attributes[ATTR_FRIENDLY_NAME] == "FSC-BP108 EEFF Estimated Distance"
)
assert distance_attributes[ATTR_UNIT_OF_MEASUREMENT] == "m"
assert distance_attributes[ATTR_STATE_CLASS] == "measurement"
distance_sensor = hass.states.get("sensor.fsc_bp108_eeff_estimated_distance_2")
distance_attributes = distance_sensor.attributes
assert distance_sensor.state == "0"
assert (
distance_attributes[ATTR_FRIENDLY_NAME] == "FSC-BP108 EEFF Estimated Distance"
)
assert distance_attributes[ATTR_UNIT_OF_MEASUREMENT] == "m"
assert distance_attributes[ATTR_STATE_CLASS] == "measurement"