""" Android Bluetooth 组件 — BluetoothAdapter, BluetoothDevice, BluetoothGatt 等 方法签名完全对齐 Android SDK """ import os import sys import uuid from typing import Any, Callable, 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, ) # ═══════════════════════════════════════════════════════════════ # BluetoothProfile # ═══════════════════════════════════════════════════════════════ class BluetoothProfile(Stub): """BluetoothProfile — 蓝牙配置档案""" A2DP = 2 HEALTH = 3 HEADSET = 1 GATT = 7 GATT_SERVER = 8 A2DP_SINK = 11 AVRC = 12 HEARING_AID = 21 LE_AUDIO = 22 LE_AUDIO_BROADCAST = 26 STATE_DISCONNECTED = 0 STATE_CONNECTING = 1 STATE_CONNECTED = 2 STATE_DISCONNECTING = 3 def getConnectedDevices(self) -> List[Any]: return [] def getDevicesMatchingConnectionState(self, state: int) -> List[Any]: return [] def getConnectionState(self, device: Any) -> int: return self.STATE_DISCONNECTED class BluetoothProfileServiceListener(Stub): """BluetoothProfile.ServiceListener""" def onServiceConnected(self, profile_id: int, proxy: Any): pass def onServiceDisconnected(self, profile_id: int): pass # ═══════════════════════════════════════════════════════════════ # BluetoothClass # ═══════════════════════════════════════════════════════════════ class BluetoothClass(Stub): """BluetoothClass — 蓝牙设备类别""" class Device(Stub): COMPUTER_DESKTOP = 0x100 COMPUTER_LAPTOP = 0x10C PHONE_SMART = 0x200 PHONE_UNCATEGORIZED = 0x204 AUDIO_VIDEO_WEARABLE_HEADSET = 0x404 AUDIO_VIDEO_HANDSFREE = 0x408 class Major(Stub): MISC = 0x0000 COMPUTER = 0x0100 PHONE = 0x0200 NETWORKING = 0x0300 AUDIO_VIDEO = 0x0400 PERIPHERAL = 0x0500 IMAGING = 0x0600 WEARABLE = 0x0700 TOY = 0x0800 HEALTH = 0x0900 UNCATEGORIZED = 0x1F00 def __init__(self): super().__init__() self._device_class = 0 self._major_class = 0 def getDeviceClass(self) -> int: return self._device_class def getMajorDeviceClass(self) -> int: return self._major_class def hasService(self, service: int) -> bool: return False def toString(self) -> str: return f"BluetoothClass(device=0x{self._device_class:04X})" # ═══════════════════════════════════════════════════════════════ # BluetoothSocket # ═══════════════════════════════════════════════════════════════ class BluetoothSocket(Stub): """BluetoothSocket — 蓝牙套接字""" TYPE_RFCOMM = 1 TYPE_L2CAP = 3 TYPE_SCO = 2 def __init__(self, type_: int = TYPE_RFCOMM): super().__init__() self._type = type_ self._closed = False self._connected = False self._device: Optional["BluetoothDevice"] = None def connect(self): self._connected = True def bind(self): pass def getInputStream(self) -> Any: return None def getOutputStream(self) -> Any: return None def close(self): self._closed = True self._connected = False def getRemoteDevice(self) -> Optional["BluetoothDevice"]: return self._device def getMaxReceivePacketSize(self) -> int: return 0 def isConnected(self) -> bool: return self._connected def isClosed(self) -> bool: return self._closed class BluetoothServerSocket(Stub): """BluetoothServerSocket — 蓝牙服务端套接字""" def __init__(self): super().__init__() self._closed = False def accept(self, timeout: int = 0) -> Optional[BluetoothSocket]: return None def close(self): self._closed = True # ═══════════════════════════════════════════════════════════════ # BluetoothDevice # ═══════════════════════════════════════════════════════════════ class BluetoothDevice(Stub): """BluetoothDevice — 蓝牙设备""" BOND_NONE = 10 BOND_BONDING = 11 BOND_BONDED = 12 DEVICE_TYPE_DUAL = 3 DEVICE_TYPE_LE = 2 DEVICE_TYPE_BREDR = 1 DEVICE_TYPE_UNKNOWN = 0 ACTION_FOUND = "android.bluetooth.device.action.FOUND" ACTION_NAME_CHANGED = "android.bluetooth.device.action.NAME_CHANGED" ACTION_BOND_STATE_CHANGED = "android.bluetooth.device.action.BOND_STATE_CHANGED" EXTRA_DEVICE = "android.bluetooth.device.extra.DEVICE" EXTRA_NAME = "android.bluetooth.device.extra.NAME" EXTRA_RSSI = "android.bluetooth.device.extra.RSSI" EXTRA_BOND_STATE = "android.bluetooth.device.extra.BOND_STATE" EXTRA_PREVIOUS_BOND_STATE = "android.bluetooth.device.extra.PREVIOUS_BOND_STATE" def __init__(self, address: str = "00:00:00:00:00:00"): super().__init__() self._address = address self._name = "Unknown" self._bond_state = self.BOND_NONE self._device_type = self.DEVICE_TYPE_UNKNOWN def getAddress(self) -> str: return self._address def getName(self) -> str: return self._name def setName(self, name: str): self._name = name def getBondState(self) -> int: return self._bond_state def setBondState(self, state: int): self._bond_state = state def getType(self) -> int: return self._device_type def untergehende(self) -> bool: return self._bond_state == self.BOND_BONDED def createRfcommSocketToServiceRecord(self, uuid_str: str) -> BluetoothSocket: sock = BluetoothSocket(BluetoothSocket.TYPE_RFCOMM) sock._device = self return sock def createInsecureRfcommSocketToServiceRecord(self, uuid_str: str) -> BluetoothSocket: sock = BluetoothSocket(BluetoothSocket.TYPE_RFCOMM) sock._device = self return sock def createL2capChannel(self, psm: int) -> BluetoothSocket: sock = BluetoothSocket(BluetoothSocket.TYPE_L2CAP) sock._device = self return sock def createInsecureL2capChannel(self, psm: int) -> BluetoothSocket: sock = BluetoothSocket(BluetoothSocket.TYPE_L2CAP) sock._device = self return sock def fetchUuidsWithSdp(self) -> bool: return False def uuidEquals(self, uuid_str: str) -> bool: return False def setPairingConfirmation(self, accept: bool): pass def setPin(self, pin: bytes) -> bool: return True def toString(self) -> str: return f"BluetoothDevice(address={self._address}, name={self._name})" # ═══════════════════════════════════════════════════════════════ # BluetoothGattDescriptor # ═══════════════════════════════════════════════════════════════ class BluetoothGattDescriptor(Stub): """BluetoothGattDescriptor — GATT 描述符""" ENABLE_NOTIFICATION_VALUE = bytes([0x01, 0x00]) ENABLE_INDICATION_VALUE = bytes([0x02, 0x00]) DISABLE_NOTIFICATION_VALUE = bytes([0x00, 0x00]) def __init__(self, uuid_str: str, permissions: int = 0): super().__init__() self._uuid = uuid_str self._value: Optional[bytes] = None self._permissions = permissions self._characteristic: Optional["BluetoothGattCharacteristic"] = None def getUuid(self) -> str: return self._uuid def getValue(self) -> Optional[bytes]: return self._value def setValue(self, value: bytes): self._value = value def getCharacteristic(self) -> Optional["BluetoothGattCharacteristic"]: return self._characteristic def getPermissions(self) -> int: return self._permissions # ═══════════════════════════════════════════════════════════════ # BluetoothGattCharacteristic # ═══════════════════════════════════════════════════════════════ class BluetoothGattCharacteristic(Stub): """BluetoothGattCharacteristic — GATT 特征""" PROPERTY_BROADCAST = 1 PROPERTY_READ = 2 PROPERTY_WRITE_NO_RESPONSE = 4 PROPERTY_WRITE = 8 PROPERTY_NOTIFY = 16 PROPERTY_INDICATE = 32 PROPERTY_SIGNED_WRITE = 64 PROPERTY_EXTENDED_PROPS = 128 WRITE_TYPE_NO_RESPONSE = 1 WRITE_TYPE_DEFAULT = 2 WRITE_TYPE_SIGNED = 4 PERMISSION_READ = 1 PERMISSION_READ_ENCRYPTED = 2 PERMISSION_READ_ENCRYPTED_MITM = 4 PERMISSION_WRITE = 16 PERMISSION_WRITE_ENCRYPTED = 32 PERMISSION_WRITE_ENCRYPTED_MITM = 64 FORMAT_FLOAT = 0x20 FORMAT_SFLOAT = 0x10 FORMAT_SINT16 = 0x02 FORMAT_SINT32 = 0x04 FORMAT_SINT8 = 0x01 FORMAT_UINT16 = 0x12 FORMAT_UINT32 = 0x14 FORMAT_UINT8 = 0x11 def __init__(self, uuid_str: str, properties: int = 0, permissions: int = 0): super().__init__() self._uuid = uuid_str self._properties = properties self._permissions = permissions self._write_type = self.WRITE_TYPE_DEFAULT self._value: Optional[bytes] = None self._service: Optional["BluetoothGattService"] = None self._descriptors: List[BluetoothGattDescriptor] = [] def getUuid(self) -> str: return self._uuid def getService(self) -> Optional["BluetoothGattService"]: return self._service def getPermissions(self) -> int: return self._permissions def getProperties(self) -> int: return self._properties def getWriteType(self) -> int: return self._write_type def setWriteType(self, write_type: int): self._write_type = write_type def getValue(self) -> Optional[bytes]: return self._value def setValue(self, value: bytes): self._value = value def getDescriptors(self) -> List[BluetoothGattDescriptor]: return list(self._descriptors) def getDescriptor(self, uuid_str: str) -> Optional[BluetoothGattDescriptor]: for d in self._descriptors: if d.getUuid() == uuid_str: return d return None def addDescriptor(self, descriptor: BluetoothGattDescriptor): self._descriptors.append(descriptor) descriptor._characteristic = self def getIntValue(self, format_type: int, offset: int) -> Optional[int]: if self._value is None: return None import struct try: fmt_map = { self.FORMAT_UINT8: ("B", 1), self.FORMAT_SINT8: ("b", 1), self.FORMAT_UINT16: ("H", 2), self.FORMAT_SINT16: ("h", 2), self.FORMAT_UINT32: ("I", 4), self.FORMAT_SINT32: ("i", 4), self.FORMAT_FLOAT: ("f", 4), self.FORMAT_SFLOAT: ("e", 2), } fmt, size = fmt_map.get(format_type, ("B", 1)) if offset + size > len(self._value): return None val = struct.unpack(fmt, self._value[offset:offset + size])[0] return int(val) except Exception: return None # ═══════════════════════════════════════════════════════════════ # BluetoothGattService # ═══════════════════════════════════════════════════════════════ class BluetoothGattService(Stub): """BluetoothGattService — GATT 服务""" SERVICE_TYPE_PRIMARY = 0 SERVICE_TYPE_SECONDARY = 1 def __init__(self, uuid_str: str, service_type: int = SERVICE_TYPE_PRIMARY): super().__init__() self._uuid = uuid_str self._type = service_type self._characteristics: List[BluetoothGattCharacteristic] = [] self._included_services: List["BluetoothGattService"] = [] self._device: Optional[BluetoothDevice] = None def getUuid(self) -> str: return self._uuid def getType(self) -> int: return self._type def getCharacteristics(self) -> List[BluetoothGattCharacteristic]: return list(self._characteristics) def getCharacteristic(self, uuid_str: str) -> Optional[BluetoothGattCharacteristic]: for c in self._characteristics: if c.getUuid() == uuid_str: return c return None def addCharacteristic(self, characteristic: BluetoothGattCharacteristic): self._characteristics.append(characteristic) characteristic._service = self def getIncludedServices(self) -> List["BluetoothGattService"]: return list(self._included_services) def addIncludedService(self, service: "BluetoothGattService"): self._included_services.append(service) def getDevice(self) -> Optional[BluetoothDevice]: return self._device # ═══════════════════════════════════════════════════════════════ # BluetoothGattCallback # ═══════════════════════════════════════════════════════════════ class BluetoothGattCallback(Stub): """BluetoothGattCallback — GATT 回调""" def onConnectionStateChange(self, gatt: "BluetoothGatt", status: int, new_state: int): pass def onServicesDiscovered(self, gatt: "BluetoothGatt", status: int): pass def onCharacteristicRead(self, gatt: "BluetoothGatt", characteristic: BluetoothGattCharacteristic, value: bytes, status: int): pass def onCharacteristicWrite(self, gatt: "BluetoothGatt", characteristic: BluetoothGattCharacteristic, status: int): pass def onDescriptorRead(self, gatt: "BluetoothGatt", descriptor: BluetoothGattDescriptor, status: int): pass def onDescriptorWrite(self, gatt: "BluetoothGatt", descriptor: BluetoothGattDescriptor, status: int): pass def onCharacteristicChanged(self, gatt: "BluetoothGatt", characteristic: BluetoothGattCharacteristic, value: bytes): pass def onReadRemoteRssy(self, gatt: "BluetoothGatt", rssi: int, status: int): pass def onMtuChanged(self, gatt: "BluetoothGatt", mtu: int, status: int): pass def onPhyUpdate(self, gatt: "BluetoothGatt", tx_phy: int, rx_phy: int, status: int): pass def onPhyRead(self, gatt: "BluetoothGatt", tx_phy: int, rx_phy: int, status: int): pass def onServiceChanged(self, gatt: "BluetoothGatt"): pass def onNotificationSent(self, gatt: "BluetoothGatt", status: int): pass def onConnectionUpdated(self, gatt: "BluetoothGatt", interval: int, latency: int, timeout: int, status: int): pass # ═══════════════════════════════════════════════════════════════ # BluetoothGatt # ═══════════════════════════════════════════════════════════════ class BluetoothGatt(Stub): """BluetoothGatt — GATT 客户端""" CONNECTION_STATE_DISCONNECTED = 0 CONNECTION_STATE_CONNECTING = 1 CONNECTION_STATE_CONNECTED = 2 CONNECTION_STATE_DISCONNECTING = 3 GATT_SUCCESS = 0 GATT_FAILURE = 1 GATT_CONNECTION_CONGESTED = 143 GATT_INSUFFICIENT_AUTHENTICATION = 5 GATT_INSUFFICIENT_ENCRYPTION = 15 GATT_INSUFFICIENT_KEY_SIZE = 12 GATT_INVALID_ATTRIBUTE_LENGTH = 13 GATT_REQUEST_NOT_SUPPORTED = 6 GATT_WRITE_NOT_PERMITTED = 3 GATT_READ_NOT_PERMITTED = 2 GATT_WRITE_NO_RESPONSE = 1 GATT_WRITE_DEFAULT = 0 def __init__(self, device: BluetoothDevice): super().__init__() self._device = device self._services: List[BluetoothGattService] = [] self._connected = False self._mtu = 23 def connect(self) -> bool: self._connected = True return True def disconnect(self): self._connected = False def close(self): self._connected = False self._services.clear() def discoverServices(self) -> bool: return True def getServices(self) -> List[BluetoothGattService]: return list(self._services) def getService(self, uuid_str: str) -> Optional[BluetoothGattService]: for s in self._services: if s.getUuid() == uuid_str: return s return None def getConnectedDevices(self) -> List[BluetoothDevice]: return [self._device] if self._connected else [] def getDevicesMatchingConnectionState(self, state: int) -> List[BluetoothDevice]: if state == self.CONNECTION_STATE_CONNECTED and self._connected: return [self._device] return [] def readCharacteristic(self, characteristic: BluetoothGattCharacteristic) -> bool: return True def writeCharacteristic(self, characteristic: BluetoothGattCharacteristic) -> bool: return True def readDescriptor(self, descriptor: BluetoothGattDescriptor) -> bool: return True def writeDescriptor(self, descriptor: BluetoothGattDescriptor) -> bool: return True def setCharacteristicNotification(self, characteristic: BluetoothGattCharacteristic, enable: bool) -> bool: return True def readRemoteRssy(self) -> bool: return True def requestMtu(self, mtu: int) -> bool: self._mtu = mtu return True def getMtu(self) -> int: return self._mtu def requestConnectionPriority(self, connection_priority: int) -> bool: return True def getDevice(self) -> BluetoothDevice: return self._device # ═══════════════════════════════════════════════════════════════ # BluetoothManager # ═══════════════════════════════════════════════════════════════ class BluetoothManager(Stub): """BluetoothManager — 蓝牙管理器""" def __init__(self, context: Any = None): super().__init__() self._context = context def getAdapter(self) -> Optional["BluetoothAdapter"]: return BluetoothAdapter.getDefaultAdapter() # ═══════════════════════════════════════════════════════════════ # BluetoothAdapter # ═══════════════════════════════════════════════════════════════ class BluetoothAdapter(Stub): """BluetoothAdapter — 蓝牙适配器""" STATE_OFF = 10 STATE_TURNING_ON = 11 STATE_ON = 12 STATE_TURNING_OFF = 13 ACTION_REQUEST_ENABLE = "android.bluetooth.adapter.action.REQUEST_ENABLE" ACTION_REQUEST_DISCOVERABLE = "android.bluetooth.adapter.action.REQUEST_DISCOVERABLE" ACTION_STATE_CHANGED = "android.bluetooth.adapter.action.STATE_CHANGED" ACTION_DISCOVERY_STARTED = "android.bluetooth.adapter.action.DISCOVERY_STARTED" ACTION_DISCOVERY_FINISHED = "android.bluetooth.adapter.action.DISCOVERY_FINISHED" SCAN_MODE_NONE = 20 SCAN_MODE_CONNECTABLE = 21 SCAN_MODE_CONNECTABLE_DISCOVERABLE = 23 _default_adapter: Optional["BluetoothAdapter"] = None def __init__(self): super().__init__() self._enabled = True self._state = self.STATE_ON self._address = "00:1A:2B:3C:4D:5E" self._name = "MockBluetooth" self._discovering = False self._bonded_devices: List[BluetoothDevice] = [] @staticmethod def getDefaultAdapter() -> Optional["BluetoothAdapter"]: if BluetoothAdapter._default_adapter is None: BluetoothAdapter._default_adapter = BluetoothAdapter() return BluetoothAdapter._default_adapter def isEnabled(self) -> bool: return self._enabled def enable(self) -> bool: self._enabled = True self._state = self.STATE_ON return True def disable(self) -> bool: self._enabled = False self._state = self.STATE_OFF return True def getAddress(self) -> str: return self._address def getName(self) -> str: return self._name def setName(self, name: str) -> bool: self._name = name return True def getBondedDevices(self) -> Set[BluetoothDevice]: return set(self._bonded_devices) def startDiscovery(self) -> bool: self._discovering = True return True def cancelDiscovery(self) -> bool: self._discovering = False return True def isDiscovering(self) -> bool: return self._discovering @staticmethod def checkBluetoothAddress(address: str) -> bool: import re return bool(re.match(r"^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$", address)) def getProfileProxy(self, context: Any, listener: BluetoothProfileServiceListener, profile_id: int) -> bool: return True def closeProfileProxy(self, profile_id: int, proxy: Any): pass def getState(self) -> int: return self._state def listenUsingRfcommWithServiceRecord(self, name: str, uuid_str: str) -> BluetoothServerSocket: return BluetoothServerSocket() def listenUsingInsecureRfcommWithServiceRecord(self, name: str, uuid_str: str) -> BluetoothServerSocket: return BluetoothServerSocket() def getRemoteDevice(self, address: str) -> Optional[BluetoothDevice]: if not BluetoothAdapter.checkBluetoothAddress(address): return None return BluetoothDevice(address) def getLeScanner(self) -> Any: from .bluetooth_le import BluetoothLeScanner return BluetoothLeScanner() def toString(self) -> str: return f"BluetoothAdapter(name={self._name})" # ═══════════════════════════════════════════════════════════════ # Additional profile stubs # ═══════════════════════════════════════════════════════════════ class BluetoothHeadset(Stub): pass class BluetoothA2dp(Stub): pass class BluetoothHealth(Stub): pass class BluetoothHealthCallback(Stub): pass class BluetoothHealthAppConfiguration(Stub): pass class BluetoothCodecConfig(Stub): pass class BluetoothLeAudio(Stub): pass class BluetoothLeBroadcast(Stub): pass class BluetoothLeBroadcastMetadata(Stub): pass class BluetoothLeBroadcastSubgroup(Stub): pass # ═══════════════════════════════════════════════════════════════ # __all__ # ═══════════════════════════════════════════════════════════════ __all__ = [ "BluetoothAdapter", "BluetoothDevice", "BluetoothServerSocket", "BluetoothSocket", "BluetoothClass", "BluetoothGatt", "BluetoothGattCallback", "BluetoothGattCharacteristic", "BluetoothGattDescriptor", "BluetoothGattService", "BluetoothHeadset", "BluetoothA2dp", "BluetoothHealth", "BluetoothHealthCallback", "BluetoothHealthAppConfiguration", "BluetoothProfile", "BluetoothProfileServiceListener", "BluetoothManager", "BluetoothCodecConfig", "BluetoothLeAudio", "BluetoothLeBroadcast", "BluetoothLeBroadcastMetadata", "BluetoothLeBroadcastSubgroup", ]