1
2
3
4
5
6

aioesphomeapi >= 45.3.1
bleak >= 3.0.2
bleak-retry-connector >= 4.6.1
bluetooth-data-tools >= 1.18.0
habluetooth >= 6.26.5
lru-dict >= 1.2.0

1
2

pip install bleak
pip install bleak-esphome

1

from bleak import BleakScanner, BleakClient

1
2
3
4
5
6

import asyncio

async def main() -> None:
    ...

asyncio.run(main())

1
2
3
4
5
6
7
8
9

import asyncio
from bleak import BleakScanner

async def main() -> None:
    devices = await BleakScanner.discover(timeout=5.0)
    for dev in devices:
        print(dev.address, dev.name, dev.details)

asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import asyncio
from bleak import BleakScanner

async def main() -> None:
    result = await BleakScanner.discover(timeout=5.0, return_adv=True)
    for address, (device, adv) in result.items():
        print("address:", address)
        print("device:", device)
        print("name:", adv.local_name)
        print("rssi:", adv.rssi)
        print("service_uuids:", adv.service_uuids)
        print("manufacturer_data:", adv.manufacturer_data)
        print("service_data:", adv.service_data)
        print()

asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import asyncio
from bleak import BleakScanner

async def main() -> None:
    dev = await BleakScanner.find_device_by_name("LYWSD03MMC", timeout=10.0)
    if dev is None:
        print("not found")
        return
    print(dev.address, dev.name)

asyncio.run(main())

1

dev = await BleakScanner.find_device_by_address("AA:BB:CC:DD:EE:FF", timeout=10.0)

1
2
3
4
5
6
7
8
9

import asyncio
from bleak import BleakScanner

async def main() -> None:
    async with BleakScanner() as scanner:
        async for device, adv in scanner.advertisement_data():
            print(device.address, adv.rssi, adv.local_name, adv.service_data)

asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import asyncio
from bleak import BleakScanner
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData


def on_detect(device: BLEDevice, adv: AdvertisementData) -> None:
    if adv.local_name:
        print(device.address, adv.local_name, adv.rssi)

async def main() -> None:
    async with BleakScanner(detection_callback=on_detect):
        await asyncio.sleep(10)

asyncio.run(main())

1
2

scanner = BleakScanner(scanning_mode="active")   # 默认
scanner = BleakScanner(scanning_mode="passive")

1

scanner = BleakScanner(bluez={"adapter": "hci0"})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import asyncio
from bleak import BleakClient, BleakScanner

MODEL_NUMBER_UUID = "2A24"

async def main() -> None:
    device = await BleakScanner.find_device_by_name("MyDevice", timeout=10.0)
    if device is None:
        raise RuntimeError("device not found")

    async with BleakClient(device) as client:
        data = await client.read_gatt_char(MODEL_NUMBER_UUID)
        print(data.decode(errors="replace"))

asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import asyncio
from bleak import BleakClient, BleakScanner

async def main() -> None:
    device = await BleakScanner.find_device_by_name("MyDevice", timeout=10.0)
    if device is None:
        return

    async with BleakClient(device) as client:
        for service in client.services:
            print(f"[Service] {service.uuid} handle={service.handle} {service.description}")
            for char in service.characteristics:
                print(
                    f"  [Char] {char.uuid} handle={char.handle} "
                    f"props={char.properties} max_wwr={char.max_write_without_response_size}"
                )
                for desc in char.descriptors:
                    print(f"    [Desc] {desc.uuid} handle={desc.handle}")

asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

target = None
for service in client.services:
    for char in service.characteristics:
        if char.uuid == "0000xxxx-0000-1000-8000-00805f9b34fb" and char.handle == 37:
            target = char

if target is None:
    raise RuntimeError("target characteristic not found")

data = await client.read_gatt_char(target)

1
2

await client.write_gatt_char(WRITE_UUID, b"\x01\x02\x03", response=True)
await client.write_gatt_char(WRITE_UUID, b"\x04\x05", response=False)

1
2
3
4
5
6
7
8
9

