custom_components/homekey_reader/sensor.py
raw ยท 4317 bytes
"""Sensor platform for HomeKey Reader."""
from __future__ import annotations
import logging
from homeassistant.components import bluetooth
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN, EVENT_PACKET, SERVICE_UUID
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the HomeKey Reader sensor."""
async_add_entities([HomeKeyReaderUidSensor(hass, entry)])
class HomeKeyReaderUidSensor(SensorEntity):
"""Last NFC UID received from one HomeKey Reader."""
_attr_has_entity_name = True
_attr_name = "Last NFC UID"
_attr_icon = "mdi:nfc"
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
self.hass = hass
self._entry = entry
self._address = entry.data[CONF_ADDRESS]
self._attr_unique_id = f"{self._address}_last_uid"
self._attr_native_value = None
self._attr_available = False
self._attrs: dict[str, object] = {}
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._address)},
name=entry.title,
manufacturer="HomeKey Reader",
model="ESP32-C3 + PN532",
)
@property
def extra_state_attributes(self) -> dict[str, object]:
return self._attrs
async def async_added_to_hass(self) -> None:
"""Subscribe to Bluetooth advertisements."""
@callback
def _packet_seen(
service_info: bluetooth.BluetoothServiceInfoBleak,
change: bluetooth.BluetoothChange,
) -> None:
payload = service_info.service_data.get(SERVICE_UUID)
if payload is None:
return
if len(payload) < 8:
_LOGGER.warning(
"Short HomeKey Reader packet from %s: %s",
service_info.address,
payload.hex(),
)
return
version = payload[0]
msg_type = payload[1]
seq = int.from_bytes(payload[2:4], "little")
data = payload[4:]
if version != 1 or msg_type != 1 or len(data) < 4:
_LOGGER.warning(
"Unsupported HomeKey Reader packet from %s: %s",
service_info.address,
payload.hex(),
)
return
uid = ":".join(f"{byte:02X}" for byte in data[:4])
self._attr_native_value = uid
self._attr_available = True
self._attrs = {
"address": service_info.address,
"rssi": service_info.rssi,
"sequence": seq,
"protocol_version": version,
"message_type": msg_type,
"raw": payload.hex(),
}
self.async_write_ha_state()
self.hass.bus.async_fire(
EVENT_PACKET,
{
"uid": uid,
**self._attrs,
},
)
@callback
def _unavailable(_service_info: bluetooth.BluetoothServiceInfoBleak) -> None:
self._attr_available = False
self.async_write_ha_state()
self.async_on_remove(
bluetooth.async_register_callback(
self.hass,
_packet_seen,
{
"address": self._address,
"service_data_uuid": SERVICE_UUID,
"connectable": False,
},
bluetooth.BluetoothScanningMode.ACTIVE,
replay=bluetooth.BluetoothCallbackReplay.NEWEST_FIRST,
)
)
self.async_on_remove(
bluetooth.async_track_unavailable(
self.hass,
_unavailable,
self._address,
connectable=False,
)
)