import _bluetooth # 特性标志位 FLAG_BROADCAST = 0x0001 FLAG_READ = 0x0002 FLAG_WRITE_NO_RESPONSE = 0x0004 FLAG_WRITE = 0x0008 FLAG_NOTIFY = 0x0010 FLAG_INDICATE = 0x0020 FLAG_AUTHENTICATED_SIGNED_WRITE = 0x0040 # 描述符标志位 FLAG_DSC_READ = 0x01 FLAG_DSC_WRITE = 0x02 FLAG_DSC_READ_ENC = 0x04 FLAG_DSC_READ_AUTHEN = 0x08 FLAG_DSC_READ_AUTHOR = 0x10 FLAG_DSC_WRITE_ENC = 0x20 FLAG_DSC_WRITE_AUTHEN = 0x40 FLAG_DSC_WRITE_AUTHOR = 0x80 FLAG_AUX_WRITE = 0x0100 FLAG_READ_ENCRYPTED = 0x0200 FLAG_READ_AUTHENTICATED = 0x0400 FLAG_READ_AUTHORIZED = 0x0800 FLAG_WRITE_ENCRYPTED = 0x1000 FLAG_WRITE_AUTHENTICATED = 0x2000 FLAG_WRITE_AUTHORIZED = 0x4000 ''' 创建具有指定值的 UUID 实例 该值可以是 - 一个 16 位整数. 例如`0x2908` - 一个 32 位整数. 例如`0x29081234` - 一个 128 位的 UUID 字符串。例如`'6E400001-B5A3-F393-E0A9-E50E24DCCA9E'`. ''' class UUID(): def __init__(self,value): # try: self.value,self._UUID_bits = self._UUID_to_bytes(value) # except: # print("ValueError: Invalid UUID") # raise ValueError def _UUID_to_bytes(self,value): value_bytes = [] UUID_bits = 0 # try: if isinstance(value,bytearray): value_bytes = bytes(value.decode()) # 暂不支持bytes(bytearray()) if len(value_bytes) > 16 : value_bytes = [] # raise ValueError raise ValueError elif isinstance(value, bytes): value_bytes = value elif isinstance(value,str): # if len(value) == 36 and (value[8] == value[13] == value[18] == value[23] == '-'): #暂不支持连等 if len(value) == 36 and (value[8] == '-') and (value[13] == '-') and (value[18] == '-') and (value[23] == '-'): value_str = value.replace("-","") value_list = [] if len(value_str) == 32: for i in range(0, 32, 2): value_list.append(int(value_str[i:i+2],16)) value_bytes = bytes(value_list) else : raise ValueError else : raise ValueError elif isinstance(value,int): value_list = [] if value >= 0 and value < 65536 : #16 bit uuid value_list.append(int(value/256)) value_list.append(value%256) elif value >= 0 and value < 65536 * 65536: #32 bit uuid value_list.append(int(value/(65536 * 256))) value_list.append(int((value%(65536 * 256))/65536)) value_list.append(int((value%(65536))/256)) value_list.append((value%(256))) else : raise ValueError value_bytes = bytes(value_list) else: raise ValueError UUID_bits = len(value_bytes) # bytes的数量 return value_bytes,UUID_bits class BLE(_bluetooth.BLE): def __init__(self): print("BLE init") # a = super().__init__() self._last_adv_data = "" #广播内容 self._last_resp_data = "" #回应扫描内容 self._addr_mode = 0 #地址类型 BLE_OWN_ADDR_PUBLIC,BLE_OWN_ADDR_RANDOM,BLE_OWN_ADDR_RPA_PUBLIC_DEFAULT,BLE_OWN_ADDR_RPA_RANDOM_DEFAULT self._callback_func = None #回调函数 self._basic_value_handle = 20 self._connectable = True self._py2c_dict = {} self._c2py_dict = {} self._c2value_dict = {} self._append_adv_data = True a = self.init() self.setCallback(self._callback) # 测试函数 def test(self): if self.pyi_check_active() == False: raise OSError def test2(self): self.test() # return self.pyi_test2()s def test3(self,connhandle,valuehandle): return self.pyi_test3(connhandle,valuehandle) def test4(self,data): pass def test_call_some_name(self): super().test_call_some_name() # 检查蓝牙活跃性 def _check_active(self): if self.pyi_check_active() == False: raise OSError ''' 可选择更改 BLE 无线电的活动状态，并返回当前状态。 Args: - active(bool): 为 1 则开启BLE, 否则关闭BLE; 为空则查询当前BLE状态 Returns: - bool:设置状态时，返回操作成功与否； 查询状态时, 返回状态, True为活跃 ''' def active(self,active_flag = None ): if active_flag == None: return self.pyi_check_active() else: if (active_flag > 0 or active_flag == True): return self.pyi_active(True) elif (active_flag == 0 or active_flag == False): return self.pyi_active(False) ''' 获取或设置 BLE 接口的配置值。 要获得一个值，参数名称应该被引用为一个字符串，并且一次只查询一个参数。 如 `config("gap_name")` 要设置值，请使用关键字语法，并且一次可以设置一个或多个参数。 如 `config(gap_name="nimble")` 当前支持的值是: - `'mac'`: TODO:需要支持设置任意值嘛？有点困难 - `'addr_mode'`: - `'gap_name'`: - `'rxbuf'`: TODO:暂未支持, 没找到对应函数 - `'mtu'`: TODO:不明意义 - `'bond'`: TODO: 目前提供的接口不支持 - `'mitm'`: TODO: 目前提供的接口不支持 - `'io'`: TODO:未找到对应函数 - `'lesure'`: TODO:只支持设置 ''' def config(self, *param_name, **kv): # a.config(mac="1123",gap_name="test") # 获取参数属性 a.config("mac") try: self._check_active() except: raise OSError if len(param_name) != 0 : first_param = param_name[0] if first_param == "mac": return (self.config_addr_mode_get(),self.config_mac_get()) elif first_param == "addr_mode": return self.config_addr_mode_get() elif first_param == "gap_name": return self.config_gap_name_get() # elif first_param == "rxbuf" : # self.config_addr_rxbuf_get() # elif first_param == "mtu" : # self.config_mtu_get() # elif first_param == "bond" : # self.config_bond_get() # elif first_param == "mitm" : # self.config_mitm_get() # elif first_param == "io": # self.config_io_get() # elif first_param == "le_secure": # self.config_le_secure_get() elif first_param == "gap_uuid": return self.config_gap_uuid_get() else: print("ValueError: unknown config param") # 设置参数 if "mac" in kv: self.config_mac_update(kv["mac"]) if ("addr_mode" in kv): self.config_addr_mode_update(kv["_addr_mode"]) if ("gap_name" in kv): self.config_gap_name_update(kv["gap_name"]) # if ("rxbuf" in kv): # return self.config_rxbuf_update(kv["rxbuf"]) # if ("mtu" in kv): # return self.config_mtu_update(kv["mtu"]) # if ("bond" in kv): # return self.config_bond_update(kv["bond"]) # if ("mitm" in kv): # return self.config_mitm_update(kv["mitm"]) # if ("bond" in kv): # return self.config_mac_update(kv["bond"]) # if ("io" in kv): # return self.config_io_update(kv["io"]) # if ("le_secire" in kv): # return self.config_le_secire_update(kv["le_secire"]) if ("gap_uuid" in kv): uuid = kv["gap_uuid"] if isinstance(uuid,UUID): self.config_gap_uuid_update(uuid.value,uuid._UUID_bits) ''' 为来自 BLE 堆栈的事件注册回调。 处理程序采用两个参数,event(在下面的代码中的一个) 和 data(其是值的特定事件元组)。 ''' # 回调事件处理函数 def irq(self,func): self._callback_func = func return 0 def _callback(self,data): event_id = data[0] # print("_callback call") # print("event_id",event_id) # print("data ",data[1]) if self._callback_func != None: if event_id > 100 : #自定义回调事件 if event_id == 101 : # 建立句柄映射 ble_value_handles = data[1] length = len(ble_value_handles) for i in range(length): key = self._basic_value_handle + i value = ble_value_handles[i] # py 映射 c handle self._py2c_dict[str(key)] = value # c 映射 py handle self._c2py_dict[str(value)] = key # c handle 映射 value, 默认值为空 self._c2value_dict[str(value)] = bytes("") elif event_id == 102: #nimble蓝牙协议栈读属性 buf = self._c2value_dict[str(data[1])] return len(buf),buf else: if event_id == 3: # write self._c2_change_value(data[2], data[3]) data = data[:3] elif event_id == 4: # read 请求 rc = self._callback_func(event_id,data[1:]) value = -99 length = -99 if rc == 0: #允许读 value = self._c2value(data[2]) length = len(value) return rc,length,value else: return self._callback_func(event_id,data[1:]) ''' 以指定的时间间隔(以微秒为单位)开始广播。 adv_data和resp_data可以是任何实现缓冲协议的类型(例如bytes, bytearray, str)。adv_data包含在所有广播中,并且resp_data被发送以响应主动扫描。 注意:如果adv_data(或resp_data)是None, 则传递给前一个调用的数据 gap_advertise将被重新使用。 要清除广告有效负载, 请传递一个空值bytes, 即b'' Args: - interval_us(int):广播时间间隔, 单位us. 此间隔将向下舍入到最接近的 625us的倍数。要停止广告, 请将interval_us设置 为 None。 - adv_data(bytes, bytearray, str): 自定义广播数据(添加到默认广播数据后),默认值为none TODO:1.append模式只能是0XFF数据; 归零模式全部自定义输入 - resp_data(bytes, bytearray, str): 扫描响应数据,默认值为none - connectable(bool): 是否可连接, True 为可连接. - adv_data_append: 是否是增加数据 TODO:存在问题 ''' def gap_advertise(self, interval_us,adv_data=None,resp_data=None, connectable=None, adv_data_append=None): try: self._check_active() except: raise OSError if connectable != None: self._connectable = connectable if interval_us is None: return self.stop_advertise() else: # 设置广播载荷 if adv_data is None: #参数为空，则使用上次数据广播数据 adv_data = self._last_adv_data else : if adv_data_append != None: # 设置是否增加数据 self._append_adv_data = adv_data_append if self._append_adv_data == True: # 新增数据 self._last_adv_data = _to_bytes(adv_data) else: # 覆盖数据 empty_list = [] for i in range(len(adv_data)): empty_list.append(len(adv_data[i])) empty_list += adv_data[i] # print(empty_list) self._last_adv_data = _to_bytes(empty_list) # 设置响应载荷 if resp_data is None: resp_data = self._last_resp_data else : self._last_resp_data = _to_bytes(resp_data) # self._adv_data_append = False # print("_addr_mode : ",self._addr_mode) # print("int(interval_us/625) : ",int(interval_us/625)) # print("self._connectable: ",self._connectable) # print("self._last_adv_data : ",self._last_adv_data) # print("len(self._last_adv_data) : ",len(self._last_adv_data)) # print("adv_data_append : ",self._append_adv_data) # print("self._b",self._b) # print("self._last_resp_data : ",self._last_resp_data) # print("len(self._last_resp_data) : ",len(self._last_resp_data)) return self.advertise(self._addr_mode,int(interval_us/625),self._connectable,self._last_adv_data, len(self._last_adv_data),self._append_adv_data,self._last_resp_data,len(self._last_resp_data)) ''' 运行持续指定持续时间(以毫秒为单位)的扫描操作。 要无限期扫描, 请将duration_ms设置为0;要停止扫描,请将duration_ms设置为 None. 使用interval_us和window_us可选择配置占空比。扫描器将每interval_us微秒运行window_us 微秒,总共持续duration_ms毫秒。 默认间隔和窗口分别为 1.28 秒和 11.25 毫秒(后台扫描). 对于每个扫描结果, _IRQ_SCAN_RESULT将引发事件, 并带有事件数据。`(addr_type, addr, adv_type, rssi, adv_data)` Args: - duration_ms(int):扫描持续时间。要无限期扫描, 请将duration_ms设置为0;要停止扫描,请将duration_ms设置为 None. - interval_us(int): 扫描间间隔时间。默认值为1280000 - window_us(int): 扫描窗口时间。默认值为11250 - active(bool): 是否接受扫描响应。默认值False(不接受) Return: - 操作成功返回 0; 否则返回错误代码 events: - 单个扫描结果触发 _IRQ_SCAN_RESULT - 扫描结束触发 _IRQ_SCAN_DONE ''' def gap_scan(self, duration_ms, interval_us=1280000, window_us=11250, active=False): try: self._check_active() except: raise OSError self._check_active() if duration_ms is None : return self.gap_stop_disc() else: return self.gap_disc(self._addr_mode, duration_ms,int(interval_us/625),int(window_us/625),active) ''' central 设备连接 peripherals 设备 Args: - peer_addr(list,bytes,bytearray): 需要连接的设备的mac地址 - peer_addr_type(int): 连接设备的地址类型 - 0x00 - PUBLIC - 使用控制器的公共地址。 - 0x01 - RANDOM - 使用生成的静态地址。 - 0x02 - RPA - 使用可解析的私有地址。 - 0x03 - NRPA - 使用不可解析的私有地址。 Return: - (int): 操作成功返回 0; 否则返回错误代码 events: - 连接成功触发 CENTRAL设备:_IRQ_PERIPHERAL_CONNECT; PERIPHERAL设备:_IRQ_CENTRAL_CONNECT ''' def gap_connect(self,peer_addr,peer_addr_type, scan_duration_ms=2000): try: self._check_active() except: raise OSError self._check_active() peer_addr_bytes = _to_bytes(peer_addr,6) return self.pyi_gap_connect(peer_addr_bytes,peer_addr_type ,scan_duration_ms) ''' Args: - conn_handle(int): 连接句柄 Return: - (int): 操作成功返回 0; 否则返回错误代码 events: - 断连成功触发 CENTRAL设备:_IRQ_PERIPHERAL_DISCONNECT; PERIPHERAL设备:_IRQ_CENTRAL_DISCONNECT ''' def gap_disconnect(self, conn_handle): try: self._check_active() except: raise OSError self._check_active() return self.pyi_gap_disconnect(conn_handle) ''' 使用指定的服务配置服务器，替换任何现有服务。 Args: - services(嵌套多层tuple): Return: - (int): 操作成功返回 0; 否则返回错误代码 ''' def gatts_register_services(self, services): try: self._check_active() except: raise OSError self._check_active() convert_services = _convert_ble_service_info(services) offset = 0 chr_list = _count_chrs(services) #计算每个服务的特性数量 all_chr_count = 0 py_handles = [] # 计算py handle返回值 # 计算chr总数量 for i in range(len(chr_list)): py_handles_one_service = [] all_chr_count += chr_list[i] for j in range(chr_list[i]): value_handle = self._basic_value_handle + offset py_handles_one_service.append(value_handle) offset += 1 py_handles.append(py_handles_one_service) rc = self.gatts_register_svcs(convert_services,all_chr_count) if rc != 0 : return rc return tuple(py_handles) ''' Args: - value_handle(int): 蓝牙特性句柄 Return: - (int): 操作成功返回 0; 否则返回错误代码 ''' def gatts_read(self,value_handle): try: self._check_active() except: raise OSError return self._py2value(value_handle) ''' Args: - value_handle(int): 蓝牙特性句柄 - data(str,int,bytes,list(元素为小于256大于0的整数)): 修改的数据 - send_update(bool):是否通知客户端, False为不通知, 默认为False Return: - (int): 操作成功返回 0; 否则返回错误代码 events: - send_update 为 True 时, 触发tx事件 ''' def gatts_write(self,value_handle, data, send_update=False): try: self._check_active() except: raise OSError print("send_update=", send_update) if send_update == False: return self._py2_change_value(value_handle,data) else : rc = self._py2_change_value(value_handle,data) if rc != 0: return rc return self.gatts_chr_updated(self._py2c_dict[str(value_handle)]) ''' 向连接的客户端发送通知请求。 Args: - conn_handle(int): 连接句柄 - value_handle(int): 蓝牙特性句柄 - data(str,int,bytes,list(元素为小于256大于0的整数)): 发送的数据, 默认值为None 如果data不是None, 则该值将作为通知的一部分发送给客户端。本地值不会被修改。否则, 如果data是None, 则将发送当前本地值 Return: - (int): 操作成功返回 0; 否则返回错误代码 events: - ''' def gatts_notify(self,conn_handle, value_handle, data=None): try: self._check_active() except: raise OSError self._check_active() value = "" if data is None: value = self._py2value(value_handle) c_value_handle = self._py2c_dict[str(value_handle)] else : if isinstance(data,bytes): #TODO:关注一下数据类型 value = data elif isinstance(data,int): value = bytes([data]) else: value = bytes(data) c_value_handle = self._py2c_dict[str(value_handle)] self.pyi_gatts_notify(conn_handle, c_value_handle,value,len(value)) ''' Args: - Return: - 操作是否成功 events: - ''' def gatts_indicate(self,conn_handle, value_handle,data=None): try: self._check_active() except: raise OSError value = "" if data is None: value = self._py2value(value_handle) c_value_handle = self._py2c_dict[str(value_handle)] else : if isinstance(data,bytes): value = data elif isinstance(data,int): value = bytes([data]) else: value = bytes(data) c_value_handle = self._py2c_dict[str(value_handle)] self.pyi_gatts_indicate(conn_handle, c_value_handle,value,len(value)) ''' Args: - Return: - 操作是否成功 events: - ''' def gatts_set_buffer(self,value_handle, len, append=False): try: self._check_active() except: raise OSError # TODO:暂不清楚对照哪个函数 pass ''' 打印全部本地服务 ''' def gatts_show_local(self): try: self._check_active() except: raise OSError return self.pyi_gatts_show_local() ''' Args: - Return: - 操作是否成功 events: - ''' def gattc_discover_services(self,conn_handle, uuid:UUID=None): try: self._check_active() except: raise OSError if uuid == None: return self.gattc_dis_svcs(conn_handle) else : return self.gattc_dis_svcs_by_uuid(conn_handle,uuid.value,len(uuid.value)) ''' Args: - Return: - 操作是否成功 events: - ''' def gattc_discover_characteristics(self,conn_handle, start_handle, end_handle, uuid:UUID=None): try: self._check_active() except: raise OSError if uuid == None: return self.gattc_dis_chrs(conn_handle,start_handle,end_handle) else : return self.gattc_dis_chrs_by_uuid(conn_handle, start_handle, end_handle,uuid.value,len(uuid.value)) ''' Args: - Return: - 操作是否成功 events: - ''' def gattc_discover_descriptors(self,conn_handle, start_handle, end_handle): try: self._check_active() except: raise OSError return self.gattc_dis_dscs(conn_handle,start_handle, end_handle) ''' Args: - Return: - 操作是否成功 events: - ''' def gattc_read(self,conn_handle, value_handle): try: self._check_active() except: raise OSError return self.pyi_gattc_read(conn_handle, value_handle) ''' Args: - Return: - 操作是否成功 注意:单次传输最多95个byte events: - ''' def gattc_write(self,conn_handle, value_handle, data, mode = 0): try: self._check_active() except: raise OSError if mode == 0: return self.gattc_write_with_no_rsp(conn_handle, value_handle, data, len(data)) elif mode == 1: return self.gattc_write_with_rsp(conn_handle, value_handle, data,len(data)) ''' Args: - Return: - 操作是否成功 events: - ''' def gattc_exchange_mtu(self,conn_handle): try: self._check_active() except: raise OSError return self.pyi_gattc_exchange_mtu(conn_handle) def gattc_subscribe(self,conn_handle,value_handle,subscribe = 0): try: self._check_active() except: raise OSError return self.pyi_gattc_subscribe(conn_handle,value_handle,subscribe) # 通过C handle找值 def _c2value(self, handle): return self._c2value_dict[str(handle)] # 通过PY handle找值 def _py2value(self,handle): c_handlue = self._py2c_dict[str(handle)] return self._c2value(c_handlue) # return self._c2value(25) # 通过C handle 改值, value均为bytes类型 def _c2_change_value(self,handle,value): value_bytes = 12 # self._c2value_dict[str(handle)] = value_bytes self._c2value_dict[str(handle)] = value return 0 # 通过py handle 改值 def _py2_change_value(self,handle,value): if len(self._py2c_dict) == 0: return -1 c_handlue = self._py2c_dict[str(handle)] value_bytes = _to_bytes(value) return self._c2_change_value(c_handlue,value_bytes) # 将UUID类型转换为字符串 def _convert_ble_service_info(data): new_list = [] for i in data : if isinstance(i,UUID) : new_list.append(i.value) new_list.append(i._UUID_bits) # print(i.value) elif isinstance(i, tuple): new_list.append(_convert_ble_service_info(i)) else: new_list.append(i) # print(i) return tuple(new_list) def _count_chrs(srvs_tuple): srv_count = len(srvs_tuple) chr_count = [] # 每个服务的特性数量 for i in range(srv_count): if len(srvs_tuple[i]) > 1 : chr_count.append(len(srvs_tuple[i][1])) else : chr_count.append(0) return chr_count def _to_bytes(data,size=None): data_bytes = [] if isinstance(data,bytes): data_bytes = data elif isinstance(data,bytearray): data_bytes = bytes(data.decode()) elif isinstance(data,str): data_bytes = bytes(data) elif isinstance(data,list): data_bytes = bytes(data) elif isinstance(data,int): data_list = [] if data == 0: data_list.append(0) else : while data > 0: remainder = data % 256 data_list.insert(0, remainder) data //= 256 data_bytes = bytes(data_list) else: raise ValueError if size != None and len(data_bytes) != size: raise ValueError return data_bytes