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

Add RSSI sensor to HomeKit Controller (#78906)

Browse source
This commit is contained in:
J. Nick Koston 2022-09-25 15:31:56 -10:00 • committed by GitHub
parent c1bc26b413
commit 92612c9fe3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 133 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
homeassistant/components/homekit_controller/connection.py
View file

@ -231,6 +231,9 @@ class HKDevice:
self.async_update_available_state,
timedelta(seconds=BLE_AVAILABILITY_CHECK_INTERVAL),
)
# BLE devices always get an RSSI sensor as well
if "sensor" not in self.platforms:
await self.async_load_platform("sensor")
async def async_add_new_entities(self) -> None:
"""Add new entities to Home Assistant."""
@ -455,7 +458,7 @@ class HKDevice:
self.entities.append((accessory.aid, None, None))
break
def add_char_factory(self, add_entities_cb) -> None:
def add_char_factory(self, add_entities_cb: AddCharacteristicCb) -> None:
"""Add a callback to run when discovering new entities for accessories."""
self.char_factories.append(add_entities_cb)
self._add_new_entities_for_char([add_entities_cb])
@ -471,7 +474,7 @@ class HKDevice:
self.entities.append((accessory.aid, service.iid, char.iid))
break
def add_listener(self, add_entities_cb) -> None:
def add_listener(self, add_entities_cb: AddServiceCb) -> None:
"""Add a callback to run when discovering new entities for services."""
self.listeners.append(add_entities_cb)
self._add_new_entities([add_entities_cb])
@ -513,22 +516,24 @@ class HKDevice:
async def async_load_platforms(self) -> None:
"""Load any platforms needed by this HomeKit device."""
tasks = []
to_load: set[str] = set()
for accessory in self.entity_map.accessories:
for service in accessory.services:
if service.type in HOMEKIT_ACCESSORY_DISPATCH:
platform = HOMEKIT_ACCESSORY_DISPATCH[service.type]
if platform not in self.platforms:
tasks.append(self.async_load_platform(platform))
to_load.add(platform)
for char in service.characteristics:
if char.type in CHARACTERISTIC_PLATFORMS:
platform = CHARACTERISTIC_PLATFORMS[char.type]
if platform not in self.platforms:
tasks.append(self.async_load_platform(platform))
to_load.add(platform)
if tasks:
await asyncio.gather(*tasks)
if to_load:
await asyncio.gather(
*[self.async_load_platform(platform) for platform in to_load]
)
@callback
def async_update_available_state(self, *_: Any) -> None:

65
homeassistant/components/homekit_controller/sensor.py
View file

