De-duplicate generation of DeviceInfo data in homekit_controller (#64751)

This commit is contained in:
Jc2k 2022-01-23 21:57:16 +00:00 committed by GitHub
parent fbe2b81cd4
commit f71e053947
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 52 additions and 85 deletions

View File

@ -15,7 +15,7 @@ from aiohomekit.model.services import Service, ServicesTypes
from homeassistant.components import zeroconf from homeassistant.components import zeroconf
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_VIA_DEVICE, EVENT_HOMEASSISTANT_STOP from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.entity import DeviceInfo, Entity from homeassistant.helpers.entity import DeviceInfo, Entity
@ -23,15 +23,7 @@ from homeassistant.helpers.typing import ConfigType
from .config_flow import normalize_hkid from .config_flow import normalize_hkid
from .connection import HKDevice, valid_serial_number from .connection import HKDevice, valid_serial_number
from .const import ( from .const import CONTROLLER, ENTITY_MAP, KNOWN_DEVICES, TRIGGERS
CONTROLLER,
DOMAIN,
ENTITY_MAP,
IDENTIFIER_ACCESSORY_ID,
IDENTIFIER_SERIAL_NUMBER,
KNOWN_DEVICES,
TRIGGERS,
)
from .storage import EntityMapStorage from .storage import EntityMapStorage
@ -45,7 +37,7 @@ class HomeKitEntity(Entity):
_attr_should_poll = False _attr_should_poll = False
def __init__(self, accessory, devinfo): def __init__(self, accessory: HKDevice, devinfo):
"""Initialise a generic HomeKit device.""" """Initialise a generic HomeKit device."""
self._accessory = accessory self._accessory = accessory
self._aid = devinfo["aid"] self._aid = devinfo["aid"]
@ -162,38 +154,7 @@ class HomeKitEntity(Entity):
@property @property
def device_info(self) -> DeviceInfo: def device_info(self) -> DeviceInfo:
"""Return the device info.""" """Return the device info."""
info = self.accessory_info return self._accessory.device_info_for_accessory(self.accessory)
accessory_serial = info.value(CharacteristicsTypes.SERIAL_NUMBER)
if valid_serial_number(accessory_serial):
# Some accessories do not have a serial number
identifier = (DOMAIN, IDENTIFIER_SERIAL_NUMBER, accessory_serial)
else:
identifier = (
DOMAIN,
IDENTIFIER_ACCESSORY_ID,
f"{self._accessory.unique_id}_{self._aid}",
)
device_info = DeviceInfo(
identifiers={identifier},
manufacturer=info.value(CharacteristicsTypes.MANUFACTURER, ""),
model=info.value(CharacteristicsTypes.MODEL, ""),
name=info.value(CharacteristicsTypes.NAME),
sw_version=info.value(CharacteristicsTypes.FIRMWARE_REVISION, ""),
hw_version=info.value(CharacteristicsTypes.HARDWARE_REVISION, ""),
)
# Some devices only have a single accessory - we don't add a
# via_device otherwise it would be self referential.
bridge_serial = self._accessory.connection_info["serial-number"]
if accessory_serial != bridge_serial:
device_info[ATTR_VIA_DEVICE] = (
DOMAIN,
IDENTIFIER_SERIAL_NUMBER,
bridge_serial,
)
return device_info
def get_characteristic_types(self): def get_characteristic_types(self):
"""Define the homekit characteristics the entity cares about.""" """Define the homekit characteristics the entity cares about."""

View File

@ -8,7 +8,7 @@ from aiohomekit.exceptions import (
AccessoryNotFoundError, AccessoryNotFoundError,
EncryptionError, EncryptionError,
) )
from aiohomekit.model import Accessories from aiohomekit.model import Accessories, Accessory
from aiohomekit.model.characteristics import CharacteristicsTypes from aiohomekit.model.characteristics import CharacteristicsTypes
from aiohomekit.model.services import ServicesTypes from aiohomekit.model.services import ServicesTypes
@ -201,6 +201,52 @@ class HKDevice:
return True return True
def device_info_for_accessory(self, accessory: Accessory) -> DeviceInfo:
"""Build a DeviceInfo for a given accessory."""
info = accessory.services.first(
service_type=ServicesTypes.ACCESSORY_INFORMATION,
)
serial_number = info.value(CharacteristicsTypes.SERIAL_NUMBER)
if valid_serial_number(serial_number):
identifiers = {(DOMAIN, IDENTIFIER_SERIAL_NUMBER, serial_number)}
else:
# Some accessories do not have a serial number
identifiers = {
(
DOMAIN,
IDENTIFIER_ACCESSORY_ID,
f"{self.unique_id}_{accessory.aid}",
)
}
if accessory.aid == 1:
# Accessory 1 is the root device (sometimes the only device, sometimes a bridge)
# Link the root device to the pairing id for the connection.
identifiers.add((DOMAIN, IDENTIFIER_ACCESSORY_ID, self.unique_id))
device_info = DeviceInfo(
identifiers=identifiers,
name=info.value(CharacteristicsTypes.NAME),
manufacturer=info.value(CharacteristicsTypes.MANUFACTURER, ""),
model=info.value(CharacteristicsTypes.MODEL, ""),
sw_version=info.value(CharacteristicsTypes.FIRMWARE_REVISION, ""),
hw_version=info.value(CharacteristicsTypes.HARDWARE_REVISION, ""),
)
if accessory.aid != 1:
# Every pairing has an accessory 1
# It *doesn't* have a via_device, as it is the device we are connecting to
# Every other accessory should use it as its via device.
device_info[ATTR_VIA_DEVICE] = (
DOMAIN,
IDENTIFIER_SERIAL_NUMBER,
self.connection_info["serial-number"],
)
return device_info
@callback @callback
def async_create_devices(self): def async_create_devices(self):
""" """
@ -219,48 +265,8 @@ class HKDevice:
for accessory in sorted( for accessory in sorted(
self.entity_map.accessories, key=lambda accessory: accessory.aid self.entity_map.accessories, key=lambda accessory: accessory.aid
): ):
info = accessory.services.first(
service_type=ServicesTypes.ACCESSORY_INFORMATION,
)
serial_number = info.value(CharacteristicsTypes.SERIAL_NUMBER)
if valid_serial_number(serial_number):
identifiers = {(DOMAIN, IDENTIFIER_SERIAL_NUMBER, serial_number)}
else:
# Some accessories do not have a serial number
identifiers = {
(
DOMAIN,
IDENTIFIER_ACCESSORY_ID,
f"{self.unique_id}_{accessory.aid}",
)
}
if accessory.aid == 1:
# Accessory 1 is the root device (sometimes the only device, sometimes a bridge)
# Link the root device to the pairing id for the connection.
identifiers.add((DOMAIN, IDENTIFIER_ACCESSORY_ID, self.unique_id))
device_info = DeviceInfo(
identifiers=identifiers,
name=info.value(CharacteristicsTypes.NAME),
manufacturer=info.value(CharacteristicsTypes.MANUFACTURER, ""),
model=info.value(CharacteristicsTypes.MODEL, ""),
sw_version=info.value(CharacteristicsTypes.FIRMWARE_REVISION, ""),
hw_version=info.value(CharacteristicsTypes.HARDWARE_REVISION, ""),
)
if accessory.aid != 1:
# Every pairing has an accessory 1
# It *doesn't* have a via_device, as it is the device we are connecting to
# Every other accessory should use it as its via device.
device_info[ATTR_VIA_DEVICE] = (
DOMAIN,
IDENTIFIER_SERIAL_NUMBER,
self.connection_info["serial-number"],
)
device_info = self.device_info_for_accessory(accessory)
device = device_registry.async_get_or_create( device = device_registry.async_get_or_create(
config_entry_id=self.config_entry.entry_id, config_entry_id=self.config_entry.entry_id,
**device_info, **device_info,