82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from importlib.metadata import version
|
|
from typing import TypedDict
|
|
|
|
from bleak import BleakScanner
|
|
from bleak.backends.device import BLEDevice
|
|
from bleak.backends.scanner import AdvertisementData
|
|
|
|
from k1link.artifacts import utc_now_iso
|
|
|
|
|
|
class BleDeviceRecord(TypedDict):
|
|
macos_uuid: str
|
|
id_kind: str
|
|
name: str | None
|
|
local_name: str | None
|
|
rssi: int
|
|
tx_power: int | None
|
|
service_uuids: list[str]
|
|
manufacturer_data_hex: dict[str, str]
|
|
service_data_hex: dict[str, str]
|
|
k1_name_candidate: bool
|
|
|
|
|
|
class BleScanResult(TypedDict):
|
|
schema_version: int
|
|
started_at_utc: str
|
|
completed_at_utc: str
|
|
duration_seconds: float
|
|
adapter: str
|
|
bleak_version: str
|
|
device_count: int
|
|
devices: list[BleDeviceRecord]
|
|
|
|
|
|
def advertisement_record(device: BLEDevice, advertisement: AdvertisementData) -> BleDeviceRecord:
|
|
name = advertisement.local_name or device.name
|
|
normalized_name = (name or "").casefold()
|
|
return {
|
|
"macos_uuid": device.address,
|
|
"id_kind": "corebluetooth_uuid",
|
|
"name": device.name,
|
|
"local_name": advertisement.local_name,
|
|
"rssi": advertisement.rssi,
|
|
"tx_power": advertisement.tx_power,
|
|
"service_uuids": sorted(advertisement.service_uuids),
|
|
"manufacturer_data_hex": {
|
|
str(company_id): data.hex()
|
|
for company_id, data in sorted(advertisement.manufacturer_data.items())
|
|
},
|
|
"service_data_hex": {
|
|
service_uuid: data.hex()
|
|
for service_uuid, data in sorted(advertisement.service_data.items())
|
|
},
|
|
"k1_name_candidate": any(marker in normalized_name for marker in ("lixel", "xgrids", "k1")),
|
|
}
|
|
|
|
|
|
async def scan(duration_seconds: float) -> BleScanResult:
|
|
if duration_seconds <= 0:
|
|
raise ValueError("duration_seconds must be positive")
|
|
|
|
started_at = utc_now_iso()
|
|
discovered = await BleakScanner.discover(timeout=duration_seconds, return_adv=True)
|
|
devices = [
|
|
advertisement_record(device, advertisement) for device, advertisement in discovered.values()
|
|
]
|
|
devices.sort(
|
|
key=lambda item: (not item["k1_name_candidate"], -item["rssi"], item["macos_uuid"])
|
|
)
|
|
return {
|
|
"schema_version": 1,
|
|
"started_at_utc": started_at,
|
|
"completed_at_utc": utc_now_iso(),
|
|
"duration_seconds": duration_seconds,
|
|
"adapter": "CoreBluetooth",
|
|
"bleak_version": version("bleak"),
|
|
"device_count": len(devices),
|
|
"devices": devices,
|
|
}
|