async def write_chunks(client, char, payload: bytes, *, response: bool = False):
    if response:
        # ATT Write Request 理论最大 512；很多设备应用层协议会更小。
        chunk_size = 512
    else:
        chunk_size = char.max_write_without_response_size

    for offset in range(0, len(payload), chunk_size):
        await client.write_gatt_char(char, payload[offset:offset + chunk_size], response=response)

1

ValueError: Cannot write to CCCD (0x2902) directly. Use start_notify() or stop_notify() instead.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import asyncio
from bleak import BleakClient, BleakScanner

NOTIFY_UUID = "0000xxxx-0000-1000-8000-00805f9b34fb"


def on_notify(sender: BleakGATTCharacteristic, data: bytearray) -> None:
    print("notify from", sender.uuid, "handle", sender.handle, "data", data.hex())

async def main() -> None:
    device = await BleakScanner.find_device_by_name("MyDevice", timeout=10.0)
    if device is None:
        return

    async with BleakClient(device) as client:
        await client.start_notify(NOTIFY_UUID, on_notify)
        await asyncio.sleep(30)
        await client.stop_notify(NOTIFY_UUID)

asyncio.run(main())

1
2

async def on_notify(sender: BleakGATTCharacteristic, data: bytearray) -> None:
    await save_to_db(sender.uuid, bytes(data))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import asyncio
from bleak import BleakClient, BleakScanner

NOTIFY_UUID = "0000xxxx-0000-1000-8000-00805f9b34fb"

async def main() -> None:
    device = await BleakScanner.find_device_by_name("MyDevice", timeout=10.0)
    if device is None:
        return

    q: asyncio.Queue[bytes] = asyncio.Queue()

    def on_notify(sender: BleakGATTCharacteristic, data: bytearray) -> None:
        q.put_nowait(bytes(data))

    async with BleakClient(device) as client:
        await client.start_notify(NOTIFY_UUID, on_notify)
        try:
            while True:
                packet = await asyncio.wait_for(q.get(), timeout=10.0)
                print(packet.hex())
        finally:
            await client.stop_notify(NOTIFY_UUID)

asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

import asyncio
from dataclasses import dataclass
from bleak import BleakClient, BleakScanner
from bleak.backends.characteristic import BleakGATTCharacteristic
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData

SERVICE_UUID = "0000fff0-0000-1000-8000-00805f9b34fb"
WRITE_UUID = "0000fff1-0000-1000-8000-00805f9b34fb"
NOTIFY_UUID = "0000fff2-0000-1000-8000-00805f9b34fb"

@dataclass
class DeviceChars:
    write: BleakGATTCharacteristic
    notify: BleakGATTCharacteristic


def build_command(cmd: int, payload: bytes = b"") -> bytes:
    # 示例帧：AA 55 LEN CMD PAYLOAD XOR
    body = bytes([cmd]) + payload
    length = len(body)
    checksum = 0
    for b in body:
        checksum ^= b
    return b"\xaa\x55" + bytes([length]) + body + bytes([checksum])


def parse_response(data: bytes) -> dict[str, int | bytes]:
    if len(data) < 5 or data[:2] != b"\xaa\x55":
        raise ValueError(f"bad frame: {data.hex()}")
    length = data[2]
    body = data[3:3 + length]
    checksum = data[3 + length]
    calc = 0
    for b in body:
        calc ^= b
    if checksum != calc:
        raise ValueError("bad checksum")
    return {"cmd": body[0], "payload": body[1:]}


async def find_target() -> BLEDevice | None:
    def match(device: BLEDevice, adv: AdvertisementData) -> bool:
        return SERVICE_UUID in adv.service_uuids or adv.local_name == "MySensor"

    return await BleakScanner.find_device_by_filter(match, timeout=10.0)


def resolve_chars(client: BleakClient) -> DeviceChars:
    write = client.services.get_characteristic(WRITE_UUID)
    notify = client.services.get_characteristic(NOTIFY_UUID)
    if write is None or notify is None:
        raise RuntimeError("required characteristics not found")
    return DeviceChars(write=write, notify=notify)