@ -3,11 +3,14 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import logging
from aiohomekit.model import Accessory, Transport
from aiohomekit.model.characteristics import Characteristic, CharacteristicsTypes
from aiohomekit.model.characteristics.const import ThreadNodeCapabilities, ThreadStatus
from aiohomekit.model.services import Service, ServicesTypes
from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
@ -25,6 +28,7 @@ from homeassistant.const import (
PERCENTAGE,
POWER_WATT,
PRESSURE_HPA,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant, callback
@ -37,6 +41,8 @@ from .connection import HKDevice
from .entity import CharacteristicEntity, HomeKitEntity
from .utils import folded_name
_LOGGER = logging.getLogger(__name__)
@dataclass
class HomeKitSensorEntityDescription(SensorEntityDescription):
@ -524,6 +530,45 @@ REQUIRED_CHAR_BY_TYPE = {
}
class RSSISensor(HomeKitEntity, SensorEntity):
"""HomeKit Controller RSSI sensor."""
_attr_device_class = SensorDeviceClass.SIGNAL_STRENGTH
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_entity_registry_enabled_default = False
_attr_has_entity_name = True
_attr_native_unit_of_measurement = SIGNAL_STRENGTH_DECIBELS_MILLIWATT
_attr_should_poll = False
def get_characteristic_types(self) -> list[str]:
"""Define the homekit characteristics the entity cares about."""
return []
@property
def available(self) -> bool:
"""Return if the bluetooth device is available."""
address = self._accessory.pairing_data["AccessoryAddress"]
return async_ble_device_from_address(self.hass, address) is not None
@property
def name(self) -> str:
"""Return the name of the sensor."""
return "Signal strength"
@property
def unique_id(self) -> str:
"""Return the ID of this device."""
serial = self.accessory_info.value(CharacteristicsTypes.SERIAL_NUMBER)
return f"homekit-{serial}-rssi"
@property
def native_value(self) -> int | None:
"""Return the current rssi value."""
address = self._accessory.pairing_data["AccessoryAddress"]
ble_device = async_ble_device_from_address(self.hass, address)
return ble_device.rssi if ble_device else None
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -531,7 +576,7 @@ async def async_setup_entry(
) -> None:
"""Set up Homekit sensors."""
hkid = config_entry.data["AccessoryPairingID"]
conn = hass.data[KNOWN_DEVICES][hkid]
conn: HKDevice = hass.data[KNOWN_DEVICES][hkid]
@callback
def async_add_service(service: Service) -> bool:
@ -542,7 +587,7 @@ async def async_setup_entry(
) and not service.has(required_char):
return False
info = {"aid": service.accessory.aid, "iid": service.iid}
async_add_entities([entity_class(conn, info)], True)
async_add_entities([entity_class(conn, info)])
return True
conn.add_listener(async_add_service)
@ -554,8 +599,22 @@ async def async_setup_entry(
if description.probe and not description.probe(char):
return False
info = {"aid": char.service.accessory.aid, "iid": char.service.iid}
async_add_entities([SimpleSensor(conn, info, char, description)], True)
async_add_entities([SimpleSensor(conn, info, char, description)])
return True
conn.add_char_factory(async_add_characteristic)
@callback
def async_add_accessory(accessory: Accessory) -> bool:
if conn.pairing.transport != Transport.BLE:
return False
accessory_info = accessory.services.first(
service_type=ServicesTypes.ACCESSORY_INFORMATION
)
info = {"aid": accessory.aid, "iid": accessory_info.iid}
async_add_entities([RSSISensor(conn, info)])
return True
conn.add_accessory_factory(async_add_accessory)

35
tests/components/homekit_controller/common.py
View file

@ -26,6 +26,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, State, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
@ -42,6 +43,19 @@ logger = logging.getLogger(__name__)
# Root device in test harness always has an accessory id of this
HUB_TEST_ACCESSORY_ID: Final[str] = "00:00:00:00:00:00:aid:1"
TEST_ACCESSORY_ADDRESS = "AA:BB:CC:DD:EE:FF"
TEST_DEVICE_SERVICE_INFO = BluetoothServiceInfo(
name="test_accessory",
address=TEST_ACCESSORY_ADDRESS,
rssi=-56,
manufacturer_data={},
service_uuids=["0000ec88-0000-1000-8000-00805f9b34fb"],
service_data={},
source="local",
)
@dataclass
class EntityTestInfo:
@ -182,15 +196,17 @@ async def setup_platform(hass):
return await async_get_controller(hass)
async def setup_test_accessories(hass, accessories):
async def setup_test_accessories(hass, accessories, connection=None):
"""Load a fake homekit device based on captured JSON profile."""
fake_controller = await setup_platform(hass)
return await setup_test_accessories_with_controller(
hass, accessories, fake_controller
hass, accessories, fake_controller, connection
)
async def setup_test_accessories_with_controller(hass, accessories, fake_controller):
async def setup_test_accessories_with_controller(
hass, accessories, fake_controller, connection=None
):
"""Load a fake homekit device based on captured JSON profile."""
pairing_id = "00:00:00:00:00:00"
@ -200,11 +216,16 @@ async def setup_test_accessories_with_controller(hass, accessories, fake_control
accessories_obj.add_accessory(accessory)
pairing = await fake_controller.add_paired_device(accessories_obj, pairing_id)
data = {"AccessoryPairingID": pairing_id}
if connection == "BLE":
data["Connection"] = "BLE"
data["AccessoryAddress"] = TEST_ACCESSORY_ADDRESS
config_entry = MockConfigEntry(
version=1,
domain="homekit_controller",
entry_id="TestData",
data={"AccessoryPairingID": pairing_id},
data=data,
title="test",
)
config_entry.add_to_hass(hass)
@ -250,7 +271,9 @@ async def device_config_changed(hass, accessories):
await hass.async_block_till_done()
async def setup_test_component(hass, setup_accessory, capitalize=False, suffix=None):
async def setup_test_component(
hass, setup_accessory, capitalize=False, suffix=None, connection=None
):
"""Load a fake homekit accessory based on a homekit accessory model.
If capitalize is True, property names will be in upper case.
@ -271,7 +294,7 @@ async def setup_test_component(hass, setup_accessory, capitalize=False, suffix=N
assert domain, "Cannot map test homekit services to Home Assistant domain"
config_entry, pairing = await setup_test_accessories(hass, [accessory])
config_entry, pairing = await setup_test_accessories(hass, [accessory], connection)
entity = "testdevice" if suffix is None else f"testdevice_{suffix}"
return Helper(hass, ".".join((domain, entity)), pairing, accessory, config_entry)

31
tests/components/homekit_controller/test_sensor.py
View file

@ -1,8 +1,12 @@
"""Basic checks for HomeKit sensor."""
from unittest.mock import patch
from aiohomekit.model import Transport
from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.characteristics.const import ThreadNodeCapabilities, ThreadStatus
from aiohomekit.model.services import ServicesTypes
from aiohomekit.protocol.statuscodes import HapStatusCode
from aiohomekit.testing import FakePairing
from homeassistant.components.homekit_controller.sensor import (
thread_node_capability_to_str,
@ -10,7 +14,9 @@ from homeassistant.components.homekit_controller.sensor import (
)
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
from .common import Helper, setup_test_component
from .common import TEST_DEVICE_SERVICE_INFO, Helper, setup_test_component
from tests.components.bluetooth import inject_bluetooth_service_info
def create_temperature_sensor_service(accessory):
@ -349,3 +355,26 @@ def test_thread_status_to_str():
assert thread_status_to_str(ThreadStatus.JOINING) == "joining"
assert thread_status_to_str(ThreadStatus.DETACHED) == "detached"
assert thread_status_to_str(ThreadStatus.DISABLED) == "disabled"
async def test_rssi_sensor(
hass, utcnow, entity_registry_enabled_by_default, enable_bluetooth
):
"""Test an rssi sensor."""
inject_bluetooth_service_info(hass, TEST_DEVICE_SERVICE_INFO)
class FakeBLEPairing(FakePairing):
"""Fake BLE pairing."""
@property
def transport(self):
return Transport.BLE
with patch("aiohomekit.testing.FakePairing", FakeBLEPairing):
# Any accessory will do for this test, but we need at least
# one or the rssi sensor will not be created
await setup_test_component(
hass, create_battery_level_sensor, suffix="battery", connection="BLE"
)
assert hass.states.get("sensor.testdevice_signal_strength").state == "-56"