""" Android BLE 组件 — BluetoothLeScanner, ScanCallback, ScanFilter 等 方法签名完全对齐 Android SDK """ import os import sys from typing import Any, Dict, List, Optional, Set, Tuple, Union from .._core import ( Stub, AndroidOnlyError, android_context, resolve_android_path, to_android_path, _sdcard_root, _app_data_dir, _package_name, ) # ═══════════════════════════════════════════════════════════════ # ScanRecord # ═══════════════════════════════════════════════════════════════ class ScanRecord(Stub): """ScanRecord — 扫描记录""" def __init__(self): super().__init__() self._service_uuids: List[str] = [] self._service_solicitation_uuids: List[str] = [] self._service_data: Dict[str, bytes] = {} self._manufacturer_specific_data: Dict[int, bytes] = {} self._tx_power_level: int = -127 self._device_name: str = "" self._advertise_flags: int = 0 self._bytes_data: bytes = b"" def getServiceUuids(self) -> Optional[List[str]]: return list(self._service_uuids) if self._service_uuids else None def getServiceSolicitationUuids(self) -> Optional[List[str]]: return list(self._service_solicitation_uuids) if self._service_solicitation_uuids else None def getServiceData(self) -> Optional[Dict[str, bytes]]: return dict(self._service_data) if self._service_data else None def getServiceDataForService(self, uuid_str: str) -> Optional[bytes]: return self._service_data.get(uuid_str) def getManufacturerSpecificData(self) -> Optional[Dict[int, bytes]]: return dict(self._manufacturer_specific_data) if self._manufacturer_specific_data else None def getTxPowerLevel(self) -> int: return self._tx_power_level def getDeviceName(self) -> Optional[str]: return self._device_name or None def getAdvertiseFlags(self) -> int: return self._advertise_flags def getBytes(self) -> bytes: return self._bytes_data # ═══════════════════════════════════════════════════════════════ # ScanFilter # ═══════════════════════════════════════════════════════════════ class ScanFilter(Stub): """ScanFilter — 扫描过滤器""" class Builder(Stub): def __init__(self): super().__init__() self._device_name: Optional[str] = None self._device_address: Optional[str] = None self._service_uuid: Optional[str] = None self._service_data: Optional[bytes] = None self._manufacturer_id: int = -1 self._manufacturer_data: Optional[bytes] = None def setDeviceName(self, name: str) -> "ScanFilter.Builder": self._device_name = name return self def setDeviceAddress(self, address: str) -> "ScanFilter.Builder": self._device_address = address return self def setServiceUuid(self, uuid_str: str) -> "ScanFilter.Builder": self._service_uuid = uuid_str return self def setServiceData(self, data: bytes) -> "ScanFilter.Builder": self._service_data = data return self def setManufacturerData(self, manufacturer_id: int, data: bytes) -> "ScanFilter.Builder": self._manufacturer_id = manufacturer_id self._manufacturer_data = data return self def build(self) -> "ScanFilter": return ScanFilter( device_name=self._device_name, device_address=self._device_address, service_uuid=self._service_uuid, service_data=self._service_data, manufacturer_id=self._manufacturer_id, manufacturer_data=self._manufacturer_data, ) def __init__(self, device_name: Optional[str] = None, device_address: Optional[str] = None, service_uuid: Optional[str] = None, service_data: Optional[bytes] = None, manufacturer_id: int = -1, manufacturer_data: Optional[bytes] = None): super().__init__() self._device_name = device_name self._device_address = device_address self._service_uuid = service_uuid self._service_data = service_data self._manufacturer_id = manufacturer_id self._manufacturer_data = manufacturer_data class ScanFilterBuilder(ScanFilter.Builder): """ScanFilter.Builder 别名""" pass # ═══════════════════════════════════════════════════════════════ # ScanSettings # ═══════════════════════════════════════════════════════════════ class ScanSettings(Stub): """ScanSettings — 扫描设置""" SCAN_MODE_OPPORTUNISTIC = -1 SCAN_MODE_LOW_POWER = 0 SCAN_MODE_BALANCED = 1 SCAN_MODE_LOW_LATENCY = 2 CALLBACK_TYPE_ALL_MATCHES = 1 CALLBACK_TYPE_FIRST_MATCH = 2 CALLBACK_TYPE_MATCH_LOST = 4 MATCH_NUM_FEW_CFG = 2 MATCH_NUM_MAX_CFG = 1 MATCH_MODE_AGGRESSIVE = 0 MATCH_MODE_STICKY = 1 class Builder(Stub): def __init__(self): super().__init__() self._scan_mode = ScanSettings.SCAN_MODE_LOW_POWER self._callback_type = ScanSettings.CALLBACK_TYPE_ALL_MATCHES self._legacy = True self._phy = 1 def setScanMode(self, scan_mode: int) -> "ScanSettings.Builder": self._scan_mode = scan_mode return self def setCallbackType(self, callback_type: int) -> "ScanSettings.Builder": self._callback_type = callback_type return self def setLegacy(self, legacy: bool) -> "ScanSettings.Builder": self._legacy = legacy return self def setPhy(self, phy: int) -> "ScanSettings.Builder": self._phy = phy return self def build(self) -> "ScanSettings": return ScanSettings( scan_mode=self._scan_mode, callback_type=self._callback_type, legacy=self._legacy, phy=self._phy, ) def __init__(self, scan_mode: int = SCAN_MODE_LOW_POWER, callback_type: int = CALLBACK_TYPE_ALL_MATCHES, legacy: bool = True, phy: int = 1): super().__init__() self._scan_mode = scan_mode self._callback_type = callback_type self._legacy = legacy self._phy = phy def getScanMode(self) -> int: return self._scan_mode def getCallbackType(self) -> int: return self._callback_type def getLegacy(self) -> bool: return self._legacy def getPhy(self) -> int: return self._phy # ═══════════════════════════════════════════════════════════════ # ScanResult # ═══════════════════════════════════════════════════════════════ class ScanResult(Stub): """ScanResult — BLE 扫描结果""" DATA_COMPLETE = 0 DATA_TRUNCATED = 2 PHY_LE_1M = 1 PHY_LE_2M = 2 PHY_LE_CODED = 3 PERIODIC_INTERVAL_NOT_PRESENT = 0 def __init__(self, device: Any = None, scan_record: Optional[ScanRecord] = None, rssi: int = -127, timestamp_nanos: int = 0): super().__init__() self._device = device self._scan_record = scan_record or ScanRecord() self._rssi = rssi self._timestamp_nanos = timestamp_nanos self._primary_phy = self.PHY_LE_1M self._secondary_phy = 0 self._advertising_sid = -1 self._tx_power = -127 self._periodic_interval = self.PERIODIC_INTERVAL_NOT_PRESENT self._data_status = self.DATA_COMPLETE def getDevice(self) -> Any: return self._device def getScanRecord(self) -> Optional[ScanRecord]: return self._scan_record def getRssi(self) -> int: return self._rssi def getTimestampNanos(self) -> int: return self._timestamp_nanos def getPrimaryPhy(self) -> int: return self._primary_phy def getSecondaryPhy(self) -> int: return self._secondary_phy def getAdvertisingSid(self) -> int: return self._advertising_sid def getTxPower(self) -> int: return self._tx_power def getPeriodicAdvertisingInterval(self) -> int: return self._periodic_interval def getDataStatus(self) -> int: return self._data_status # ═══════════════════════════════════════════════════════════════ # ScanCallback # ═══════════════════════════════════════════════════════════════ class ScanCallback(Stub): """ScanCallback — 扫描回调""" SCAN_FAILED_ALREADY_STARTED = 1 SCAN_FAILED_APPLICATION_REGISTRATION_FAILED = 2 SCAN_FAILED_INTERNAL_ERROR = 3 SCAN_FAILED_FEATURE_UNSUPPORTED = 4 SCAN_FAILED_OUT_OF_HARDWARE_RESOURCES = 5 SCAN_FAILED_SCANNING_TOO_FREQUENTLY = 6 def onScanResult(self, callback_type: int, result: ScanResult): pass def onBatchScanResults(self, results: List[ScanResult]): pass def onScanFailed(self, error_code: int): pass # ═══════════════════════════════════════════════════════════════ # BluetoothLeScanner # ═══════════════════════════════════════════════════════════════ class BluetoothLeScanner(Stub): """BluetoothLeScanner — BLE 扫描器""" def __init__(self): super().__init__() def startScan(self, filters: Optional[List[ScanFilter]] = None, settings: Optional[ScanSettings] = None, callback: Optional[ScanCallback] = None): pass def startScanWithFilters(self, filters: List[ScanFilter], callback: ScanCallback): self.startScan(filters, None, callback) def stopScan(self, callback: Optional[ScanCallback] = None): pass def flushPendingScanResults(self, callback: ScanCallback): pass # ═══════════════════════════════════════════════════════════════ # AdvertiseData / AdvertiseSettings / AdvertiseCallback # ═══════════════════════════════════════════════════════════════ class AdvertiseData(Stub): """AdvertiseData — 广播数据""" class Builder(Stub): def __init__(self): super().__init__() self._include_device_name = False self._include_tx_power_level = False self._service_uuids: List[str] = [] self._service_data: Dict[str, bytes] = {} self._manufacturer_data: Dict[int, bytes] = {} def setIncludeDeviceName(self, include: bool) -> "AdvertiseData.Builder": self._include_device_name = include return self def setIncludeTxPowerLevel(self, include: bool) -> "AdvertiseData.Builder": self._include_tx_power_level = include return self def addServiceUuid(self, uuid_str: str) -> "AdvertiseData.Builder": self._service_uuids.append(uuid_str) return self def addServiceData(self, uuid_str: str, data: bytes) -> "AdvertiseData.Builder": self._service_data[uuid_str] = data return self def addManufacturerData(self, manufacturer_id: int, data: bytes) -> "AdvertiseData.Builder": self._manufacturer_data[manufacturer_id] = data return self def build(self) -> "AdvertiseData": return AdvertiseData( include_device_name=self._include_device_name, include_tx_power_level=self._include_tx_power_level, service_uuids=self._service_uuids, service_data=self._service_data, manufacturer_data=self._manufacturer_data, ) def __init__(self, include_device_name: bool = False, include_tx_power_level: bool = False, service_uuids: Optional[List[str]] = None, service_data: Optional[Dict[str, bytes]] = None, manufacturer_data: Optional[Dict[int, bytes]] = None): super().__init__() self._include_device_name = include_device_name self._include_tx_power_level = include_tx_power_level self._service_uuids = service_uuids or [] self._service_data = service_data or {} self._manufacturer_data = manufacturer_data or {} class AdvertiseDataBuilder(AdvertiseData.Builder): pass class AdvertiseSettings(Stub): """AdvertiseSettings — 广播设置""" ADVERTISE_MODE_LOW_POWER = 0 ADVERTISE_MODE_BALANCED = 1 ADVERTISE_MODE_LOW_LATENCY = 2 ADVERTISE_TX_POWER_ULTRA_LOW = 0 ADVERTISE_TX_POWER_LOW = 1 ADVERTISE_TX_POWER_MEDIUM = 2 ADVERTISE_TX_POWER_HIGH = 3 class Builder(Stub): def __init__(self): super().__init__() self._advertise_mode = AdvertiseSettings.ADVERTISE_MODE_LOW_POWER self._tx_power_level = AdvertiseSettings.ADVERTISE_TX_POWER_MEDIUM self._connectable = True self._timeout = 0 def setAdvertiseMode(self, mode: int) -> "AdvertiseSettings.Builder": self._advertise_mode = mode return self def setTxPowerLevel(self, power: int) -> "AdvertiseSettings.Builder": self._tx_power_level = power return self def setConnectable(self, connectable: bool) -> "AdvertiseSettings.Builder": self._connectable = connectable return self def setTimeout(self, timeout: int) -> "AdvertiseSettings.Builder": self._timeout = timeout return self def build(self) -> "AdvertiseSettings": return AdvertiseSettings( advertise_mode=self._advertise_mode, tx_power_level=self._tx_power_level, connectable=self._connectable, timeout=self._timeout, ) def __init__(self, advertise_mode: int = ADVERTISE_MODE_LOW_POWER, tx_power_level: int = ADVERTISE_TX_POWER_MEDIUM, connectable: bool = True, timeout: int = 0): super().__init__() self._advertise_mode = advertise_mode self._tx_power_level = tx_power_level self._connectable = connectable self._timeout = timeout def getAdvertiseMode(self) -> int: return self._advertise_mode def getTxPowerLevel(self) -> int: return self._tx_power_level def isConnectable(self) -> bool: return self._connectable def getTimeout(self) -> int: return self._timeout class AdvertiseCallback(Stub): """AdvertiseCallback — 广播回调""" ADVERTISE_FAILED_DATA_TOO_LARGE = 1 ADVERTISE_FAILED_TOO_MANY_ADVERTISERS = 2 ADVERTISE_FAILED_ALREADY_STARTED = 3 ADVERTISE_FAILED_INTERNAL_ERROR = 4 ADVERTISE_FAILED_FEATURE_UNSUPPORTED = 5 def onStartSuccess(self, settings_in_effect: AdvertiseSettings): pass def onStartFailure(self, error_code: int): pass class BluetoothLeAdvertiser(Stub): """BluetoothLeAdvertiser — BLE 广播器""" def startAdvertising(self, settings: AdvertiseSettings, advertise_data: AdvertiseData, scan_response: Optional[AdvertiseData], callback: AdvertiseCallback): try: callback.onStartSuccess(settings) except Exception: pass def stopAdvertising(self, callback: AdvertiseCallback): pass class PeriodicAdvertisingManager(Stub): pass class PeriodicAdvertisingCallback(Stub): pass class DistanceMeasurementManager(Stub): pass class DistanceMeasurementResult(Stub): pass # ═══════════════════════════════════════════════════════════════ # __all__ # ═══════════════════════════════════════════════════════════════ __all__ = [ "BluetoothLeScanner", "ScanCallback", "ScanResult", "ScanFilter", "ScanFilterBuilder", "ScanSettings", "ScanRecord", "BluetoothLeAdvertiser", "AdvertiseCallback", "AdvertiseData", "AdvertiseDataBuilder", "AdvertiseSettings", "PeriodicAdvertisingManager", "PeriodicAdvertisingCallback", "DistanceMeasurementManager", "DistanceMeasurementResult", ]