async def main() -> None:
    device = await find_target()
    if device is None:
        raise RuntimeError("device not found")

    q: asyncio.Queue[bytes] = asyncio.Queue()

    def on_notify(sender: BleakGATTCharacteristic, data: bytearray) -> None:
        q.put_nowait(bytes(data))

    async with BleakClient(device, timeout=30.0) as client:
        chars = resolve_chars(client)
        await client.start_notify(chars.notify, on_notify)

        command = build_command(0x01)
        await client.write_gatt_char(chars.write, command, response=True)

        raw = await asyncio.wait_for(q.get(), timeout=5.0)
        print(parse_response(raw))

asyncio.run(main())

1
2

Len Type Data
... 0xFF ...

1

adv.manufacturer_data

1

adv.service_data

1

45 42 13 00 FA C0 E4 80 39 C0 16 24 10 34 51

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

import asyncio
from dataclasses import dataclass
from bleak import BleakScanner
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData


# 手册把厂商 ID 写成 0x4542，对应原始字节 45 42。
# BLE Manufacturer Specific Data 规范里的 Company Identifier 是 little-endian；
# 不同后端交给 Bleak 时通常已经剥离 company id，并把它作为 dict key。
# 为了兼容，这里同时接受 0x4542 和 0x4245。
EWD_COMPANY_IDS = {0x4542, 0x4245}


@dataclass(slots=True)
class EWD104BT58SensorFrame:
    address: str
    name: Optional[str]
    rssi: int
    company_id: int
    connectable: bool
    encrypted: bool
    accel_x_raw: int
    accel_y_raw: int
    accel_z_raw: int
    temperature_c: float
    humidity_percent: float
    battery_percent: int
    raw: bytes


def _parse_signed_be16(data: bytes) -> int:
    return int.from_bytes(data, byteorder="big", signed=True)


def _parse_temperature(integer_byte: int, fraction_byte: int) -> float:
    """解析手册中的温度格式。

    正温：0x16 0x24 -> 22.36℃。
    负温：整数部分最高位为 1；按 8 位补码解释整数部分，小数部分作为百分位。
    例如整数字节 0xEA 表示 -22，则 0xEA 0x24 -> -22.36℃。
    """
    fraction = fraction_byte / 100.0
    if integer_byte & 0x80:
        integer = integer_byte - 0x100
        return integer - fraction
    return integer_byte + fraction


def _parse_humidity(integer_byte: int, fraction_byte: int) -> float:
    return integer_byte + fraction_byte / 100.0


def _looks_like_sensor_payload(payload: bytes) -> bool:
    """判断是否像 EWD104-BT58 带传感器格式。

    Bleak 的 manufacturer_data value 通常不包含前 2 字节 company id，长度为 13：
        13 00 FA C0 E4 80 39 C0 16 24 10 34 51

    如果你从原始 AD 结构里自己剥数据，可能会拿到含 company id 的 15 字节：
        45 42 13 00 FA C0 E4 80 39 C0 16 24 10 34 51
    """
    if len(payload) == 13:
        flag = payload[0]
        encrypted = payload[1]
        battery = payload[12]
    elif len(payload) == 15 and payload[:2] == b"\x45\x42":
        flag = payload[2]
        encrypted = payload[3]
        battery = payload[14]
    else:
        return False

    return flag in (0x12, 0x13) and encrypted in (0x00, 0x01) and battery <= 100


def parse_ewd104_bt58_sensor_frame(device: BLEDevice, adv: AdvertisementData) -> EWD104BT58SensorFrame | None:
    """从 Bleak 的 AdvertisementData 中解析 EWD104-BT58 带传感器扫描响应。"""
    for company_id, payload in adv.manufacturer_data.items():
        # 常见 Bleak 表示：company_id 作为 key，payload 不再包含 company id。
        if company_id in EWD_COMPANY_IDS and _looks_like_sensor_payload(payload):
            return _parse_sensor_payload(
                address=device.address,
                name=adv.local_name or device.name,
                rssi=adv.rssi,
                company_id=company_id,
                payload=payload,
                raw=b"\x45\x42" + payload,
            )

        # 保险：有些自定义扫描栈可能把 45 42 留在 value 里。
        if _looks_like_sensor_payload(payload):
            return _parse_sensor_payload(
                address=device.address,
                name=adv.local_name or device.name,
                rssi=adv.rssi,
                company_id=company_id,
                payload=payload,
                raw=payload,
            )

    return None


