mirror of
https://github.com/home-assistant/core.git
synced 2025-07-18 18:57:06 +00:00
Relocate BluetoothServiceInfo to helpers.service_info (#75195)
This commit is contained in:
parent
5e46fa6f8b
commit
fde3489e86
@ -2,16 +2,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
import dataclasses
|
||||
from enum import Enum
|
||||
import fnmatch
|
||||
from functools import cached_property
|
||||
import logging
|
||||
import platform
|
||||
from typing import Final, TypedDict
|
||||
|
||||
from bleak import BleakError
|
||||
from bleak.backends.device import MANUFACTURERS, BLEDevice
|
||||
from bleak.backends.device import BLEDevice
|
||||
from bleak.backends.scanner import AdvertisementData
|
||||
from lru import LRU # pylint: disable=no-name-in-module
|
||||
|
||||
@ -23,8 +21,8 @@ from homeassistant.core import (
|
||||
HomeAssistant,
|
||||
callback as hass_callback,
|
||||
)
|
||||
from homeassistant.data_entry_flow import BaseServiceInfo
|
||||
from homeassistant.helpers import discovery_flow
|
||||
from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.loader import (
|
||||
BluetoothMatcher,
|
||||
@ -74,48 +72,6 @@ MANUFACTURER_ID: Final = "manufacturer_id"
|
||||
MANUFACTURER_DATA_FIRST_BYTE: Final = "manufacturer_data_first_byte"
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BluetoothServiceInfo(BaseServiceInfo):
|
||||
"""Prepared info from bluetooth entries."""
|
||||
|
||||
name: str
|
||||
address: str
|
||||
rssi: int
|
||||
manufacturer_data: dict[int, bytes]
|
||||
service_data: dict[str, bytes]
|
||||
service_uuids: list[str]
|
||||
|
||||
@classmethod
|
||||
def from_advertisement(
|
||||
cls, device: BLEDevice, advertisement_data: AdvertisementData
|
||||
) -> BluetoothServiceInfo:
|
||||
"""Create a BluetoothServiceInfo from an advertisement."""
|
||||
return cls(
|
||||
name=advertisement_data.local_name or device.name or device.address,
|
||||
address=device.address,
|
||||
rssi=device.rssi,
|
||||
manufacturer_data=advertisement_data.manufacturer_data,
|
||||
service_data=advertisement_data.service_data,
|
||||
service_uuids=advertisement_data.service_uuids,
|
||||
)
|
||||
|
||||
@cached_property
|
||||
def manufacturer(self) -> str | None:
|
||||
"""Convert manufacturer data to a string."""
|
||||
for manufacturer in self.manufacturer_data:
|
||||
if manufacturer in MANUFACTURERS:
|
||||
name: str = MANUFACTURERS[manufacturer]
|
||||
return name
|
||||
return None
|
||||
|
||||
@cached_property
|
||||
def manufacturer_id(self) -> int | None:
|
||||
"""Get the first manufacturer id."""
|
||||
for manufacturer in self.manufacturer_data:
|
||||
return manufacturer
|
||||
return None
|
||||
|
||||
|
||||
BluetoothChange = Enum("BluetoothChange", "ADVERTISEMENT")
|
||||
BluetoothCallback = Callable[[BluetoothServiceInfo, BluetoothChange], None]
|
||||
|
||||
|
@ -61,7 +61,7 @@ def is_bluetooth_device(device: Device) -> bool:
|
||||
def discover_devices(device_id: int) -> list[tuple[str, str]]:
|
||||
"""Discover Bluetooth devices."""
|
||||
try:
|
||||
result = bluetooth.discover_devices(
|
||||
result = bluetooth.discover_devices( # type: ignore[attr-defined]
|
||||
duration=8,
|
||||
lookup_names=True,
|
||||
flush_cache=True,
|
||||
@ -124,7 +124,7 @@ async def get_tracking_devices(hass: HomeAssistant) -> tuple[set[str], set[str]]
|
||||
def lookup_name(mac: str) -> str | None:
|
||||
"""Lookup a Bluetooth device name."""
|
||||
_LOGGER.debug("Scanning %s", mac)
|
||||
return bluetooth.lookup_name(mac, timeout=5) # type: ignore[no-any-return]
|
||||
return bluetooth.lookup_name(mac, timeout=5) # type: ignore[attr-defined,no-any-return]
|
||||
|
||||
|
||||
async def async_setup_scanner(
|
||||
@ -180,7 +180,7 @@ async def async_setup_scanner(
|
||||
if tasks:
|
||||
await asyncio.wait(tasks)
|
||||
|
||||
except bluetooth.BluetoothError:
|
||||
except bluetooth.BluetoothError: # type: ignore[attr-defined]
|
||||
_LOGGER.exception("Error looking up Bluetooth device")
|
||||
|
||||
async def update_bluetooth(now: datetime | None = None) -> None:
|
||||
|
@ -28,12 +28,12 @@ from .util import uuid as uuid_util
|
||||
from .util.decorator import Registry
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .components.bluetooth import BluetoothServiceInfo
|
||||
from .components.dhcp import DhcpServiceInfo
|
||||
from .components.hassio import HassioServiceInfo
|
||||
from .components.ssdp import SsdpServiceInfo
|
||||
from .components.usb import UsbServiceInfo
|
||||
from .components.zeroconf import ZeroconfServiceInfo
|
||||
from .helpers.service_info.bluetooth import BluetoothServiceInfo
|
||||
from .helpers.service_info.mqtt import MqttServiceInfo
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
@ -15,11 +15,11 @@ from .typing import UNDEFINED, DiscoveryInfoType, UndefinedType
|
||||
if TYPE_CHECKING:
|
||||
import asyncio
|
||||
|
||||
from homeassistant.components.bluetooth import BluetoothServiceInfo
|
||||
from homeassistant.components.dhcp import DhcpServiceInfo
|
||||
from homeassistant.components.ssdp import SsdpServiceInfo
|
||||
from homeassistant.components.zeroconf import ZeroconfServiceInfo
|
||||
|
||||
from .service_info.bluetooth import BluetoothServiceInfo
|
||||
from .service_info.mqtt import MqttServiceInfo
|
||||
|
||||
_R = TypeVar("_R", bound="Awaitable[bool] | bool")
|
||||
|
58
homeassistant/helpers/service_info/bluetooth.py
Normal file
58
homeassistant/helpers/service_info/bluetooth.py
Normal file
@ -0,0 +1,58 @@
|
||||
"""The bluetooth integration service info."""
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
from functools import cached_property
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from homeassistant.data_entry_flow import BaseServiceInfo
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from bleak.backends.device import BLEDevice
|
||||
from bleak.backends.scanner import AdvertisementData
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BluetoothServiceInfo(BaseServiceInfo):
|
||||
"""Prepared info from bluetooth entries."""
|
||||
|
||||
name: str
|
||||
address: str
|
||||
rssi: int
|
||||
manufacturer_data: dict[int, bytes]
|
||||
service_data: dict[str, bytes]
|
||||
service_uuids: list[str]
|
||||
|
||||
@classmethod
|
||||
def from_advertisement(
|
||||
cls, device: BLEDevice, advertisement_data: AdvertisementData
|
||||
) -> BluetoothServiceInfo:
|
||||
"""Create a BluetoothServiceInfo from an advertisement."""
|
||||
return cls(
|
||||
name=advertisement_data.local_name or device.name or device.address,
|
||||
address=device.address,
|
||||
rssi=device.rssi,
|
||||
manufacturer_data=advertisement_data.manufacturer_data,
|
||||
service_data=advertisement_data.service_data,
|
||||
service_uuids=advertisement_data.service_uuids,
|
||||
)
|
||||
|
||||
@cached_property
|
||||
def manufacturer(self) -> str | None:
|
||||
"""Convert manufacturer data to a string."""
|
||||
from bleak.backends.device import ( # pylint: disable=import-outside-toplevel
|
||||
MANUFACTURERS,
|
||||
)
|
||||
|
||||
for manufacturer in self.manufacturer_data:
|
||||
if manufacturer in MANUFACTURERS:
|
||||
name: str = MANUFACTURERS[manufacturer]
|
||||
return name
|
||||
return None
|
||||
|
||||
@cached_property
|
||||
def manufacturer_id(self) -> int | None:
|
||||
"""Get the first manufacturer id."""
|
||||
for manufacturer in self.manufacturer_data:
|
||||
return manufacturer
|
||||
return None
|
Loading…
x
Reference in New Issue
Block a user