Improve bluetooth generic typing (#117157)

This commit is contained in:
Marc Mueller 2024-05-12 13:15:30 +02:00 committed by GitHub
parent 92254772ca
commit 4f4389ba85
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 126 additions and 69 deletions

View file

@ -143,7 +143,7 @@ def _sensor_device_info_to_hass(
def sensor_update_to_bluetooth_data_update(
adv: Aranet4Advertisement,
) -> PassiveBluetoothDataUpdate:
) -> PassiveBluetoothDataUpdate[Any]:
"""Convert a sensor update to a Bluetooth data update."""
data: dict[PassiveBluetoothEntityKey, Any] = {}
names: dict[PassiveBluetoothEntityKey, str | None] = {}
@ -171,9 +171,9 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Aranet sensors."""
coordinator: PassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
coordinator: PassiveBluetoothProcessorCoordinator[Aranet4Advertisement] = hass.data[
DOMAIN
][entry.entry_id]
processor = PassiveBluetoothDataProcessor(sensor_update_to_bluetooth_data_update)
entry.async_on_unload(
processor.async_add_entities_listener(
@ -184,7 +184,9 @@ async def async_setup_entry(
class Aranet4BluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, Aranet4Advertisement],
],
SensorEntity,
):
"""Representation of an Aranet sensor."""

View file

@ -134,7 +134,9 @@ async def async_setup_entry(
class BlueMaestroBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a BlueMaestro sensor."""

View file

@ -7,7 +7,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine
import logging
from typing import Any, Generic, TypeVar
from typing import Any, TypeVar
from bleak import BleakError
from bluetooth_data_tools import monotonic_time_coarse
@ -21,12 +21,10 @@ from .passive_update_processor import PassiveBluetoothProcessorCoordinator
POLL_DEFAULT_COOLDOWN = 10
POLL_DEFAULT_IMMEDIATE = True
_T = TypeVar("_T")
_DataT = TypeVar("_DataT")
class ActiveBluetoothProcessorCoordinator(
Generic[_T], PassiveBluetoothProcessorCoordinator[_T]
):
class ActiveBluetoothProcessorCoordinator(PassiveBluetoothProcessorCoordinator[_DataT]):
"""A processor coordinator that parses passive data.
Parses passive data from advertisements but can also poll.
@ -63,11 +61,11 @@ class ActiveBluetoothProcessorCoordinator(
*,
address: str,
mode: BluetoothScanningMode,
update_method: Callable[[BluetoothServiceInfoBleak], _T],
update_method: Callable[[BluetoothServiceInfoBleak], _DataT],
needs_poll_method: Callable[[BluetoothServiceInfoBleak, float | None], bool],
poll_method: Callable[
[BluetoothServiceInfoBleak],
Coroutine[Any, Any, _T],
Coroutine[Any, Any, _DataT],
]
| None = None,
poll_debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None,
@ -110,7 +108,7 @@ class ActiveBluetoothProcessorCoordinator(
async def _async_poll_data(
self, last_service_info: BluetoothServiceInfoBleak
) -> _T:
) -> _DataT:
"""Fetch the latest data from the source."""
if self._poll_method is None:
raise NotImplementedError("Poll method not implemented")

View file

@ -6,7 +6,7 @@ import dataclasses
from datetime import timedelta
from functools import cache
import logging
from typing import TYPE_CHECKING, Any, Generic, TypedDict, TypeVar, cast
from typing import TYPE_CHECKING, Any, Generic, Self, TypedDict, TypeVar, cast
from habluetooth import BluetoothScanningMode
@ -42,7 +42,9 @@ STORAGE_KEY = "bluetooth.passive_update_processor"
STORAGE_VERSION = 1
STORAGE_SAVE_INTERVAL = timedelta(minutes=15)
PASSIVE_UPDATE_PROCESSOR = "passive_update_processor"
_T = TypeVar("_T")
_DataT = TypeVar("_DataT")
@dataclasses.dataclass(slots=True, frozen=True)
@ -73,7 +75,7 @@ class PassiveBluetoothEntityKey:
class PassiveBluetoothProcessorData:
"""Data for the passive bluetooth processor."""
coordinators: set[PassiveBluetoothProcessorCoordinator]
coordinators: set[PassiveBluetoothProcessorCoordinator[Any]]
all_restore_data: dict[str, dict[str, RestoredPassiveBluetoothDataUpdate]]
@ -220,7 +222,7 @@ class PassiveBluetoothDataUpdate(Generic[_T]):
def async_register_coordinator_for_restore(
hass: HomeAssistant, coordinator: PassiveBluetoothProcessorCoordinator
hass: HomeAssistant, coordinator: PassiveBluetoothProcessorCoordinator[Any]
) -> CALLBACK_TYPE:
"""Register a coordinator to have its processors data restored."""
data: PassiveBluetoothProcessorData = hass.data[PASSIVE_UPDATE_PROCESSOR]
@ -242,7 +244,7 @@ async def async_setup(hass: HomeAssistant) -> None:
storage: Store[dict[str, dict[str, RestoredPassiveBluetoothDataUpdate]]] = Store(
hass, STORAGE_VERSION, STORAGE_KEY
)
coordinators: set[PassiveBluetoothProcessorCoordinator] = set()
coordinators: set[PassiveBluetoothProcessorCoordinator[Any]] = set()
all_restore_data: dict[str, dict[str, RestoredPassiveBluetoothDataUpdate]] = (
await storage.async_load() or {}
)
@ -276,7 +278,7 @@ async def async_setup(hass: HomeAssistant) -> None:
class PassiveBluetoothProcessorCoordinator(
Generic[_T], BasePassiveBluetoothCoordinator
Generic[_DataT], BasePassiveBluetoothCoordinator
):
"""Passive bluetooth processor coordinator for bluetooth advertisements.
@ -294,12 +296,12 @@ class PassiveBluetoothProcessorCoordinator(
logger: logging.Logger,
address: str,
mode: BluetoothScanningMode,
update_method: Callable[[BluetoothServiceInfoBleak], _T],
update_method: Callable[[BluetoothServiceInfoBleak], _DataT],
connectable: bool = False,
) -> None:
"""Initialize the coordinator."""
super().__init__(hass, logger, address, mode, connectable)
self._processors: list[PassiveBluetoothDataProcessor] = []
self._processors: list[PassiveBluetoothDataProcessor[Any, _DataT]] = []
self._update_method = update_method
self.last_update_success = True
self.restore_data: dict[str, RestoredPassiveBluetoothDataUpdate] = {}
@ -327,7 +329,7 @@ class PassiveBluetoothProcessorCoordinator(
@callback
def async_register_processor(
self,
processor: PassiveBluetoothDataProcessor,
processor: PassiveBluetoothDataProcessor[Any, _DataT],
entity_description_class: type[EntityDescription] | None = None,
) -> Callable[[], None]:
"""Register a processor that subscribes to updates."""
@ -388,11 +390,11 @@ class PassiveBluetoothProcessorCoordinator(
_PassiveBluetoothDataProcessorT = TypeVar(
"_PassiveBluetoothDataProcessorT",
bound="PassiveBluetoothDataProcessor[Any]",
bound="PassiveBluetoothDataProcessor[Any, Any]",
)
class PassiveBluetoothDataProcessor(Generic[_T]):
class PassiveBluetoothDataProcessor(Generic[_T, _DataT]):
"""Passive bluetooth data processor for bluetooth advertisements.
The processor is responsible for keeping track of the bluetooth data
@ -413,7 +415,7 @@ class PassiveBluetoothDataProcessor(Generic[_T]):
is available in the devices, entity_data, and entity_descriptions attributes.
"""
coordinator: PassiveBluetoothProcessorCoordinator
coordinator: PassiveBluetoothProcessorCoordinator[_DataT]
data: PassiveBluetoothDataUpdate[_T]
entity_names: dict[PassiveBluetoothEntityKey, str | None]
entity_data: dict[PassiveBluetoothEntityKey, _T]
@ -423,7 +425,7 @@ class PassiveBluetoothDataProcessor(Generic[_T]):
def __init__(
self,
update_method: Callable[[_T], PassiveBluetoothDataUpdate[_T]],
update_method: Callable[[_DataT], PassiveBluetoothDataUpdate[_T]],
restore_key: str | None = None,
) -> None:
"""Initialize the coordinator."""
@ -444,7 +446,7 @@ class PassiveBluetoothDataProcessor(Generic[_T]):
@callback
def async_register_coordinator(
self,
coordinator: PassiveBluetoothProcessorCoordinator,
coordinator: PassiveBluetoothProcessorCoordinator[_DataT],
entity_description_class: type[EntityDescription] | None,
) -> None:
"""Register a coordinator."""
@ -482,7 +484,7 @@ class PassiveBluetoothDataProcessor(Generic[_T]):
@callback
def async_add_entities_listener(
self,
entity_class: type[PassiveBluetoothProcessorEntity],
entity_class: type[PassiveBluetoothProcessorEntity[Self]],
async_add_entities: AddEntitiesCallback,
) -> Callable[[], None]:
"""Add a listener for new entities."""
@ -495,7 +497,7 @@ class PassiveBluetoothDataProcessor(Generic[_T]):
"""Listen for new entities."""
if data is None or created.issuperset(data.entity_descriptions):
return
entities: list[PassiveBluetoothProcessorEntity] = []
entities: list[PassiveBluetoothProcessorEntity[Self]] = []
for entity_key, description in data.entity_descriptions.items():
if entity_key not in created:
entities.append(entity_class(self, entity_key, description))
@ -578,7 +580,7 @@ class PassiveBluetoothDataProcessor(Generic[_T]):
@callback
def async_handle_update(
self, update: _T, was_available: bool | None = None
self, update: _DataT, was_available: bool | None = None
) -> None:
"""Handle a Bluetooth event."""
try:
@ -666,7 +668,8 @@ class PassiveBluetoothProcessorEntity(Entity, Generic[_PassiveBluetoothDataProce
@callback
def _handle_processor_update(
self, new_data: PassiveBluetoothDataUpdate | None
self,
new_data: PassiveBluetoothDataUpdate[_PassiveBluetoothDataProcessorT] | None,
) -> None:
"""Handle updated data from the processor."""
self.async_write_ha_state()

View file

@ -145,7 +145,7 @@ BINARY_SENSOR_DESCRIPTIONS = {
def sensor_update_to_bluetooth_data_update(
sensor_update: SensorUpdate,
) -> PassiveBluetoothDataUpdate:
) -> PassiveBluetoothDataUpdate[bool | None]:
"""Convert a binary sensor update to a bluetooth data update."""
return PassiveBluetoothDataUpdate(
devices={
@ -193,7 +193,7 @@ async def async_setup_entry(
class BTHomeBluetoothBinarySensorEntity(
PassiveBluetoothProcessorEntity[BTHomePassiveBluetoothDataProcessor],
PassiveBluetoothProcessorEntity[BTHomePassiveBluetoothDataProcessor[bool | None]],
BinarySensorEntity,
):
"""Representation of a BTHome binary sensor."""

View file

@ -2,9 +2,9 @@
from collections.abc import Callable
from logging import Logger
from typing import Any
from typing import TypeVar
from bthome_ble import BTHomeBluetoothDeviceData
from bthome_ble import BTHomeBluetoothDeviceData, SensorUpdate
from homeassistant.components.bluetooth import (
BluetoothScanningMode,
@ -19,8 +19,12 @@ from homeassistant.core import HomeAssistant
from .const import CONF_SLEEPY_DEVICE
_T = TypeVar("_T")
class BTHomePassiveBluetoothProcessorCoordinator(PassiveBluetoothProcessorCoordinator):
class BTHomePassiveBluetoothProcessorCoordinator(
PassiveBluetoothProcessorCoordinator[SensorUpdate]
):
"""Define a BTHome Bluetooth Passive Update Processor Coordinator."""
def __init__(
@ -29,7 +33,7 @@ class BTHomePassiveBluetoothProcessorCoordinator(PassiveBluetoothProcessorCoordi
logger: Logger,
address: str,
mode: BluetoothScanningMode,
update_method: Callable[[BluetoothServiceInfoBleak], Any],
update_method: Callable[[BluetoothServiceInfoBleak], SensorUpdate],
device_data: BTHomeBluetoothDeviceData,
discovered_event_classes: set[str],
entry: ConfigEntry,
@ -47,7 +51,9 @@ class BTHomePassiveBluetoothProcessorCoordinator(PassiveBluetoothProcessorCoordi
return self.entry.data.get(CONF_SLEEPY_DEVICE, self.device_data.sleepy_device)
class BTHomePassiveBluetoothDataProcessor(PassiveBluetoothDataProcessor):
class BTHomePassiveBluetoothDataProcessor(
PassiveBluetoothDataProcessor[_T, SensorUpdate]
):
"""Define a BTHome Bluetooth Passive Update Data Processor."""
coordinator: BTHomePassiveBluetoothProcessorCoordinator

View file

@ -2,6 +2,8 @@
from __future__ import annotations
from typing import cast
from bthome_ble import SensorDeviceClass as BTHomeSensorDeviceClass, SensorUpdate, Units
from bthome_ble.const import (
ExtendedSensorDeviceClass as BTHomeExtendedSensorDeviceClass,
@ -363,7 +365,7 @@ SENSOR_DESCRIPTIONS = {
def sensor_update_to_bluetooth_data_update(
sensor_update: SensorUpdate,
) -> PassiveBluetoothDataUpdate:
) -> PassiveBluetoothDataUpdate[float | None]:
"""Convert a sensor update to a bluetooth data update."""
return PassiveBluetoothDataUpdate(
devices={
@ -378,7 +380,9 @@ def sensor_update_to_bluetooth_data_update(
if description.device_class
},
entity_data={
device_key_to_bluetooth_entity_key(device_key): sensor_values.native_value
device_key_to_bluetooth_entity_key(device_key): cast(
float | None, sensor_values.native_value
)
for device_key, sensor_values in sensor_update.entity_values.items()
},
entity_names={
@ -411,7 +415,7 @@ async def async_setup_entry(
class BTHomeBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[BTHomePassiveBluetoothDataProcessor],
PassiveBluetoothProcessorEntity[BTHomePassiveBluetoothDataProcessor[float | None]],
SensorEntity,
):
"""Representation of a BTHome BLE sensor."""

View file

@ -124,7 +124,7 @@ async def async_setup_entry(
class GoveeBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | str | None]
PassiveBluetoothDataProcessor[float | int | str | None, SensorUpdate]
],
SensorEntity,
):

View file

@ -114,7 +114,9 @@ async def async_setup_entry(
class INKBIRDBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a inkbird ble sensor."""

View file

@ -126,7 +126,9 @@ async def async_setup_entry(
class KegtronBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a Kegtron sensor."""

View file

@ -125,7 +125,9 @@ async def async_setup_entry(
class LeaoneBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a Leaone sensor."""

View file

@ -121,7 +121,9 @@ async def async_setup_entry(
class MoatBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a moat ble sensor."""

View file

@ -133,7 +133,9 @@ async def async_setup_entry(
class MopekaBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a Mopeka sensor."""

View file

@ -128,7 +128,9 @@ async def async_setup_entry(
class OralBBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[str | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[str | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a OralB sensor."""

View file

@ -94,7 +94,9 @@ async def async_setup_entry(
class QingpingBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[bool | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[bool | None, SensorUpdate]
],
BinarySensorEntity,
):
"""Representation of a Qingping binary sensor."""

View file

@ -162,7 +162,9 @@ async def async_setup_entry(
class QingpingBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a Qingping sensor."""

View file

@ -115,7 +115,9 @@ async def async_setup_entry(
class RAPTPillBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a RAPT Pill BLE sensor."""

View file

@ -142,7 +142,9 @@ async def async_setup_entry(
class RuuvitagBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a Ruuvitag BLE sensor."""

View file

@ -122,7 +122,9 @@ async def async_setup_entry(
class SensirionBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a Sensirion BLE sensor."""

View file

@ -127,7 +127,9 @@ async def async_setup_entry(
class SensorProBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a SensorPro sensor."""

View file

@ -117,7 +117,9 @@ async def async_setup_entry(
class SensorPushBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a sensorpush ble sensor."""

View file

@ -129,7 +129,9 @@ async def async_setup_entry(
class ThermoBeaconBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a ThermoBeacon sensor."""

View file

@ -127,7 +127,9 @@ async def async_setup_entry(
class ThermoProBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a thermopro ble sensor."""

View file

@ -102,7 +102,9 @@ async def async_setup_entry(
class TiltBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[float | int | None, SensorUpdate]
],
SensorEntity,
):
"""Representation of a Tilt Hydrometer BLE sensor."""

View file

@ -107,7 +107,7 @@ BINARY_SENSOR_DESCRIPTIONS = {
def sensor_update_to_bluetooth_data_update(
sensor_update: SensorUpdate,
) -> PassiveBluetoothDataUpdate:
) -> PassiveBluetoothDataUpdate[bool | None]:
"""Convert a sensor update to a bluetooth data update."""
return PassiveBluetoothDataUpdate(
devices={
@ -155,7 +155,7 @@ async def async_setup_entry(
class XiaomiBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[XiaomiPassiveBluetoothDataProcessor],
PassiveBluetoothProcessorEntity[XiaomiPassiveBluetoothDataProcessor[bool | None]],
BinarySensorEntity,
):
"""Representation of a Xiaomi binary sensor."""

View file

@ -2,9 +2,9 @@
from collections.abc import Callable, Coroutine
from logging import Logger
from typing import Any
from typing import Any, TypeVar
from xiaomi_ble import XiaomiBluetoothDeviceData
from xiaomi_ble import SensorUpdate, XiaomiBluetoothDeviceData
from homeassistant.components.bluetooth import (
BluetoothScanningMode,
@ -22,8 +22,12 @@ from homeassistant.helpers.debounce import Debouncer
from .const import CONF_SLEEPY_DEVICE
_T = TypeVar("_T")
class XiaomiActiveBluetoothProcessorCoordinator(ActiveBluetoothProcessorCoordinator):
class XiaomiActiveBluetoothProcessorCoordinator(
ActiveBluetoothProcessorCoordinator[SensorUpdate]
):
"""Define a Xiaomi Bluetooth Active Update Processor Coordinator."""
def __init__(
@ -33,13 +37,13 @@ class XiaomiActiveBluetoothProcessorCoordinator(ActiveBluetoothProcessorCoordina
*,
address: str,
mode: BluetoothScanningMode,
update_method: Callable[[BluetoothServiceInfoBleak], Any],
update_method: Callable[[BluetoothServiceInfoBleak], SensorUpdate],
needs_poll_method: Callable[[BluetoothServiceInfoBleak, float | None], bool],
device_data: XiaomiBluetoothDeviceData,
discovered_event_classes: set[str],
poll_method: Callable[
[BluetoothServiceInfoBleak],
Coroutine[Any, Any, Any],
Coroutine[Any, Any, SensorUpdate],
]
| None = None,
poll_debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None,
@ -68,7 +72,9 @@ class XiaomiActiveBluetoothProcessorCoordinator(ActiveBluetoothProcessorCoordina
return self.entry.data.get(CONF_SLEEPY_DEVICE, self.device_data.sleepy_device)
class XiaomiPassiveBluetoothDataProcessor(PassiveBluetoothDataProcessor):
class XiaomiPassiveBluetoothDataProcessor(
PassiveBluetoothDataProcessor[_T, SensorUpdate]
):
"""Define a Xiaomi Bluetooth Passive Update Data Processor."""
coordinator: XiaomiActiveBluetoothProcessorCoordinator

View file

@ -2,6 +2,8 @@
from __future__ import annotations
from typing import cast
from xiaomi_ble import DeviceClass, SensorUpdate, Units
from xiaomi_ble.parser import ExtendedSensorDeviceClass
@ -162,7 +164,7 @@ SENSOR_DESCRIPTIONS = {
def sensor_update_to_bluetooth_data_update(
sensor_update: SensorUpdate,
) -> PassiveBluetoothDataUpdate:
) -> PassiveBluetoothDataUpdate[float | None]:
"""Convert a sensor update to a bluetooth data update."""
return PassiveBluetoothDataUpdate(
devices={
@ -177,7 +179,9 @@ def sensor_update_to_bluetooth_data_update(
if description.device_class
},
entity_data={
device_key_to_bluetooth_entity_key(device_key): sensor_values.native_value
device_key_to_bluetooth_entity_key(device_key): cast(
float | None, sensor_values.native_value
)
for device_key, sensor_values in sensor_update.entity_values.items()
},
entity_names={
@ -210,7 +214,7 @@ async def async_setup_entry(
class XiaomiBluetoothSensorEntity(
PassiveBluetoothProcessorEntity[XiaomiPassiveBluetoothDataProcessor],
PassiveBluetoothProcessorEntity[XiaomiPassiveBluetoothDataProcessor[float | None]],
SensorEntity,
):
"""Representation of a xiaomi ble sensor."""