def _parse_sensor_payload(
    *,
    address: str,
    name: Optional[str],
    rssi: int,
    company_id: int,
    payload: bytes,
    raw: bytes,
) -> EWD104BT58SensorFrame:
    if len(payload) == 15:
        # 原始数据包含 45 42 厂商 ID 时，先剥掉。
        company_id = int.from_bytes(payload[0:2], "big")
        payload = payload[2:]

    if len(payload) != 13:
        raise ValueError(f"unexpected EWD104-BT58 sensor payload length: {len(payload)}")

    connect_flag = payload[0]
    encrypted_flag = payload[1]
    accel_x = _parse_signed_be16(payload[2:4])
    accel_y = _parse_signed_be16(payload[4:6])
    accel_z = _parse_signed_be16(payload[6:8])
    temperature = _parse_temperature(payload[8], payload[9])
    humidity = _parse_humidity(payload[10], payload[11])
    battery = payload[12]

    return EWD104BT58SensorFrame(
        address=address,
        name=name,
        rssi=rssi,
        company_id=company_id,
        connectable=connect_flag == 0x13,
        encrypted=encrypted_flag == 0x01,
        accel_x_raw=accel_x,
        accel_y_raw=accel_y,
        accel_z_raw=accel_z,
        temperature_c=temperature,
        humidity_percent=humidity,
        battery_percent=battery,
        raw=raw,
    )


def on_detect(device: BLEDevice, adv: AdvertisementData) -> None:
    frame = parse_ewd104_bt58_sensor_frame(device, adv)
    if frame is None:
        return

    print(
        f"{frame.address} {frame.name or ''} RSSI={frame.rssi} "
        f"connectable={frame.connectable} encrypted={frame.encrypted} "
        f"accel=({frame.accel_x_raw},{frame.accel_y_raw},{frame.accel_z_raw}) "
        f"temp={frame.temperature_c:.2f}℃ "
        f"humidity={frame.humidity_percent:.2f}%RH "
        f"battery={frame.battery_percent}% "
        f"raw={frame.raw.hex()}"
    )


async def main() -> None:
    async with BleakScanner(detection_callback=on_detect, scanning_mode="active"):
        # 扫描响应通常需要 active scan 才更容易拿全。
        await asyncio.sleep(30)


if __name__ == "__main__":
    asyncio.run(main())

1

45 42 13 00 FA C0 E4 80 39 C0 16 24 10 34 51

1
2
3
4
5
6
7
8
9

company_id      = 0x4542
connectable     = True
encrypted       = False
accel_x_raw     = -1344
accel_y_raw     = -7040
accel_z_raw     = 14784
temperature_c   = 22.36
humidity_percent= 16.52
battery_percent = 81

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

@dataclass(slots=True)
class EWD104BT58PlainFrame:
    mac: str
    major: int
    minor: int
    tx_power_index: int
    adv_interval_ms: int
    battery_percent: int


def parse_plain_payload(payload: bytes) -> EWD104BT58PlainFrame | None:
    # Bleak manufacturer_data 的 value 通常已经去掉 2 字节 company id，
    # 无传感器格式剩余长度应为：MAC 6 + Major 2 + Minor 2 + Power 1 + Interval 2 + Battery 1 = 14。
    if len(payload) == 16:
        # 如果还带着 0x18 0x03 这类前缀，先剥掉。
        payload = payload[2:]
    if len(payload) != 14:
        return None

    mac = ":".join(f"{b:02X}" for b in payload[0:6])
    major = int.from_bytes(payload[6:8], "big")
    minor = int.from_bytes(payload[8:10], "big")
    tx_power_index = payload[10]
    adv_interval_ms = int.from_bytes(payload[11:13], "big")
    battery_percent = payload[13]

    return EWD104BT58PlainFrame(
        mac=mac,
        major=major,
        minor=minor,
        tx_power_index=tx_power_index,
        adv_interval_ms=adv_interval_ms,
        battery_percent=battery_percent,
    )

1
2
3
4
5
6

sem = asyncio.Semaphore(3)

async def poll_device(device: BLEDevice) -> bytes:
    async with sem:
        async with BleakClient(device, timeout=20.0) as client:
            return await client.read_gatt_char("2A19")

1
2
3

bluetoothctl show
bluetoothctl scan on
hciconfig -a

1
2

async with BleakClient(device, pair=True, timeout=60.0) as client:
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

Python 应用
  │
  │ Bleak API: BleakScanner / BleakClient
  │
bleak-esphome backend
  │
  │ ESPHome Native API / aioesphomeapi，默认 6053 端口，可带 noise_psk
  │
ESP32 + ESPHome bluetooth_proxy
  │
  │ 本地 BLE radio
  │
BLE 外设：温湿度计、门锁、手环、传感器……

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

bleak_esphome/
  __init__.py
  connect.py
  connection_manager.py
  _cancellation.py
  backend/
    cache.py
    client.py
    device.py
    scanner.py

1
2
3
4
5
6

from .connect import connect_scanner
from .connection_manager import (
    APIConnectionManager,
    ESPHomeDeviceConfig,
    ESPHomeStartAborted,
)

1
2
3

class ESPHomeDeviceConfig(TypedDict):
    address: str
    noise_psk: str | None

1
2
3
4
5
6

APIClient(
    address=self._address,
    port=6053,
    password=None,
    noise_psk=self._noise_psk,
)

1
2

async_on_advertisement(self, adv: BluetoothLEAdvertisement)
async_on_raw_advertisements(self, raw: BluetoothLERawAdvertisementsResponse)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

self._async_on_advertisement(
    int_to_bluetooth_address(adv.address),
    adv.rssi,
    adv.name,
    adv.service_uuids,
    adv.service_data,
    adv.manufacturer_data,
    None,
    {"address_type": adv.address_type},
    MONOTONIC_TIME(),
)

1
2
3
4
5
6
7

self._async_on_raw_advertisement(
    int_to_bluetooth_address(adv.address),
    adv.rssi,
    adv.data,
    {"address_type": adv.address_type},
    now,
)

1
2
3

await self._wait_for_free_connection_slot(CONNECT_FREE_SLOT_TIMEOUT)
await self._client.bluetooth_device_connect(...)
await self._get_services(...)

1
2
3
4

BleakGATTService(...)
BleakGATTCharacteristic(...)
BleakGATTDescriptor(...)
BleakGATTServiceCollection()

1

cccd_descriptor = characteristic.get_descriptor(CCCD_UUID)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import asyncio
from bleak_esphome import APIConnectionManager

async def main() -> None:
    manager = APIConnectionManager({
        "address": "192.168.1.50",
        "noise_psk": None,  # 如果 ESPHome API 配了 encryption key，这里填 noise_psk
    })
    await manager.start()
    try:
        print("ESPHome bluetooth proxy registered")
        await asyncio.Event().wait()
    finally:
        await manager.stop()

asyncio.run(main())

1
2
3
4

async def read_battery(device: BLEDevice) -> int:
    async with BleakClient(device, timeout=30.0) as client:
        data = await client.read_gatt_char("2A19")
        return data[0]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

bluetooth_adapters = get_adapters()
bluetooth_storage = BluetoothStorage(hass)
slot_manager = BleakSlotManager()
integration_matcher = IntegrationMatcher(await async_get_bluetooth(hass))

manager = HomeAssistantBluetoothManager(
    hass, integration_matcher, bluetooth_adapters, bluetooth_storage, slot_manager
)
set_manager(manager)
await manager.async_setup()

1
2
3
4

@hass_callback
def async_get_scanner(hass: HomeAssistant) -> BleakScanner:
    """Return a HaBleakScannerWrapper cast to BleakScanner."""
    return cast(BleakScanner, HaBleakScannerWrapper())

1
2
3
4
5
6
7

self._scale = AcaiaScale(
    address_or_ble_device=entry.data[CONF_ADDRESS],
    name=entry.title,
    is_new_style_scale=entry.data[CONF_IS_NEW_STYLE_SCALE],
    notify_callback=debouncer.async_schedule_call,
    scanner=async_get_scanner(hass),
)

1
2
3
4
5
6
7

bleak_scanner_instance = bluetooth.async_get_scanner(hass)

controller = Controller(
    async_zeroconf_instance=async_zeroconf_instance,
    bleak_scanner_instance=bleak_scanner_instance,
    char_cache=char_cache,
)

1
2

type BluetoothCallback = Callable[[BluetoothServiceInfoBleak, BluetoothChange], None]
type ProcessAdvertisementCallback = Callable[[BluetoothServiceInfoBleak], bool]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def _discover_service_info(self, service_info: BluetoothServiceInfoBleak) -> None:
    matched_domains = self._integration_matcher.match_domains(service_info)

    for match in self._callback_index.match_callbacks(service_info):
        callback = match[CALLBACK]
        callback(service_info, BluetoothChange.ADVERTISEMENT)

    for domain in matched_domains:
        discovery_flow.async_create_flow(
            self.hass,
            domain,
            {"source": config_entries.SOURCE_BLUETOOTH},
            service_info,
            discovery_key=discovery_key,
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def async_register_callback(
    hass: HomeAssistant,
    callback: BluetoothCallback,
    match_dict: BluetoothCallbackMatcher | None,
    mode: BluetoothScanningMode,
    *,
    scan_interval: float | None = None,
    scan_duration: float | None = None,
) -> Callable[[], None]:
    return _get_manager(hass).async_register_callback(
        callback, match_dict, mode, scan_interval, scan_duration
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

async def async_process_advertisements(
    hass: HomeAssistant,
    callback: ProcessAdvertisementCallback,
    match_dict: BluetoothCallbackMatcher,
    mode: BluetoothScanningMode,
    timeout: int,
) -> BluetoothServiceInfoBleak:
    done: Future[BluetoothServiceInfoBleak] = hass.loop.create_future()

    def _async_discovered_device(
        service_info: BluetoothServiceInfoBleak, change: BluetoothChange
    ) -> None:
        if not done.done() and callback(service_info):
            done.set_result(service_info)

    with ExitStack() as stack:
        unload = manager.async_register_callback(
            _async_discovered_device, match_dict, mode
        )
        stack.callback(unload)

        if mode is BluetoothScanningMode.ACTIVE:
            task = hass.async_create_task(manager.async_request_active_scan(timeout))
            stack.callback(task.cancel)

        async with asyncio.timeout(timeout):
            return await done

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

def _data_callback(info: BluetoothServiceInfoBleak) -> bool:
    if info.device.address != address:
        return False

    data.update(info.manufacturer_data.get(ManufacturerData.company, b""))
    return data.product_type is not ProductType.UNKNOWN

await bluetooth.async_process_advertisements(
    hass,
    _data_callback,
    bluetooth.BluetoothCallbackMatcher(
        address=address, manufacturer_id=ManufacturerData.company
    ),
    mode=bluetooth.BluetoothScanningMode.ACTIVE,
    timeout=PRODUCT_TYPE_TIMEOUT,
)

1
2
3
4
5
6

@hass_callback
def async_ble_device_from_address(
    hass: HomeAssistant, address: str, connectable: bool = True
) -> BLEDevice | None:
    """Return BLEDevice for an address if its present."""
    return _get_manager(hass).async_ble_device_from_address(address, connectable)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

if service_info.connectable:
    device = service_info.device
elif connectable_device := bluetooth.async_ble_device_from_address(
    hass, service_info.device.address, True
):
    device = connectable_device
else:
    return

async with BleakClient(device) as client:
    bat_char = await client.read_gatt_char(BATTERY_CHARACTERISTIC_UUID)
    battery = ord(bat_char)

1

update_method: Callable[[BluetoothServiceInfoBleak], _DataT]

1
2

async def poll_method(svc_info: BluetoothServiceInfoBleak) -> YourDataType:
    return YourDataType(...)

1
2

BluetoothServiceInfoBleak.device contains a BLEDevice. You should use this in
your poll function, as it is the most efficient way to get a BleakClient.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from bleak_esphome import connect_scanner

client_data = connect_scanner(cli, device_info, entry_data.available)
scanner = client_data.scanner

callbacks = [
    async_register_scanner(
        hass,
        scanner,
        source_domain=DOMAIN,
        source_model=device_info.model,
        source_config_entry_id=entry_data.entry_id,
        source_device_id=device_id,
    ),
    scanner.async_setup(),
]

1
2
3
4
5
6

ESP32 广播 / GATT 能力
  -> bleak-esphome ESPHomeScanner / ESPHomeClient
  -> habluetooth manager
  -> HomeAssistantBluetoothManager
  -> BluetoothServiceInfoBleak / BLEDevice
  -> 各个 HA 集成

1
2
3
4

广播解析：注册 bluetooth callback，处理 BluetoothServiceInfoBleak；
需要等待特定广播：用 async_process_advertisements；
需要连接：从 service_info.device 或 async_ble_device_from_address() 取得 BLEDevice，再交给 BleakClient；
第三方库要求 BleakScanner：传 bluetooth.async_get_scanner(hass)。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

from dataclasses import dataclass

@dataclass
class SensorAdvertisement:
    address: str
    rssi: int
    temperature: float | None
    humidity: float | None
    battery: int | None


def parse_adv(device: BLEDevice, adv: AdvertisementData) -> SensorAdvertisement | None:
    # 假设 service data UUID 是 0000181a-0000-1000-8000-00805f9b34fb
    data = adv.service_data.get("0000181a-0000-1000-8000-00805f9b34fb")
    if not data or len(data) < 5:
        return None

    temp_raw = int.from_bytes(data[0:2], "little", signed=True)
    hum_raw = int.from_bytes(data[2:4], "little")
    battery = data[4]

    return SensorAdvertisement(
        address=device.address,
        rssi=adv.rssi,
        temperature=temp_raw / 100,
        humidity=hum_raw / 100,
        battery=battery,
    )

1
2
3
4
5
6

HISTORY_CHAR = "0000fff3-0000-1000-8000-00805f9b34fb"

async def read_history(client: BleakClient) -> bytes:
    raw = await client.read_gatt_char(HISTORY_CHAR)
    # 解析历史记录……
    return raw

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import asyncio
from bleak import BleakClient, BleakScanner

async def scan_loop() -> None:
    async with BleakScanner() as scanner:
        async for device, adv in scanner.advertisement_data():
            parsed = parse_adv(device, adv)
            if parsed:
                print(parsed)

async def read_history_once(device: BLEDevice) -> bytes:
    async with BleakClient(device, timeout=30.0) as client:
        return await read_history(client)

1
2
3
4
5

async def read_once(device: BLEDevice, char_uuid: str) -> bytes:
    async with BleakClient(device, timeout=30.0) as client:
        return await asyncio.wait_for(client.read_gatt_char(char_uuid), timeout=10.0)

data = await asyncio.wait_for(read_once(device, UUID), timeout=60.0)

1
2

def on_notify(sender: BleakGATTCharacteristic, data: bytearray) -> None:
    requests.post("https://example.com", data=data)  # 阻塞，危险

1
2

def on_notify(sender: BleakGATTCharacteristic, data: bytearray) -> None:
    queue.put_nowait(bytes(data))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

async def read_with_retry(device: BLEDevice, char_uuid: str, attempts: int = 3) -> bytes:
    last_error = None
    for i in range(attempts):
        try:
            async with BleakClient(device, timeout=30.0) as client:
                return await client.read_gatt_char(char_uuid)
        except Exception as err:
            last_error = err
            await asyncio.sleep(0.5 * (i + 1))
    raise last_error

1
2
3
4

广播能解决的，尽量只解析广播；
必须连接时，短连接 + 超时 + 重试 + 明确释放；
业务协议只依赖 Bleak 抽象，不绑定本机蓝牙或某个 proxy；
覆盖不够时，用 ESPHome Bluetooth Proxy 扩展空间范围。