import struct as _struct import traceback as _trace from datetime import datetime as _date from random import randint from time import sleep, time as _time from biopro.impl import LowLevelHardwareInterface from biopro.util.address import str_address, address_bytes from biopro.util.console import hex_table as hx, SEND, YELLOW, LINE_BREAK from biopro.util.logger import LoggerFlag from .event import * E = TypeVar('E', bound='Event') HCI_PACKET_COMMAND = 0x01 HCI_PACKET_ASYNC_DATA = 0x02 HCI_PACKET_SYNC_DATA = 0x03 HCI_PACKET_EVENT = 0x04 # noinspection PyUnusedLocal class CC2650(LoggerFlag): __slots__ = ('_interface', '_info') def __init__(self, interface: LowLevelHardwareInterface): super().__init__('CC2650') self._interface = interface self._info = pp('CC2650', 33, '[') def _send(self, f: str, *args): data = _struct.pack(f, *args) if self.is_log_enable(self.LEVEL_VERBOSE): hx(data, SEND) print(LINE_BREAK) self._interface.send_byte(data) self._interface.flush() sleep(0.01) def send_command(self, opcode: int, f: str = '', *args: Union[int, bytes]): if len(f) == 0: data = _struct.pack(' Union[None, Event]: packet = self._recv_byte() if packet is None: return None if packet == HCI_PACKET_EVENT: return self._recv_event(timeout) else: return None def _recv_byte(self) -> Optional[int]: ret = self._recv_bytes(1) return ret[0] if ret is not None else None def _recv_bytes(self, size: int = 1) -> Union[None, bytes]: start = _time() while True: ret = self._interface.recv_byte(size) if ret is not None and len(ret) > 0: return ret elif _time() - start > 1: return None else: sleep(0.01) @overload def recv_util(self, event: Type[E], when: RecvCond = None, timeout: int = 1) -> E: pass @overload def recv_util(self, event: Union[None, EVENT_ID] = None, when: RecvCond = None, timeout: int = 1) -> Optional[Event]: pass def recv_util(self, event: Union[None, EVENT_ID] = None, when: RecvCond = None, timeout: int = 1) -> Optional[Event]: """receive event data until specific event got. ================= ========================================== event handle mean ================= ========================================== None pass event (Event) -> None function call with event as argument RecvReturn recv_until return Optional[event] UnexpectedEvent raise UnexpectedEvent KeyboardInterrupt raise KeyboardInterrupt ================= ========================================== :param event: expected event, None and ``when`` is None, return any non-None event. None and ``when`` is not None, do not return except Error raising. Ellipsis (``...``) for receive event forever :param timeout: if timeout, raise RecvTimeout. but when *event* is ``...``, return None. :param when: un-expected event handle :return: received event. """ if isinstance(event, int) and event <= 0: raise RuntimeError('illegal event code : ' + x16(event)) start = _time() while True: receive_event = self.recv(timeout) if receive_event is None: if _time() - start > timeout: if event == Ellipsis: return None else: ### device timeout raise RecvTimeout() else: start = _time() if event is None and when is None: # return directly return receive_event elif event == Ellipsis: pass elif event is not None and RecvCond.match_event(receive_event, event): # return expected return receive_event try: # handle other case if when is not None and when.invoke(receive_event): # event consumed, next loop continue # default handle general case elif isinstance(receive_event, Evt_CommandStatus): try: receive_event.ensure_event_success() except RuntimeError as e: self.log_warn(e) raise elif isinstance(receive_event, HciEvt_CommandComplete): # XXX Is it necessary check the status of the HciEvt_CommandComplete pass elif isinstance(receive_event, Evt_Unknown): pass else: # fail all raise UnexpectedEvent(receive_event) except RecvReturn as e: if e.event is not None: e.event.consume() return e.event else: receive_event.consume() return receive_event def recv_collect(self, event: Type[E], when: RecvCond = None, timeout: int = 10) -> List[E]: """receive event data until specific event produce completed. :param event: expected event :param timeout: if timeout, raise RecvTimeout. :param when: un-expected event handle :return: collected event """ back: List[E] = [] def _recv_collect_inner(e: Event): status = e.event_status if status == ErrorCode.SUCCESS: back.append(e) e.consume() elif status == ErrorCode.ProcedureComplete: raise RecvReturn() elif status == ErrorCode.Timeout: raise RecvTimeout() else: err = ErrorCode.error_str(status, None) raise RuntimeError(e.event_name + ' with status ' + x8(status) + ' ' + pp(err, None)) cond = when.cover({event: _recv_collect_inner}) self.recv_util(..., when=cond, timeout=timeout) return back def _recv_event(self, timeout: Optional[float] = None) -> Optional[Event]: code = self._recv_byte() if code is None: return None length = self._recv_byte() if code is None: return None if self.is_log_enable(self.LEVEL_VERBOSE): print(pc('Rx', RECV), x8(code), pp({ 0xFF: 'Vendor Specific Event', 0x05: 'BT Disconnection Complete', 0x08: 'BT Encryption Change', 0x0C: 'BT Read Remote Version Information Complete', 0x0E: 'BT Command Complete', 0x0F: 'BT Command Status', 0x10: 'BT Hardware Error', 0x13: 'BT Number Of Completed Packets', 0x1A: 'BT Data Buffer Overflow', 0x30: 'BT Encryption Key Refresh Complete', 0x3E: 'LE event', }.get(code, None))) print(pc('time', RECV), ':', _date.now()) _start = _time() data = b'' while len(data) < length: ret = self._recv_bytes(length - len(data)) if ret is not None: data += ret elif timeout is not None and _time() - _start > timeout: raise RecvTimeout() evt = None try: if code == 0xFF: evt = unpack_vendor_specific_event(data) elif code == 0x05: self.log_warn(pp('FIXME', ']'), 'BT Disconnection Complete') elif code == 0x08: self.log_warn(pp('FIXME', ']'), 'BT Encryption Change') elif code == 0x0C: self.log_warn(pp('FIXME', ']'), 'BT Read Remote Version Information Complete') elif code == 0x0E: evt = HciEvt_CommandComplete.unpack(data) elif code == 0x0F: self.log_warn(pp('FIXME', ']'), 'BT Command Status') elif code == 0x10: self.log_warn(pp('FIXME', ']'), 'BT Hardware Error (optional)') elif code == 0x13: self.log_warn(pp('FIXME', ']'), 'BT Number Of Completed Packets') elif code == 0x1A: self.log_warn(pp('FIXME', ']'), 'BT Data Buffer Overflow') elif code == 0x30: self.log_warn(pp('FIXME', ']'), 'BT Encryption Key Refresh Complete') elif code == 0x3E: self.log_warn(pp('FIXME', ']'), 'LE event') except KeyboardInterrupt: raise except: evt = Evt_InternalError.unpack(data) if evt is None: if code == 0xFF: evt = Evt_Unknown(_struct.unpack(' bool: if self.is_log_enable(self.LEVEL_VERBOSE): print(pc('Tx', SEND), x16(f.EVENT_CODE), pp(f.__name__)) print(pc('time', SEND), ':', _date.now()) return True elif self.is_log_enable(self.LEVEL_INFO): self.log_info(pc(f.__name__, YELLOW), *msg) return False @instruction(0x0406) def hci_disconnect(self, handle: HANDLE, reason: Optional[HciDisconnectReason] = None): """This BT API is used to terminate a connection. Related Events: HCI_CommandStatusEvent , DisconnectEvent :param handle: Connection handle. :param reason: Reason for disconnection """ self._check_connection_handler(handle, 'hci_disconnect') if reason is None: reason = HciDisconnectReason.REMOTE_USER_TERM_CONN elif not HciDisconnectReason.valid_parameter(reason): raise RuntimeError('illegal hci_disconnect reason : ' + x8(reason)) t = HciDisconnectReason.reason_str(reason) if self._log_command(CC2650.hci_disconnect, x16(handle), t): print(pc('handle', SEND), ':', x16(handle)) print(pc('reason', SEND), ':', x8(reason), pp(t)) self.send_command(0x0406, 'HB', handle, reason) @instruction(0x041D) def hci_read_remote_version_info(self, handle: HANDLE): """This BT API is used to request version information from the remote device in a connection. Related Events: HCI_CommandStatusEvent, ReadRemoteVersionInfoEvent :param handle: Connection handle. """ self._check_connection_handler(handle, 'hci_read_remote_version_info') if self._log_command(CC2650.hci_read_remote_version_info, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) self.send_command(0x041D, 'H', handle) @instruction(0x0C03) def hci_reset(self): """This BT API is used to reset the Link Layer. Related Events: HCI_CommandCompleteEvent """ self._log_command(CC2650.hci_reset) self.send_command(0x0C03) @instruction(0X1001) def hci_read_local_version_info(self): """This BT API is used to read the local version information. """ self._log_command(CC2650.hci_read_local_version_info) self.send_command(0x1001) @instruction(0x1002) def hci_read_local_supported_commands(self): """This BT API is used to read the locally supported commands. """ self._log_command(CC2650.hci_read_local_supported_commands) self.send_command(0x1002) @instruction(0x1003) def hci_read_local_supported_features(self): """This BT API is used to read the locally supported features. """ self._log_command(CC2650.hci_read_local_supported_features) self.send_command(0x1003) # noinspection PyPep8Naming @instruction(0x1009) def hci_read_BDADDR(self): """This BT API is used to read this device's BLE address (BDADDR). """ self._log_command(CC2650.hci_read_BDADDR) self.send_command(0x1009) @instruction(0x1405) def hci_read_rssi(self, handle: HANDLE): """ This BT API is used to read the RSSI of the last packet received on a connection given by the connection handle. If the Receiver Modem test is running (HCI_EXT_ModemTestRx), then the RF RSSI for the last received data will be returned. If there is no RSSI value, then HCI_RSSI_NOT_AVAILABLE will be returned. :param handle: Connection handle. """ if self._log_command(CC2650.hci_read_rssi, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) self.send_command(0x1405) @instruction(0x2003) def hci_le_read_local_supported_features(self): """This LE API is used to read the LE locally supported features. """ self._log_command(CC2650.hci_le_read_local_supported_features) self.send_command(0x2003) @instruction(0x200E) def hci_le_create_connection_cancel(self): """This LE API is used to cancel a create connection. """ self._log_command(CC2650.hci_le_create_connection_cancel) self.send_command(0x200E) @instruction(0x200F) def hci_le_read_white_list_size(self): """This LE API is used to read the white list. """ self._log_command(CC2650.hci_le_read_white_list_size) self.send_command(0x200F) @instruction(0x2010) def hci_le_clear_white_list(self): """This LE API is used to clear the white list. """ self._log_command(CC2650.hci_le_clear_white_list) self.send_command(0x2010) @instruction(0x2011) def hci_le_add_white_list(self, address: Union[str, ADDRESS], address_type: AddressType): """This LE API is used to add a white list entry. :param address_type: :param address: address of device to put in white list. """ if AddressType.type_str(address_type) is None: raise RuntimeError('illegal hci_le_add_white_list address_type : ' + x8(address_type)) if isinstance(address, str): a = address_bytes(str_address(address)) else: a = address_bytes(address) if self._log_command(CC2650.hci_le_add_white_list, hl(a, ':')): print(pc('address', SEND), ':', hl(a, ':')) print(pc('address_type', SEND), ':', x8(address_type), pp(AddressType.type_str(address_type, None))) self.send_command(0x2011, 'B6s', address_type, a) @instruction(0x2012) def hci_le_remove_white_list(self, address: ADDRESS, address_type: AddressType): """This LE API is used to remove a white list entry. :param address_type: :param address: address of device to remove from the white list. """ if AddressType.type_str(address_type) is None: raise RuntimeError('illegal hci_le_add_white_list address_type : ' + x8(address_type)) if isinstance(address, str): a = address_bytes(str_address(address)) else: a = address_bytes(address) if self._log_command(CC2650.hci_le_remove_white_list, hl(a, ':')): print(pc('address', SEND), ':', hl(a, ':')) print(pc('address_type', SEND), ':', x8(address_type), pp(AddressType.type_str(address_type, None))) self.send_command(0x2012, 'B6s', address_type, a) @instruction(0x2013) def hci_le_connection_update(self, handle: HANDLE, link_parameter: LinkParameter): """This LE API is used to update the connection parameters. :param handle: Time between Init scan events. :param link_parameter: link parameter. """ if self._log_command(CC2650.hci_le_connection_update, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) link_parameter.dump(SEND) self.send_command(0x2013, '7H', handle, link_parameter.interval_min, link_parameter.interval_max, link_parameter.latency, link_parameter.timeout, link_parameter.info_min, link_parameter.info_max) @instruction(0xFC00) def hci_ext_set_rx_gain(self, gain: bool): """This HCI Extension API is used to set the receiver gain. ===== ===== ==================== gain value mean ===== ===== ==================== False 0x00 HCI_EXT_RX_GAIN_STD True 0x01 HCI_EXT_RX_GAIN_HIGH ===== ===== ==================== :param gain """ t = 'HCI_EXT_RX_GAIN_HIGH' if gain else 'HCI_EXT_RX_GAIN_STD' if self._log_command(CC2650.hci_ext_set_rx_gain, t): print(pc('gain', SEND), ':', gain, pp(t)) self.send_command(0xFC00, 'B', u8(gain)) @instruction(0xFC01) def hci_ext_set_tx_power(self, power: uint8): """This HCI Extension API is used to set the transmit power. ===== ============================ power mean ===== ============================ 0x00 LL_EXT_TX_POWER_MINUS_23_DBM 0x01 LL_EXT_TX_POWER_MINUS_6_DBM 0x02 LL_EXT_TX_POWER_0_DBM 0x03 LL_EXT_TX_POWER_4_DBM ===== ============================ :param power: """ if power not in (0, 1, 2, 3): raise RuntimeError('illegal hci_ext_set_tx_power value : ' + x8(power)) if self._log_command(CC2650.hci_ext_set_tx_power, power): print(pc('power', SEND), ':', x8(power), pp({ 0x00: 'TX_POWER_MINUS_23_DBM', 0x01: 'TX_POWER_MINUS_6_DBM', 0x02: 'TX_POWER_0_DBM', 0x03: 'TX_POWER_4_DBM', }[power])) self.send_command(0xFC01, 'B', power) @instruction(0xFC02) def hci_ext_one_packet_per_event(self, control: bool): """This HCI Extension API is used to set whether a connection will be limited to one packet per event. ======= ===== =============================== control value mean ======= ===== =============================== False 0x00 HCI_EXT_DISABLE_ONE_PKT_PER_EVT True 0x01 HCI_EXT_ENABLE_ONE_PKT_PER_EVT ======= ===== =============================== :param control: """ t = 'ENABLE_ONE_PKT_PER_EVT' if control else 'DISABLE_ONE_PKT_PER_EVT' if self._log_command(CC2650.hci_ext_one_packet_per_event, t): print(pc('control', SEND), ':', control, pp(t)) self.send_command(0xFC02, 'B', u8(control)) @instruction(0xFC03) def hci_ext_clock_divided_on_halt(self, control: bool): """This HCI Extension API is used to set whether the system clock will be divided when the MCU is halted. ======= ===== ================================== control value mean ======= ===== ================================== False 0x00 HCI_EXT_DISABLE_CLK_DIVIDE_ON_HALT True 0x01 HCI_EXT_ENABLE_CLK_DIVIDE_ON_HALT ======= ===== ================================== :param control: """ t = 'ENABLE_CLK_DIVIDE_ON_HALT' if control else 'DISABLE_CLK_DIVIDE_ON_HALT' if self._log_command(CC2650.hci_ext_clock_divided_on_halt, t): print(pc('control', SEND), ':', control, pp(t)) self.send_command(0xFC03, 'B', u8(control)) @instruction(0xFC04) def hci_ext_declare_nv_usage(self, mode: bool): """This HCI Extension API is used to indicate to the Controller whether or not the Host will be using the NV memory during BLE operations. ======= ===== ================================== mode value mean ======= ===== ================================== False 0x00 HCI_EXT_NV_NOT_IN_USE True 0x01 HCI_EXT_NV_IN_USE ======= ===== ================================== :param mode: """ t = 'NV_IN_USE' if mode else 'NV_NOT_IN_USE' if self._log_command(CC2650.hci_ext_declare_nv_usage, t): print(pc('control', SEND), ':', mode, pp(t)) self.send_command(0xFC04, 'B', u8(mode)) @instruction(0xFC05) def hci_ext_decrypt(self, key: bytes, text: bytes): """This HCI Extension API is used to decrypt encrypted data using AES128. :param key: 16 byte encryption key. :param text: 16 byte encrypted data. """ if len(key) != 16: raise RuntimeError('illegal hci_ext_decrypt key length : ' + str(len(key))) if len(text) != 16: raise RuntimeError('illegal hci_ext_decrypt text length : ' + str(len(text))) if self._log_command(CC2650.hci_ext_decrypt): print(pc('key ', SEND), ':', hl(key)) print(pc('text', SEND), ':', hl(text)) self.send_command(0xFC05, '16s16s', key, text) @instruction(0xFC06) def hci_ext_set_local_supported_features(self, features: uint64): """This HCI Extension API is used to write this device's supported features. :param features: Pointer to eight bytes of local features. """ if self._log_command(CC2650.hci_ext_set_local_supported_features, x64(features)): print(pc('features', SEND), ':', x64(features)) self.send_command(0xFC06, '2L', (features & 0xFFFFFFFF), ((features >> 32) & 0xFFFFFFFF)) @instruction(0xFC07) def hci_ext_set_fast_tx_response_time(self, control: bool): """This HCI Extension API is used to set whether transmit data is sent as soon as possible even when slave latency is used. ======= ===== ================================== control value mean ======= ===== ================================== False 0x00 HCI_EXT_DISABLE_FAST_TX_RESP_TIME True 0x01 HCI_EXT_ENABLE_FAST_TX_RESP_TIME ======= ===== ================================== :param control: """ t = 'ENABLE_FAST_TX_RESP_TIME' if control else 'DISABLE_FAST_TX_RESP_TIME' if self._log_command(CC2650.hci_ext_set_fast_tx_response_time, t): print(pc('control', SEND), ':', control, pp(t)) self.send_command(0xFC07, 'B', u8(control)) @instruction(0xFC08) def hci_ext_modem_test_tx(self, mode: bool, freq: uint8): """This API is used start a continuous transmitter modem test, using either a modulated or un-modulated carrier wave tone, at the frequency that corresponds to the specified RF channel. Use HCI_EXT_EndModemTest command to end the test. Note: A Controller reset will be issued by HCI_EXT_EndModemTest! Note: The BLE device will transmit at maximum power. Note: This API can be used to verify this device meets Japan's TELEC regulations. ======= ===== ================================== mode value mean ======= ===== ================================== False 0x00 HCI_EXT_TX_MODULATED_CARRIER True 0x01 HCI_EXT_TX_UNMODULATED_CARRIER ======= ===== ================================== ===== ================ freq BLE freq ===== ================ 0..39 2402 + 2k MHz ===== ================ :param mode: :param freq: RF channel of transmit frequency. """ if not 0 <= freq <= 39: raise RuntimeError('illegal hci_ext_modem_test_tx freq : ' + x8(freq)) t = 'TX_UNMODULATED_CARRIER' if mode else 'TX_MODULATED_CARRIER' f = '%d MHz' % (freq * 2 + 2402) if self._log_command(CC2650.hci_ext_modem_test_tx, t, f): print(pc('control', SEND), ':', mode, pp(t)) print(pc('freq', SEND), ':', x8(freq), pp(f)) self.send_command(0xFC08, 'BB', u8(mode), freq) @instruction(0xFC09) def hci_ext_modem_hop_test_tx(self): """This API is used to start a continuous transmitter direct test mode test using a modulated carrier wave and transmitting a 37 byte packet of Pseudo-Random 9-bit data. A packet is transmitted on a different frequency (linearly stepping through all RF channels 0..39) every 625us. Use HCI_EXT_EndModemTest command to end the test. Note: A Controller reset will be issued by HCI_EXT_EndModemTest! Note: The BLE device will transmit at maximum power. Note: This API can be used to verify this device meets Japan's TELEC regulations. """ self._log_command(CC2650.hci_ext_modem_hop_test_tx) self.send_command(0xFC09) @instruction(0xFC0A) def hci_ext_modem_test_rx(self, freq: uint8): """This API is used to start a continuous receiver modem test using a modulated carrier wave tone, at the frequency that corresponds to the specific RF channel. Any received data is discarded. Receiver gain may be adjusted using the HCI_EXT_SetRxGain command. RSSI may be read during this test by using the HCI_ReadRssi command. Use HCI_EXT_EndModemTest command to end the test. Note: A Controller reset will be issued by HCI_EXT_EndModemTest! Note: The BLE device will transmit at maximum power. ===== ================ freq BLE freq ===== ================ 0..39 2402 + 2k MHz ===== ================ :param freq: Receiver RF channel """ if not 0 <= freq <= 39: raise RuntimeError('illegal hci_ext_modem_test_rx freq : ' + x8(freq)) f = '%d MHz' % (freq * 2 + 2402) if self._log_command(CC2650.hci_ext_modem_test_rx, f): print(pc('freq', SEND), ':', x8(freq), pp(f)) self.send_command(0xFC0A, 'B', freq) @instruction(0xFC0B) def hci_ext_end_modem_test(self): """This API is used to shutdown a modem test. A complete Controller reset will take place. """ self._log_command(CC2650.hci_ext_end_modem_test) self.send_command(0xFC0B) # noinspection PyPep8Naming @instruction(0xFC0C) def hci_ext_set_BDADDR(self, address: Union[None, str, ADDRESS]): """This API is used to set this device's BLE address (BDADDR). Note: This command is only allowed when the device's state is Standby. ================================ ============================== address mean ================================ ============================== 0x000000000000 .. 0xFFFFFFFFFFFE Valid BLE device address. 0xFFFFFFFFFFFF or None Invalid BLE device address.[1] ================================ ============================== [1] An invalid address (i.e. all FF's) will restore this device's address to the address set at initialization. :param address: device's address. """ if address is None: if self._log_command(CC2650.hci_ext_set_BDADDR): print(pc('address', SEND), ':', 'FF:FF:FF:FF:FF:FF') self.send_command(0xFC0C, '6B', 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF) else: if isinstance(address, str): a = address_bytes(str_address(address)) else: a = address_bytes(address) if self._log_command(CC2650.hci_ext_set_BDADDR, hl(a, ':')): print(pc('address', SEND), ':', hl(a, ':')) self.send_command(0xFC0C, '6s', a) @instruction(0xFC0D) def hci_ext_set_sca(self, value: uint16): """This API is used to set this device's Sleep Clock Accuracy. Note: For a slave device, this value is directly used, but only if power management is enabled. For a master device, this value is converted one: into of eight ordinal values representing a SCA range, as specified in Table 2.2, Vol. 6, Part B, Section 2.3.3.1 of the Core specification. Note: This command is only allowed when the device is not in a connection. Note: The device's SCA value remains unaffected by a HCI_Reset. ================ ================== value mean ================ ================== 0 .. 0x1F4(=500) Valid SCA value. 0x1F5 .. 0xFFFF Invalid SCA value. ================ ================== :param value: A SCA value in PPM. """ if not 0x00 <= value <= 0x1F4: raise RuntimeError('illegal hci_ext_set_sca value : ' + x16(value)) if self._log_command(CC2650.hci_ext_set_sca, x16(value)): print(pc('value', SEND), ':', x16(value)) self.send_command(0xFC0D, 'H', value) @instruction(0xFC0E) def hci_ext_enable_ptm(self): """This HCI Extension API is used to enable Production Test Mode. Note: This function can only be directly called from the application and is not available via an external transport interface such as RS232. Also, no vendor specific command complete will be returned. """ self._log_command(CC2650.hci_ext_enable_ptm) self.send_command(0xFC0E) @instruction(0xFC0F) def hci_ext_set_freq_tune(self, step: bool): """This HCI Extension API is used to set the frequency tuning up or down. Setting the mode up/down decreases/increases the amount of capacitance on the external crystal oscillator. Note: This is a Production Test Mode only command! ===== ===== ========================== step value mean ===== ===== ========================== False 0x00 HCI_PTM_SET_FREQ_TUNE_DOWN True 0x01 HCI_PTM_SET_FREQ_TUNE_UP ===== ===== ========================== :param step: """ t = 'HCI_PTM_SET_FREQ_TUNE_UP' if step else 'HCI_PTM_SET_FREQ_TUNE_DOWN' if self._log_command(CC2650.hci_ext_set_freq_tune, t): print(pc('value', SEND), ':', step, pp(t)) self.send_command(0xFC0F, 'B', u8(step)) @instruction(0xFC10) def hci_ext_save_freq_tune(self): """This HCI Extension API is used to save the frequency tuning value to flash. """ self._log_command(CC2650.hci_ext_save_freq_tune) self.send_command(0xFC10) @instruction(0xFC11) def hci_ext_set_max_dtm_tx_power(self, power: uint8): """This HCI Extension API is used to set the maximum transmit output power for Direct Test Mode. ===== ==================================== power mean ===== ==================================== 0x00 HCI_EXT_TX_POWER_MINUS_23_DBM 0x01 HCI_EXT_TX_POWER_MINUS_6_DBM 0x02 HCI_EXT_TX_POWER_0_DBM 0x03 HCI_EXT_TX_POWER_4_DBM (CC2540 only) ===== ==================================== :param power: """ if power not in (0, 1, 2): raise RuntimeError('illegal hci_ext_set_max_dtm_tx_power power : ' + x8(power)) if self._log_command(CC2650.hci_ext_set_max_dtm_tx_power): print(pc('power', SEND), ':', x8(power), pp({ 0x00: 'TX_POWER_MINUS_23_DBM', 0x01: 'TX_POWER_MINUS_6_DBM', 0x02: 'TX_POWER_0_DBM', 0x03: 'TX_POWER_4_DBM', }[power])) self.send_command(0xFC11, 'B', power) @instruction(0xFC12) def hci_ext_map_pm_io_port(self, port: uint8, pin: uint8): """This HCI Extension API is used to configure and map a CC254x I/O Port as a General Purpose I/O (GPIO) output signal that reflects the Power Management (PM) state of the CC254x device. The GPIO output will be High on Wake, and Low upon entering Sleep. This feature can be disabled by specifying HCI_EXT_PM_IO_PORT_NONE for the ioPort (ioPin is then ignored). The system default value upon hardware reset is disabled. This command can be used to control an external DC-DC Converter (its actual intent) such has the TI TPS62730 (or any similar converter that works the same way). This command should be used with extreme care as it will override how the Port/Pin was previously configured! This includes the mapping of Port 0 pins to 32kHz clock output, Analog I/O, UART, Timers; Port 1 pins to Observables, Digital Regulator status, UART, Timers; Port 2 pins to an external 32kHz XOSC. The selected Port/Pin will be configured as an output GPIO with interrupts masked. Careless use can result in a reconfiguration that could disrupt the system. It is therefore the user's responsibility to ensure the selected Port/Pin does not cause any conflicts in the system. Note: Only Pins 0, 3 and 4 are valid for Port 2 since Pins 1 and 2 are mapped to debugger signals DD and DC. Note: Port/Pin signal change will only occur when Power Savings is enabled. ===== ==================================== port mean ===== ==================================== 0x00 HCI_EXT_PM_IO_PORT_P0 0x01 HCI_EXT_PM_IO_PORT_P1 0x02 HCI_EXT_PM_IO_PORT_P2 0xFF HCI_EXT_PM_IO_PORT_NONE ===== ==================================== ===== ==================================== pin mean ===== ==================================== 0x00 HCI_EXT_PM_IO_PORT_PIN0 0x01 HCI_EXT_PM_IO_PORT_PIN1 0x02 HCI_EXT_PM_IO_PORT_PIN2 0x03 HCI_EXT_PM_IO_PORT_PIN3 0x04 HCI_EXT_PM_IO_PORT_PIN4 0x05 HCI_EXT_PM_IO_PORT_PIN5 0x06 HCI_EXT_PM_IO_PORT_PIN6 0x07 HCI_EXT_PM_IO_PORT_PIN7 ===== ==================================== :param port: :param pin: """ if port not in (0, 1, 2, 0xFF): raise RuntimeError('illegal hci_ext_map_pm_io_port port : ' + x8(port)) if not 0x00 <= pin <= 0x07: raise RuntimeError('illegal hci_ext_map_pm_io_port pin : ' + x8(pin)) if self._log_command(CC2650.hci_ext_map_pm_io_port): print(pc('port', SEND), ':', x8(port), pp({ 0x00: 'PM_IO_PORT_P0', 0x01: 'PM_IO_PORT_P2', 0x02: 'PM_IO_PORT_P3', 0xFF: 'PM_IO_PORT_NONE', }[port])) print(pc('pin', SEND), ':', x8(pin), pp({ 0x00: 'PM_IO_PORT_PIN0', 0x01: 'PM_IO_PORT_PIN1', 0x02: 'PM_IO_PORT_PIN2', 0x03: 'PM_IO_PORT_PIN3', 0x04: 'PM_IO_PORT_PIN4', 0x05: 'PM_IO_PORT_PIN5', 0x06: 'PM_IO_PORT_PIN6', 0x07: 'PM_IO_PORT_PIN7', }[pin])) self.send_command(0xFC12, 'BB', port, pin) @instruction(0xFC13) def hci_ext_disconnect_immediate(self, handle: HANDLE): """This HCI Extension API is used to disconnect the connection immediately. Note: The connection (if valid) is immediately terminated without notifying the remote device. The Host is still notified. ================ ======================= handle mean ================ ======================= 0x0000 .. 0x0EFF valid connection handle 0x0F00 .. 0x0FFF reserved for future use ================ ======================= :param handle: Connection handle. """ self._check_connection_handler(handle, 'hci_ext_disconnect_immediate') if self._log_command(CC2650.hci_ext_disconnect_immediate, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) self.send_command(0xFC13, 'H', handle) @instruction(0xFC14) def hci_ext_packet_error_rate(self, handle: HANDLE, command: bool = False): """This function is used to Reset or Read the Packet Error Rate counters for a connection. Note: The counters are only 16 bits. At the shortest connection interval, this provides a bit over 8 minutes of data. ================ ======================= handle mean ================ ======================= 0x0000 .. 0x0EFF valid connection handle 0x0F00 .. 0x0FFF reserved for future use ================ ======================= ======= ================= command mean ======= ================= False HCI_EXT_PER_RESET True HCI_EXT_PER_READ ======= ================= :param handle: The LL connection ID on which to send this data. :param command: """ if self._log_command(CC2650.hci_ext_packet_error_rate, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('command', SEND), ':', command, pp('PER_READ' if command else 'PER_RESET')) self.send_command(0xFC14, 'HB', handle, u8(command)) @instruction(0xFC16) def hci_ext_extend_rf_range(self): """This HCI Extension API is used to Extend Rf Range using the TI CC2590 2.4 GHz RF Front End device. """ self._log_command(CC2650.hci_ext_extend_rf_range) self.send_command(0xFC16) @instruction(0xFC17) def hci_ext_advertising_event_notice(self, task: uint8, event: uint16): """This HCI Extension API is used to enable or disable a notification to the specified task using the specified task event whenever a Adv event ends. A non-zero taskEvent value is taken to be "enable", while a zero valued taskEvent is taken to be "disable". :param task: User's task ID. :param event: User's task event. """ if not 0 <= task <= 0xFF: raise RuntimeError('illegal hci_ext_advertiser_event_notice task ID value : ' + x8(task)) if not 0 <= event <= 0x3FFF: raise RuntimeError('illegal hci_ext_advertiser_event_notice task event value : ' + x16(event)) if self._log_command(CC2650.hci_ext_advertising_event_notice): print(pc('task', SEND), ':', x8(task)) print(pc('event', SEND), ':', x16(event)) self.send_command(0xFC17, 'BH', task, event) @instruction(0xFC19) def hci_ext_halt_during_rf(self, mode: bool): """This HCI Extension API is used to enable or disable halting the CPU during RF. The system defaults to enabled. ===== ===== ============================== mode value mean ===== ===== ============================== False 0x00 HCI_EXT_HALT_DURING_RF_DISABLE True 0x01 HCI_EXT_HALT_DURING_RF_ENABLE ===== ===== ============================== :param mode: """ t = 'HCI_EXT_HALT_DURING_RF_ENABLE' if mode else 'HCI_EXT_HALT_DURING_RF_DISABLE' if self._log_command(CC2650.hci_ext_halt_during_rf, t): print(pc('mode', SEND), ':', mode, pp(t)) self.send_command(0xFC19, 'B', u8(mode)) @instruction(0xFC1A) def hci_ext_set_slave_latency_override(self, control: bool): """This HCI Extension API is used to to enable or disable suspending slave latency. ======= ===== =========================== control value mean ======= ===== =========================== False 0x00 HCI_EXT_DISABLE_SL_OVERRIDE True 0x01 HCI_EXT_ENABLE_SL_OVERRIDE ======= ===== =========================== :param control: """ t = 'ENABLE_SL_OVERRIDE' if control else 'HCI_EXT_DISABLE_SL_OVERRIDE' if self._log_command(CC2650.hci_ext_set_slave_latency_override, t): print(pc('control', SEND), ':', control, pp(t)) self.send_command(0xFC1A, 'B', u8(control)) @instruction(0xFC1B) def hci_ext_build_revision(self, mode: bool, reversion: uint16): """This HCI Extension API is used set a user revision number or read the build revision number. ======= ===== =========================== mode value mean ======= ===== =========================== False 0x00 HCI_EXT_SET_APP_REVISION True 0x01 HCI_EXT_READ_BUILD_REVISION ======= ===== =========================== :param mode: :param reversion: Any value the application wishes to use as their revision number. """ t = 'READ_BUILD_REVISION' if mode else 'SET_APP_REVISION' if self._log_command(CC2650.hci_ext_build_revision, t): print(pc('mode', SEND), ':', mode, pp(t)) self.send_command(0xFC1B, 'BH', u8(mode), reversion) @instruction(0xFC1C) def hci_ext_delay_sleep(self, delay: uint16): """This HCI Extension API is used set the sleep delay. :param delay: 0 .. 0x03E8 (=1000) in milliseconds. """ if not 0 <= delay <= 0x03E8: raise RuntimeError('illegal hci_ext_delay_sleep delay value : ' + x16(delay)) t = '%d ms' % delay if self._log_command(CC2650.hci_ext_delay_sleep, t): print(pc('delay', SEND), ':', x16(delay), pp(t)) self.send_command(0xFC1C, 'H', delay) @instruction(0xFC1D) def hci_ext_reset_system(self, mode: bool): """This HCI Extension API is used to issue a soft or hard system reset. ======= ===== =========================== mode value mean ======= ===== =========================== False 0x00 HCI_EXT_RESET_SYSTEM_HARD True 0x01 HCI_EXT_RESET_SYSTEM_SOFT ======= ===== =========================== :param mode: """ t = 'RESET_SYSTEM_SOFT' if mode else 'RESET_SYSTEM_HARD' if self._log_command(CC2650.hci_ext_reset_system, t): print(pc('mode', SEND), ':', mode, pp(t)) self.send_command(0xFC1D, 'B', u8(mode)) @instruction(0xFC1E) def hci_ext_overlapped_processing(self, mode: bool): """This HCI Extension API is used to enable or disable overlapped processing. ======= ===== ===================================== mode value mean ======= ===== ===================================== False 0x00 HCI_EXT_DISABLE_OVERLAPPED_PROCESSING True 0x01 HCI_EXT_ENABLE_OVERLAPPED_PROCESSING ======= ===== ===================================== :param mode: """ t = 'ENABLE_OVERLAPPED_PROCESSING' if mode else 'DISABLE_OVERLAPPED_PROCESSING' if self._log_command(CC2650.hci_ext_overlapped_processing, t): print(pc('mode', SEND), ':', mode, pp(t)) self.send_command(0xFC1E, 'B', u8(mode)) @instruction(0xFC1F) def hci_ext_number_complete_packets_limit(self, limit: uint8, mode: bool): """This HCI Extension API is used to set the minimum number of completed packets which must be met before a Number of Completed Packets event is returned. If the limit is not reach by the end of the connection event, then a Number of Completed Packets event will be returned (if non-zero) based on the flushOnEvt flag. ====== ===== ============================================ mode value mean ====== ===== ============================================ False 0x00 HCI_EXT_DISABLE_NUM_COMPL_PKTS_ON_EVENT [1]_ True 0x01 HCI_EXT_ENABLE_NUM_COMPL_PKTS_ON_EVENT [2]_ ====== ===== ============================================ .. [1] Only return a Number of Completed Packets event when the number of completed packets is greater than or equal to the limit. .. [2] Return a Number of Complete Packets event if the number of completed packets is less than the limit. :param limit: From 1 to max_data_buffer which return from hci_le_read_buffer_size() :param mode: flush on event. """ if self._log_command(CC2650.hci_ext_number_complete_packets_limit): print(pc('limit', SEND), ':', x8(limit)) print(pc('mode', SEND), ':', mode, pp('ENABLE' if mode else 'DISABLE')) self.send_command(0xFC1F, 'BB', limit, u8(mode)) @instruction(0xFE00) def gap_device_init(self, profile_role: uint8 = 0x08, max_scan_responses: uint8 = 5, identify_root_key: Optional[IRK] = None, sign_resolving_key: Optional[SRK] = None, sign_counter: uint32 = 1): r"""Called to setup the device. Call just once on initialization. ===================== ================ response status ===================== ================ GapEvt_DeviceInitDone SUCCESS \ INVALIDPARAMETER ===================== ================ :param profile_role: GAP Profile Roles :param max_scan_responses: maximum number to scan responses we can receive during a device discovery. :param identify_root_key: pointer to Identity Root Key, ``None`` (all zeroes) if the app wants the GAP to generate the key. :param sign_resolving_key: pointer to Sign Resolving Key, ``None`` if the app wants the GAP to generate the key. :param sign_counter: 32 bit value used in the SM Signing algorithm that shall be initialized to zero and incremented with every new signing. This variable must also be maintained by the application. """ if len(ProfileRole.mask_str(profile_role)) == 0: raise RuntimeError('illegal gap_device_init profile_role : ' + x8(profile_role)) if identify_root_key is None: irk = _struct.pack('16x') else: irk = _struct.pack('16B', *identify_root_key) if sign_resolving_key is None: srk = _struct.pack('16x') else: srk = _struct.pack('16B', *sign_resolving_key) if self._log_command(CC2650.gap_device_init): print(pc('profile_role', SEND), ':', x8(profile_role), pp(ProfileRole.mask_str(profile_role))) print(pc('max_scan_responses', SEND), ':', max_scan_responses) print(pc('IRK', SEND), ':', hl(irk, ':')) print(pc('SRK', SEND), ':', hl(srk, ':')) print(pc('sign_counter', SEND), ':', x32(sign_counter), pp(sign_counter)) self.send_command(0xFE00, '2B16s16sL', profile_role, max_scan_responses, irk, srk, sign_counter) @instruction(0xFE03) def gap_config_device_address(self, address_type: AddressType, address: Union[None, str, ADDRESS] = None): """Setup the device's address type. If AddressType.PRIVATE_NON_RESOLVE is selected, the address will change periodically. If return value isn't SUCCESS, the address type remains the same as before this call. :param address_type: :param address: Only used with AddressType.STATIC or AddressType.PRIVATE_NON_RESOLVE type. None to auto generate otherwise the application can specify the address value """ if AddressType.type_str(address_type) is None: raise RuntimeError('illegal gap_config_device_address address_type : ' + x8(address_type)) if address_type in (AddressType.STATIC, AddressType.PRIVATE_NON_RESOLVE): if address is None: address: ADDRESS = tuple(randint(0, 0xFF) for _ in range(6)) a = address_bytes(address) elif isinstance(address, str): a = address_bytes(str_address(address)) else: a = address_bytes(address) else: a = _struct.pack('6x') if self._log_command(CC2650.gap_config_device_address, hl(a, ':')): print(pc('address_type', SEND), ':', x8(address_type), pp(AddressType.type_str(address_type, None))) print(pc('address', SEND), ':', hl(a, ':')) self.send_command(0xFE03, 'B6s', address_type, a) @instruction(0xFE04) def gap_device_discovery(self, mode: uint8 = 3, active_scan: bool = True, white_list: bool = False): r"""Start a device discovery scan. ==== ============================ ============================ mode value mean ==== ============================ ============================ 0x00 DEVDISC_MODE_NONDISCOVERABLE No discoverable setting 0x01 DEVDISC_MODE_GENERAL General Discoverable devices 0x02 DEVDISC_MODE_LIMITED Limited Discoverable devices 0x03 DEVDISC_MODE_ALL Not filtered ==== ============================ ============================ =========== ===== ======================== active_scan value mean =========== ===== ======================== False 0x00 Turn off active scanning True 0x01 Turn on active scanning =========== ===== ======================== =========== ===== ====================================== white_list value mean =========== ===== ====================================== False 0x00 Don’t use the white list during a scan True 0x01 Use the white list during a scan =========== ===== ====================================== ======================== ====================== ========================= response status mean ======================== ====================== ========================= Evt_CommandStatus SUCCESS, \ AlreadyInRequestedMode Scan is not available. \ IncorrectMode Invalid profile role. GapEvt_DeviceInformation SUCCESS found device GapEvt_DeviceDiscovery SUCCESS complete ======================== ====================== ========================= :param mode: :param active_scan: :param white_list: """ if mode not in (0, 1, 2, 3): raise RuntimeError('illegal gap_device_discovery mode : ' + x8(mode)) if self._log_command(CC2650.gap_device_discovery): print(pc('mode', SEND), ':', x8(mode), pp({ 0x00: 'NON_DISCOVERABLE', 0x01: 'GENERAL', 0x02: 'LIMITED', 0x03: 'ALL', }[mode])) print(pc('active_scan', SEND), ':', active_scan) print(pc('white_list', SEND), ':', white_list) self.send_command(0xFE04, '3B', mode, u8(active_scan), u8(white_list)) @instruction(0xFE05) def gap_device_discovery_cancel(self): """Cancel an existing device discovery request. """ self._log_command(CC2650.gap_device_discovery_cancel) self.send_command(0xFE05) @instruction(0xFE06) def gap_make_discoverable(self, init_address: Union[str, ADDRESS], init_address_type: AddressType, event_type: uint8, chanel_map: uint8, filter_policy: uint8): r"""Setup or change advertising. Also starts advertising. =========== =========== channel_map mean =========== =========== 0 Channel 37 1 Channel 38 2 Channel 39 3 .. 7 reserved =========== =========== ============= ======================== ========================== filter_policy Allow scan requests From Allow connect request from ============= ======================== ========================== 0x00 any any 0x01 white list only any 0x02 any white list only 0x03 white list only white list only 0x04 .. 0xFF reserved ============= ======================== ========================== ================= ====================================================== response status ================= ====================================================== Evt_CommandStatus SUCCESS \ NotReady (Advertising data isn’t setup.) \ AlreadyInRequestedMode (Not available at this time.) \ IncorrectMode (Invalid profile role.) ================= ====================================================== :param event_type: :param init_address_type: :param init_address: :param chanel_map: :param filter_policy: """ if event_type not in (0, 1, 2, 3): raise RuntimeError('illegal gap_make_discoverable event_type : ' + x8(event_type)) if AddressType.type_str(init_address_type) is None: raise RuntimeError('illegal gap_make_discoverable init_address_type : ' + x8(init_address_type)) if chanel_map not in (0, 1, 2): raise RuntimeError('illegal gap_make_discoverable chanel_map : ' + x8(chanel_map)) if filter_policy not in (0, 1, 2, 3): raise RuntimeError('illegal gap_make_discoverable filter_policy : ' + x8(filter_policy)) if isinstance(init_address, str): a = address_bytes(str_address(init_address)) else: a = address_bytes(init_address) if self._log_command(CC2650.gap_make_discoverable, hl(a, ':')): print(pc('event_type', SEND), ':', x8(event_type), pp(EVENT_TYPE.get(event_type, None))) print(pc('init_address_type', SEND), ':', x8(init_address_type), pp(AddressType.type_str(init_address_type, None))) print(pc('address', SEND), ':', hl(a, ':')) print(pc('channel_map', SEND), ':', x8(chanel_map), pp({ 0x00: 'Channel 37', 0x01: 'Channel 38', 0x02: 'Channel 39', }[chanel_map])) print(pc('filter_policy', SEND), ':', x8(filter_policy), pp(','.join([ 'scan:' + ('white list only' if (filter_policy & 0b01) else 'any'), 'connect:' + ('white list only' if (filter_policy & 0b10) else 'any'), ]))) self.send_command(0xFE06, '2B6s2B', event_type, init_address_type, a, chanel_map, filter_policy) @instruction(0xFE07) def gap_update_advertising_data(self, ad_type: bool, data: Union[bytes, List[AdvertisingToken]]): r"""Setup or change advertising and scan response data. NOTE: if the return status from this function is SUCCESS, the task isn't complete until the GAP_ADV_DATA_UPDATE_DONE_EVENT is sent to the calling application task. ======= ===== ================== ad_type value mean ======= ===== ================== False 0x00 SCAN_RSP data True 0x01 Advertisement data ======= ===== ================== ================= ============= ======================== response status mean ================= ============= ======================== Evt_CommandStatus SUCCESS \ IncorrectMode Invalid profile role. ================= ============= ======================== :param ad_type: :param data: advertising or scan response data """ if isinstance(data, list): token_list = data data = b'' for token in token_list: data += _struct.pack('2B%ds' % len(token.ad_data), len(token.ad_data) + 1, token.ad_type, token.ad_data) length = len(data) if length > 31: raise RuntimeError('gap_update_advertising_data data too long : ' + str(length)) if self._log_command(CC2650.gap_update_advertising_data): print(pc('ad_type', SEND), ':', ad_type, pp('Advertisement data' if ad_type else 'SCAN_RSP data')) print(pc('length', SEND), ':', x8(length), pp(length)) print(pc('data', SEND), ':', hl(data, truncate=True)) self.send_command(0xFE07, '2B%ds' % length, u8(ad_type), length, data) @instruction(0xFE08) def gap_end_discoverable(self): r"""Stops advertising. ================= ============= ===================== response status mean ================= ============= ===================== Evt_CommandStatus SUCCESS \ IncorrectMode Not advertising ================= ============= ===================== """ self._log_command(CC2650.gap_end_discoverable) self.send_command(0xFE08) @instruction(0xFE09) def gap_establish_link(self, address: Union[str, ADDRESS, DeviceInfo], address_type: AddressType = AddressType.PUBLIC, high_duty_cycle: bool = False, white_list: bool = False): r"""Establish a link to a slave device. =============== ===== ========== high_duty_cycle value mean =============== ===== ========== False 0x00 disabled True 0x01 enabled =============== ===== ========== =============== ===== =========================================== white_list value mean =============== ===== =========================================== False 0x00 Don’t use the white list True 0x01 Only connect to a device in the white list =============== ===== =========================================== ====================== ============= ==================================================== response status mean ====================== ============= ==================================================== Evt_CommandStatus SUCCESS \ NotReady Not ready to perform this action. Performing a scan. \ IncorrectMode Invalid profile role. GapEvt_LinkEstablished SUCCESS complete ====================== ============= ==================================================== :param address: Advertiser's address :param address_type: Address type of the advertiser :param high_duty_cycle: :param white_list: :return:SUCCESS: started establish link process, bleIncorrectMode: invalid profile role, bleNotReady: a scan is in progress, bleAlreadyInRequestedMode: can’t process now, bleNoResources: Too many links """ if isinstance(address, DeviceInfo): if address_type < 0: address_type = address.address_type address = address.address elif isinstance(address, str): address = str_address(address) if AddressType.type_str(address_type) is None: raise RuntimeError('illegal gap_establish_link address_type : ' + x8(address_type)) if self._log_command(CC2650.gap_establish_link, address_str(address)): print(pc('high_duty_cycle', SEND), ':', high_duty_cycle) print(pc('white_list', SEND), ':', white_list) print(pc('address_type', SEND), ':', x8(address_type), pp(AddressType.type_str(address_type, None))) print(pc('address', SEND), ':', address_str(address)) self.send_command(0xFE09, '3B6s', u8(high_duty_cycle), u8(white_list), address_type, address_bytes(address)) @instruction(0xFE0A) def gap_terminate_link(self, handle: HANDLE, reason: Optional[HciDisconnectReason] = None): r"""Terminate a link connection. ================ ======================================= handle mean ================ ======================================= 0x0000 .. 0xFFFD Existing connection handle to terminate 0xFFFE Terminate the 'Establish Link Request' 0xFFFF Terminate all links ================ ======================================= ====================== ============= ===================== response status mean ====================== ============= ===================== Evt_CommandStatus SUCCESS \ IncorrectMode No link to terminate GapEvt_LinkTerminated SUCCESS complete ====================== ============= ===================== :param handle: :param reason: terminate reason. """ if reason is None: reason = HciDisconnectReason.REMOTE_USER_TERM_CONN elif HciDisconnectReason.reason_str(reason) is None: raise RuntimeError('illegal gap_terminate_link reason : ' + x8(reason)) t = HciDisconnectReason.reason_str(reason, None) if self._log_command(CC2650.gap_terminate_link, x16(handle), t): print(pc('handle', SEND), ':', x16(handle)) print(pc('reason', SEND), ':', x8(reason), pp(t)) self.send_command(0xFE0A, 'HB', handle, reason) @instruction(0xFE0B) def gap_authenticate(self, handle: HANDLE, a: GAP_Authentication, p: Optional[GAP_Pair] = None): r"""Start the Authentication process with the requested device. This function is used to Initiate/Allow pairing. Called by both master and slave device (Central and Peripheral). NOTE: This function is called after the link is established. ============================= ====================== ========================== response status mean ============================= ====================== ========================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER Invalid Parameter \ AlreadyInRequestedMode Already in this mode \ IncorrectMode Incorrect Profile Role \ NotConnected Invalid connection handle GapEvt_AuthenticationComplete SUCCESS complete ============================= ====================== ========================== :param handle: connection handle :param a: Authentication parameters :param p: Enter these parameters if the Pairing Request was already received. None, if waiting for Pairing Request or if initiating. """ # XXX data struct not correct if not 0 <= handle <= 0xFFFD: raise RuntimeError('illegal gap_authenticate handle : ' + x16(handle)) if a.io_caps not in (0, 1, 2, 3, 4): raise RuntimeError('illegal gap_authenticate GAP_Authentication.io_caps : ' + x8(a.io_caps)) if a.auth_request not in (0, 2): raise RuntimeError('illegal gap_authenticate GAP_Authentication.auth_request : ' + x8(a.auth_request)) if not 7 <= a.max_enc_key_size <= 16: raise RuntimeError('illegal gap_authenticate GAP_Authentication.max_enc_key_size : ' + str(a.max_enc_key_size)) if a.key_dist not in (0, 1, 2, 3, 4, 5): raise RuntimeError('illegal gap_authenticate GAP_Authentication.key_dist : ' + x8(a.key_dist)) if self._log_command(CC2650.gap_authenticate, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) a.dump(SEND) if p is not None: p.dump(SEND) if p is not None: if p.io_caps not in (0, 1, 2, 3, 4): raise RuntimeError('illegal gap_authenticate GAP_Pair.io_caps : ' + x8(p.io_caps)) if p.auth_request not in (0, 2): raise RuntimeError('illegal gap_authenticate GAP_Pair.auth_request : ' + x8(p.auth_request)) if not 7 <= p.max_enc_key_size <= 16: raise RuntimeError('illegal gap_authenticate GAP_Pair.max_enc_key_size : ' + str(p.max_enc_key_size)) if p.key_dist not in (0, 1, 2, 3, 4, 5): raise RuntimeError('illegal gap_authenticate GAP_Pair.key_dist : ' + x8(p.key_dist)) self.send_command(0xFE0B, 'H2B16s9B', handle, a.io_caps, u8(a.oob_available), _struct.pack('16B', a.oob), a.auth_request, a.max_enc_key_size, a.key_dist, 0x01, p.io_caps, u8(p.oob_data_flag), p.auth_request, p.max_enc_key_size, p.key_dist) else: self.send_command(0xFE0B, 'H2B16sB8x', handle, a.io_caps, u8(a.oob_available), _struct.pack('16B', a.oob), a.auth_request, a.max_enc_key_size, a.key_dist, 0x00) @instruction(0xFE11) def gap_update_link(self, handle: HANDLE, link_parameter: LinkParameter): r"""Update the link parameters to a slave device. ======================= ================= ======== response status mean ======================= ================= ======== Evt_CommandStatus SUCCESS \ FAILURE \ INVALID_PARAMETER \ IncorrectMode \ NotConnected GapEvt_LinkParamUpdate SUCCESS complete ======================= ================= ======== :param handle: Connection handle of the update :param link_parameter: link parameter. """ if not 0x00 <= handle <= 0xFFFD: raise RuntimeError('illegal gap_update_link handle : ' + x16(handle)) link_parameter.valid('gap_update_link', contain_info_length=False) if self._log_command(CC2650.gap_update_link, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) link_parameter.dump(SEND, contain_info_length=False) self.send_command(0xFE11, '5H', handle, link_parameter.interval_min, link_parameter.interval_max, link_parameter.latency, link_parameter.timeout) @instruction(0xFE0C) def gap_passkey_update(self, handle: HANDLE, passkey: str): r"""Update the passkey in string format. This function is called by the application/profile in response to receiving the GapEvt_PasskeyNeeded event. =================== ============= =============================================== response status mean =================== ============= =============================================== Evt_CommandStatus SUCCESS \ INVALID_TASK Passkey is NULL, or isn’t formatted properly \ IncorrectMode Link not found =================== ============= =============================================== :param handle: connection handle. :param passkey: new passkey - pointer to numeric string (ie. "019655" ). This string's range is "000000" to "999999". """ if not 0x00 <= handle <= 0xFFFD: raise RuntimeError('illegal gap_passkey_update handle : ' + x16(handle)) if len(passkey) != 6: raise RuntimeError('illegal gap_passkey_update passkey length : ' + passkey) for c in passkey: if c not in '0123456789': raise RuntimeError('illegal gap_passkey_update passkey : ' + passkey) p = passkey.encode('ascii') if self._log_command(CC2650.gap_passkey_update, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('passkey', SEND), ':', hl(p), pp(passkey)) self.send_command(0xFE0C, 'H6s', handle, p) @instruction(0xFE0D) def gap_slave_security(self, handle: HANDLE, auth_request: uint8): r"""Generate a Slave Requested Security message to the master. =================== ====================== ============================================= response status mean =================== ====================== ============================================= Evt_CommandStatus SUCCESS \ AlreadyInRequestedMode wrong GAP role, must be a Peripheral Role \ NotConnected Link not found =================== ====================== ============================================= :param handle: connection handle. :param auth_request: """ if not 0x00 <= handle <= 0xFFFD: raise RuntimeError('illegal gap_slave_security handle : ' + x16(handle)) if auth_request not in AUTHENTICATION_REQUEST: raise RuntimeError('illegal gap_slave_security auth_request : ' + x8(auth_request)) if self._log_command(CC2650.gap_slave_security, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('auth_request', SEND), ':', x8(auth_request), pp(AUTHENTICATION_REQUEST.get(auth_request, None))) self.send_command(0xFE0D, 'HB', handle, auth_request) @instruction(0xFE0E) def gap_signable(self, handle: HANDLE, authenticated: bool, sign_resolving_key: SRK, sign_counter: uint32): r"""Set up the connection to accept signed data. NOTE: This function is called after the link is established. ============= ===== ========================= authenticated value mean ============= ===== ========================= False 0x00 CSRK is not authenticated True 0x01 CSRK is authenticated ============= ===== ========================= =================== ============= ========================================== response status mean =================== ============= ========================================== Evt_CommandStatus SUCCESS \ IncorrectMode wrong GAP role, must be a Peripheral Role \ NotConnected Link not found =================== ============= ========================================== :param handle: connection handle of the signing information :param authenticated: :param sign_resolving_key: CSRK of connected device. :param sign_counter: """ if not 0x00 <= handle <= 0xFFFD: raise RuntimeError('illegal gap_signable handle : ' + x16(handle)) if self._log_command(CC2650.gap_signable, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('authenticated', SEND), ':', authenticated) print(pc('SRK', SEND), ':', address_str(sign_resolving_key)) print(pc('sign_counter', SEND), ':', '0x%8X' % sign_counter, pp(sign_counter)) self.send_command(0xFE0E, 'HB16sL', handle, u8(authenticated), _struct.pack('16B', sign_resolving_key), sign_counter) @instruction(0xFE0F) def gap_bond(self, handle: HANDLE, authenticated: bool, long_term_key: LTK, div: uint16, rand: uint64, key_size: uint8): r"""Set up the connection's bound parameters. NOTE: This function is called after the link is established. ============= ===== ========================= authenticated value mean ============= ===== ========================= False 0x00 CSRK is not authenticated True 0x01 CSRK is authenticated ============= ===== ========================= ======== ====================================== key_size mean ======== ====================================== 7 .. 16 Maximum encryption key size to support ======== ====================================== =================== ============= ===================================================== response status mean =================== ============= ===================================================== Evt_CommandStatus SUCCESS \ IncorrectMode wrong GAP role, must be a Peripheral Role \ NotConnected Link not found GapEvt_BondComplete SUCCESS complete =================== ============= ===================================================== :param handle: connection handle of the signing information :param authenticated: :param long_term_key: :param div: The DIV used with this long_term_key. :param rand: The 8 byte random number generated for this LTK. :param key_size: Encryption key size """ if not 0x00 <= handle <= 0xFFFD: raise RuntimeError('illegal gap_bond handle : ' + x16(handle)) if not 7 <= key_size <= 16: raise RuntimeError('illegal gap_bond key_size : ' + x8(key_size)) if self._log_command(CC2650.gap_bond, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('authenticated', SEND), ':', authenticated) print(pc('LTK', SEND), ':', address_str(long_term_key)) print(pc('div', SEND), ':', x16(div)) print(pc('rand', SEND), ':', x64(rand)) print(pc('key_size', SEND), ':', x8(key_size), pp(key_size)) self.send_command(0xFE0F, 'HB16sH2LB', handle, u8(authenticated), _struct.pack('16B', long_term_key), div, (rand & 0xFFFFFFFF), ((rand >> 32) & 0xFFFFFFFF), key_size) @instruction(0xFE10) def gap_terminate_auth(self, handle: HANDLE, reason: uint8): r"""Send a Pairing Failed message and end any existing pairing. =================== ============= =========================================== response status mean =================== ============= =========================================== Evt_CommandStatus SUCCESS \ IncorrectMode wrong GAP role, must be a Peripheral Role \ NotConnected Link not found GapEvt_BondComplete SUCCESS complete =================== ============= =========================================== :param handle: connection handle. :param reason: Pairing Failed reason code. """ if not 0x00 <= handle <= 0xFFFD: raise RuntimeError('illegal gap_terminate_auth handle : ' + x16(handle)) if self._log_command(CC2650.gap_terminate_auth, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('reason', SEND), ':', x8(reason)) self.send_command(0xFE10, 'HB', handle, reason) @instruction(0xFE30) def gap_set_parameter(self, parameter: GapParameters, value: uint16): r"""Set a GAP Parameter value. Use this function to change the default GAP parameter values. =================== ================= response status =================== ================= Evt_CommandStatus SUCCESS \ INVALID_PARAMETER =================== ================= :param parameter: parameter ID :param value: new param value """ if GapParameters.parameter_str(parameter) is None: raise RuntimeError('illegal gap_get_parameter parameter : ' + x8(parameter)) t = GapParameters.parameter_str(parameter, None) if self._log_command(CC2650.gap_set_parameter, t, x16(value)): print(pc('parameter', SEND), ':', x8(parameter), pp(t)) print(pc('value', SEND), ':', x16(value)) self.send_command(0xFE30, 'BH', parameter, value) @instruction(0xFE31) def gap_get_parameter(self, parameter: GapParameters): """Get a GAP Parameter value. =================== ======= ================ response status =================== ======= ================ Evt_CommandStatus SUCCESS with 2 byte data =================== ======= ================ :param parameter: parameter ID """ if GapParameters.parameter_str(parameter) is None: raise RuntimeError('illegal gap_get_parameter parameter : ' + x8(parameter)) t = GapParameters.parameter_str(parameter, None) if self._log_command(CC2650.gap_get_parameter, t): print(pc('parameter', SEND), ':', x8(parameter), pp(t)) self.send_command(0xFE31, 'B', parameter) @instruction(0xFE32) def gap_resolve_private_address(self, identify_root_key: IRK, address: Union[str, ADDRESS]): r"""Resolves a private address against an IRK. =================== ======= response status =================== ======= Evt_CommandStatus SUCCESS \ FAILURE =================== ======= :param identify_root_key: IRK :param address: the Resolvable Private address """ if isinstance(address, str): a = address_bytes(str_address(address)) else: a = address_bytes(address) if self._log_command(CC2650.gap_resolve_private_address): print(pc('IRK', SEND), ':', address_str(identify_root_key)) print(pc('address', SEND), ':', hl(a, ':')) self.send_command(0xFE32, '16s6s', _struct.pack('16B', identify_root_key), a) @instruction(0xFE33) def gap_set_advertising_token(self, token: AdvertisingToken): r"""Called to setup a GAP Advertisement/Scan Response data token. =================== ================= ================================= response status mean =================== ================= ================================= Evt_CommandStatus SUCCESS \ INVALID_PARAMETER Invalid Advertisement Type \ INVALID_MEM_SIZE Token too large to fit into \ IncorrectMode Invalid GAP Profile Role \ InvalidRange Advertisement Type already exists =================== ================= ================================= :param token: Advertisement/Scan response token. """ if token.ad_type not in GapAdvertisingDataType: raise RuntimeError('illegal gap_set_advertising_token ad_type : ' + x8(token.ad_type)) if self._log_command(CC2650.gap_set_advertising_token): token.dump(SEND) self.send_command(0xFE33, 'BB%ds' % len(token.ad_data), token.ad_type, len(token.ad_data), token.ad_data) @instruction(0xFE34) def gap_remove_advertising_token(self, ad_type: Union[GapAdvertisingDataType, AdvertisingToken]): r"""Called to remove a GAP Advertisement/Scan Response data token. =================== ================= ================================= response status mean =================== ================= ================================= Evt_CommandStatus SUCCESS \ INVALID_PARAMETER ad_type not found. =================== ================= ================================= :param ad_type: Advertisement type to remove """ if isinstance(ad_type, AdvertisingToken): ad_type = ad_type.ad_type if ad_type not in GapAdvertisingDataType: raise RuntimeError('illegal gap_remove_advertising_token ad_type : ' + x8(ad_type)) if self._log_command(CC2650.gap_remove_advertising_token): print(pc('ad_type', SEND), ':', x8(ad_type), pp(GapAdvertisingDataType.type_str(ad_type, None))) self.send_command(0xFE34, 'B', ad_type) @instruction(0xFE35) def gap_update_advertising_tokens(self): """Called to rebuild and load Advertisement and Scan Response data from existing GAP Advertisement Tokens. =================== ======= response status =================== ======= Evt_CommandStatus SUCCESS =================== ======= """ self._log_command(CC2650.gap_update_advertising_tokens) self.send_command(0xFE35) @instruction(0xFE36) def gap_bound_set_parameter(self, parameter: uint16, value: bytes): r"""Send this command to set a GAP Bond parameter. =================== ================= ==================== response status mean =================== ================= ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER Invalid paramID \ InvalidRange Invalid paramDataLen =================== ================= ==================== :param parameter: :param value: """ if parameter not in GapBondParameters: raise RuntimeError('illegal gap_bound_set_parameter parameter : ' + x16(parameter)) t = GapBondParameters.parameter_str(parameter, None) if self._log_command(CC2650.gap_bound_set_parameter, t): print(pc('parameter', SEND), ':', x16(parameter), pp(t)) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFE36, 'HB%ds' % len(value), parameter, len(value), value) @instruction(0xFE37) def gap_bound_get_parameter(self, parameter: uint16): r"""Send this command to read a GAP Bond parameter. =================== ================= ==================== response status mean =================== ================= ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER Invalid paramID \ InvalidRange Invalid paramDataLen =================== ================= ==================== :param parameter: """ if parameter not in GapBondParameters: raise RuntimeError('illegal gap_bound_set_parameter parameter : ' + x16(parameter)) t = GapBondParameters.parameter_str(parameter, None) if self._log_command(CC2650.gap_bound_get_parameter, t): print(pc('parameter', SEND), ':', x16(parameter), pp(t)) self.send_command(0xFE37, 'H', parameter) @instruction(0xFD02) def att_exchange_mtu(self, handle: HANDLE, client_rx_mtu: uint16): r"""Send Exchange MTU Request. =================== ========================== response status =================== ========================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ Pending \ LinkEncrypted \ InvalidPDU \ InsufficientAuthentication \ InsufficientEncrypt \ InsufficientKeySize AttEvt_ExchangeMTU SUCCESS =================== ========================== :param handle: connection to use :param client_rx_mtu: pointer to request to be sent """ if self._log_command(CC2650.att_exchange_mtu, x16(handle), client_rx_mtu): print(pc('handle', SEND), ':', x16(handle)) print(pc('client_rx_mtu', SEND), ':', x16(client_rx_mtu), pp(client_rx_mtu)) self.send_command(0xFD02, 'HH', handle, client_rx_mtu) @instruction(0xFD82) def gatt_exchange_mtu(self, handle: HANDLE, client_rx_mtu: uint16): r"""This sub-procedure is used by the client to set the ATT_MTU to the maximum possible value that can be supported by both devices when the client supports a value greater than the default ATT_MTU for the Attribute Protocol. This sub-procedure shall only be initiated once during a connection. The ATT Exchange MTU Request is used by this sub-procedure. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_ExchangeMTU SUCCESS \ Timeout Att_Error SUCCESS =================== ==================== :param handle: connection to use :param client_rx_mtu: pointer to request to be sent """ if self._log_command(CC2650.gatt_exchange_mtu, x16(handle), client_rx_mtu): print(pc('handle', SEND), ':', x16(handle)) print(pc('client_rx_mtu', SEND), ':', x16(client_rx_mtu), pp(client_rx_mtu)) self.send_command(0xFD82, 'HH', handle, client_rx_mtu) @instruction(0xFD04) def att_find_information(self, handle: HANDLE, start_handle: HANDLE = 0x0001, end_handle: HANDLE = 0xFFFF): r"""Send Find Information Request. ====================== ========================== response status ====================== ========================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ Pending \ LinkEncrypted \ InvalidPDU \ InsufficientAuthentication \ InsufficientEncrypt \ InsufficientKeySize AttEvt_FindInformation SUCCESS \ ProcedureComplete Att_Error SUCCESS ====================== ========================== :param handle: connection to use :param start_handle: first requested handle number :param end_handle: last requested handle number """ if self._log_command(CC2650.att_find_information, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('start_handle', SEND), ':', x16(start_handle)) print(pc('end_handle', SEND), ':', x16(end_handle)) self.send_command(0xFD04, '3H', handle, start_handle, end_handle) @instruction(0xFD06) def att_find_by_type_value(self, handle: HANDLE, att_type: BT_UUID, value: bytes, start_handle: HANDLE = 0x0001, end_handle: HANDLE = 0xFFFF): r"""Send Find By Type Value Request. ======================= ========================== response status ======================= ========================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ Pending \ LinkEncrypted \ InvalidPDU \ InsufficientAuthentication \ InsufficientEncrypt \ InsufficientKeySize AttEvt_FindByTypeValue SUCCESS \ ProcedureComplete AttEvt_Error SUCCESS ======================= ========================== :param handle: connection to use :param att_type: UUID to find :param value: Attribute value to find, 0 .. ATT_MTU - 7 :param start_handle: first requested handle number :param end_handle: last requested handle number """ if self._log_command(CC2650.att_find_by_type_value, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('start_handle', SEND), ':', x16(start_handle)) print(pc('end_handle', SEND), ':', x16(end_handle)) print(pc('att_type', SEND), ':', x16(att_type), pp(GattUUID.att_str(att_type, None))) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFD06, '4H%ds' % len(value), handle, start_handle, end_handle, att_type, value) @instruction(0xFDB2) def gatt_discover_all_characteristics(self, handle: HANDLE, start_handle: HANDLE = 0x0001, end_handle: HANDLE = 0xFFFF): r"""This sub-procedure is used by a client to find all the characteristic declarations within a service definition on a server when only the service handle range is known. The service specified is identified by the service handle range. The ATT Read By Type Request is used with the Attribute Type parameter set to the UUID for "Characteristic". The Starting Handle is set to starting handle of the specified service and the Ending Handle is set to the ending handle of the specified service. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_ReadByType SUCCESS \ ProcedureComplete \ Timeout AttEvt_Error SUCCESS =================== ==================== :param handle: connection to use :param start_handle: starting handle :param end_handle: end handle """ if self._log_command(CC2650.gatt_discover_all_characteristics, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('start_handle', SEND), ':', x16(start_handle)) print(pc('end_handle', SEND), ':', x16(end_handle)) self.send_command(0xFDB2, '3H', handle, start_handle, end_handle) # noinspection PyPep8Naming @instruction(0xFD88) def gatt_discover_characteristics_by_UUID(self, handle: HANDLE, att_type: ATT_UUID, start_handle: HANDLE = 0x0001, end_handle: HANDLE = 0xFFFF): r"""This sub-procedure is used by a client to discover service characteristics on a server when only the service handle ranges are known and the characteristic UUID is known. The specific service may exist multiple times on a server. The characteristic being discovered is identified by the characteristic UUID. The ATT Read By Type Request is used with the Attribute Type is set to the UUID for "Characteristic" and the Starting Handle and Ending Handle parameters is set to the service handle range. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_ReadByType SUCCESS \ ProcedureComplete \ Timeout AttEvt_Error SUCCESS =================== ==================== :param handle: connection to use :param att_type: UUID :param start_handle: starting handle :param end_handle: end handle """ if isinstance(att_type, int): t = _struct.pack(' 512: raise RuntimeError('gatt_reliable_writes request too large') if self._log_command(CC2650.gatt_reliable_writes, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) for i, r in request: write_handle, write_offset, value = r print(pc('[%d] length' % i, SEND), ':', x8(len(value) + 4), pp(len(value) + 4)) print(pc('[%d] write_handle' % i, SEND), ':', x16(write_handle)) print(pc('[%d] write_offset' % i, SEND), ':', x16(write_offset)) print(pc('[%d] value' % i, SEND), ':', hl(value, truncate=True)) self.send_command(0xFDBA, 'H' + ''.join(f), handle, *d) @instruction(0xFDC2) def gatt_write_long_characteristic_descriptor(self, handle: HANDLE, write_handle: HANDLE, value: bytes, write_offset: uint16 = 0): r"""This sub-procedure is used to write a Characteristic Value to a server when the client knows the Characteristic Value Handle but the length of the Characteristic Value is longer than can be sent in a single Write Request Attribute Protocol message. The ATT Prepare Write Request and Execute Write Request are used to perform this sub-procedure. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_PrepareWrite SUCCESS \ Timeout AttEvt_ExecuteWrite SUCCESS \ Timeout AttEvt_Error SUCCESS =================== ==================== :param handle: connection to use :param write_handle: the handle of the attribute to be Written. :param write_offset: the offset of the first octet to be Written. :param value: part of the value of the attribute to be written. """ if value is None: raise RuntimeError('illegal gatt_write_long_characteristic_descriptor value : None') if len(value) > 512: raise RuntimeError('gatt_write_long_characteristic_descriptor value too large') if self._log_command(CC2650.gatt_write_long_characteristic_descriptor, x16(handle), x16(write_handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('write_handle', SEND), ':', x16(write_handle)) print(pc('write_offset', SEND), ':', x16(write_offset)) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFDC2, '3H%ds' % len(value), handle, write_handle, write_offset, value) @instruction(0xFD1E) def att_handle_value_confirmation(self, handle: HANDLE): r"""Send Handle Value Confirmation. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU =================== ==================== :param handle: connection to use """ if self._log_command(CC2650.att_handle_value_confirmation, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) self.send_command(0xFD1E, 'H', handle) @instruction(0xFD1B) def att_handle_value_notification(self, handle: HANDLE, authenticated: bool, att_handle: HANDLE, value: bytes): r"""Send Handle Value Notification. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU =================== ==================== :param handle: connection to use :param authenticated: whether or not an authenticated link is required: :param att_handle: the handle of the attribute. :param value: """ if self._log_command(CC2650.att_handle_value_notification, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('authenticated', SEND), ':', authenticated) print(pc('att_handle', SEND), ':', x16(att_handle)) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFD1B, 'HBH%ds' % len(value), handle, u8(authenticated), att_handle, value) @instruction(0xFD9B) def gatt_notification(self, handle: HANDLE, authenticated: bool, att_handle: HANDLE, value: bytes): r"""This sub-procedure is used when a server is configured to notify a characteristic value to a client without expecting any attribute protocol layer acknowledgement that the notification was successfully received. The ATT Handle Value Notification is used in this sub-procedure. Note: A notification may be sent at any time and does not invoke a confirmation. No confirmation will be sent to the calling application task for this sub-procedure. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU =================== ==================== :param handle: connection to use :param authenticated: whether an authenticated link is required :param att_handle: :param value: """ if self._log_command(CC2650.gatt_notification, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('authenticated', SEND), ':', authenticated) print(pc('att_handle', SEND), ':', x16(att_handle)) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFD9B, 'HBH%ds' % len(value), handle, u8(authenticated), att_handle, value) @instruction(0xFD1D) def att_handle_value_indication(self, handle: HANDLE, authenticated: bool, att_handle: HANDLE, value: bytes): r"""Send Handle Value Indication. =================== ==================== response status =================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU =================== ==================== :param handle: connection to use :param authenticated: whether or not an authenticated link is required: :param att_handle: the handle of the attribute. :param value: """ if self._log_command(CC2650.att_handle_value_indication, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('authenticated', SEND), ':', authenticated) print(pc('att_handle', SEND), ':', x16(att_handle)) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFD1D, 'HBH%ds' % len(value), handle, u8(authenticated), att_handle, value) @instruction(0xFD9D) def gatt_indication(self, handle: HANDLE, authenticated: bool, att_handle: HANDLE, value: bytes): r"""This sub-procedure is used when a server is configured to indicate a characteristic value to a client and expects an attribute protocol layer acknowledgement that the indication was successfully received. The ATT Handle Value Indication is used in this sub-procedure. =============================== ==================== response status =============================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_HandleValueConfirmation SUCCESS \ Timeout =============================== ==================== :param handle: connection to use :param authenticated: whether an authenticated link is required :param att_handle: :param value: """ if self._log_command(CC2650.gatt_indication, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('authenticated', SEND), ':', authenticated) print(pc('att_handle', SEND), ':', x16(att_handle)) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFD9B, 'HBH%ds' % len(value), handle, u8(authenticated), att_handle, value) @instruction(0xFD90) def gatt_discover_all_primary_services(self, handle: HANDLE): r"""This sub-procedure is used by a client to discover all the primary services on a server. The ATT Read By Group Type Request is used with the Attribute Type parameter set to the UUID for "Primary Service". The Starting Handle is set to 0x0001 and the Ending Handle is set to 0xFFFF. =============================== ==================== response status =============================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_ReadByGrpType SUCCESS \ Timeout \ ProcedureComplete AttEvt_Error SUCCESS =============================== ==================== :param handle: connection to use """ if not 0x001 <= handle <= 0xFFFF: raise RuntimeError('illegal gatt_discover_all_primary_services handle : ' + x16(handle)) if self._log_command(CC2650.gatt_discover_all_primary_services, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) self.send_command(0xFD90, 'H', handle) # noinspection PyPep8Naming @instruction(0xFD86) def gatt_discover_primary_service_by_UUID(self, handle: HANDLE, value: bytes): r"""This sub-procedure is used by a client to discover a specific primary service on a server when only the Service UUID is known. The primary specific service may exist multiple times on a server. The primary service being discovered is identified by the service UUID. The ATT Find By Type Value Request is used with the Attribute Type parameter set to the UUID for "Primary Service" and the Attribute Value set to the 16-bit Bluetooth UUID or 128-bit UUID for the specific primary service. The Starting Handle shall be set to 0x0001 and the Ending Handle shall be set to 0xFFFF. =============================== ==================== response status =============================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_ReadByTypeValue SUCCESS \ Timeout \ ProcedureComplete AttEvt_Error SUCCESS =============================== ==================== :param handle: connection to use :param value: """ if not 0x001 <= handle <= 0xFFFF: raise RuntimeError('illegal gatt_discover_primary_service_by_UUID handle : ' + x16(handle)) if self._log_command(CC2650.gatt_discover_primary_service_by_UUID, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('value', SEND), ':', hl(value, truncate=True)) self.send_command(0xFD86, 'H%ds' % len(value), handle, value) @instruction(0xFD80) def gatt_find_included_services(self, handle: HANDLE, start_handle: HANDLE = 0x0001, end_handle: HANDLE = 0xFFFF): r"""This sub-procedure is used by a client to find include service declarations within a service definition on a server. The service specified is identified by the service handle range. The ATT Read By Type Request is used with the Attribute Type parameter set to the UUID for "Included Service". The Starting Handle is set to starting handle of the specified service and the Ending Handle is set to the ending handle of the specified service. ==================== ==================== response status ==================== ==================== Evt_CommandStatus SUCCESS \ INVALID_PARAMETER \ MSG_BUFFER_NOT_AVAIL \ MemAllocError \ NotConnected \ InvalidPDU AttEvt_ReadByType SUCCESS \ Timeout \ ProcedureComplete AttEvt_Error SUCCESS ==================== ==================== :param handle: connection to use :param start_handle: starting handle :param end_handle: end handle """ if self._log_command(CC2650.gatt_find_included_services, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('start_handle', SEND), ':', x16(start_handle)) print(pc('end_handle', SEND), ':', x16(end_handle)) self.send_command(0xFD80, '3H', handle, start_handle, end_handle) @instruction(0xFDFC) def gatt_add_service(self, handle: HANDLE, primary_service: bool, number_attribute: uint16): """This command is used to add a new service to the GATT Server on the Network Processor when the GATT Database is implemented on the Application Processor. The GATT_AddAttribute command described in Section 18.25 must be used to add additional attributes to the service. The new service will be automatically registered with the GATT Server if it has no additional attribute to be added. Note: The Command Status Event will have the Start Handle and End Handle for the service registered with the GATT Server. No ATT request is used to perform this command. :param handle: :param primary_service: :param number_attribute: """ if self._log_command(CC2650.gatt_add_service, x16(handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('primary_service', SEND), ':', primary_service) print(pc('number_attribute', SEND), ':', x16(number_attribute), pp(number_attribute)) self.send_command(0xFDFC, '3H', handle, 0x2800 if primary_service else 0x2801, number_attribute) @instruction(0xFDFD) def gatt_del_service(self, handle: HANDLE, del_handle: HANDLE): """This command is used to delete a service from the GATT Server on the Network Processor when the GATT Database is implemented on the Application Processor. No ATT request is used to perform this command. :param handle: :param del_handle: The handle of the service to be deleted (0x0000 if the service hasn't been registered with the GATT Server yet). """ if self._log_command(CC2650.gatt_del_service, x16(handle), x16(del_handle)): print(pc('handle', SEND), ':', x16(handle)) print(pc('del_handle', SEND), ':', x16(del_handle)) self.send_command(0xFDFD, '2H', handle, del_handle) @instruction(0xFDFE) def gatt_add_attribute(self, handle: HANDLE, attribute: ATT_UUID, permissions: uint8): """This command is used to add a new attribute to the service being added to the GATT Server on the Network Processor when the GATT Database is implemented on the Application Processor. The service will be automatically registered with the GATT Server when its last attribute is added. Note: The Command Status Event will have the Start Handle and End Handle for the service registered with the GATT Server. No ATT request is used to perform this command. =========== ========================= permissions mean =========== ========================= 0x01 GATT_PERMIT_READ 0x02 GATT_PERMIT_WRITE 0x04 GATT_PERMIT_AUTHEN_READ 0x08 GATT_PERMIT_AUTHEN_WRITE 0x10 GATT_PERMIT_AUTHOR_READ 0x20 GATT_PERMIT_AUTHOR_WRITE =========== ========================= :param handle: :param attribute: :param permissions: """ if isinstance(attribute, int): a = _struct.pack(' Union[Optional[tuple], Any]: ret = None size = ''.join(x for x in f if x.isdigit()) type = ''.join(x for x in f if x.isalpha()) data = self._interface.recv_byte(int(size)) if data is not None and len(data) > 0: try: ret = _struct.unpack(str(len(data)) + type, data) except BaseException as e: print("decode recv data error:", data) finally: self._interface.flush() return ret else: self._interface.flush() return None def receive_timeout(self, f: str, timeout: float = 1) -> Union[Optional[tuple], Any]: start = _time() while True: ret = self.receive(f) if ret is None: if _time() - start > timeout: self._interface.flush() raise RecvTimeout() else: start = _time() self._interface.flush() return ret def recv(self, timeout: Optional[float] = None) -> Optional[list]: packet = self._recv_byte() # print("packet = ", packet) if packet is None: return None if packet == HCI_PACKET_EVENT: return self._recv_event(timeout) else: return None def _recv_byte(self) -> Optional[int]: ret = self._recv_bytes(1) return ret[0] if ret is not None else None def _recv_bytes(self, size: int = 1) -> Union[None, bytes]: start = _time() while True: ret = self._interface.recv_byte(size) if ret is not None and len(ret) > 0: return ret elif _time() - start > 0.01: # read timeout return None def _recv_event(self, timeout: Optional[float] = 1) -> Optional[list]: code = self._recv_byte() # print("code = ", code) if code is None: return None length = self._recv_byte() # print('code::',code,'length::',length) _start = _time() data = b'' while len(data) < length: ret = self._recv_bytes(length - len(data)) # print("ret = ", ret) if ret is not None: data += ret elif timeout is not None and _time() - _start > timeout: raise RecvTimeout() self._interface.flush() return list(data) def recv_uart(self, timeout: float = 1) -> Union[Optional[list], Any]: start = _time() while True: uart_data = self.recv(timeout) if uart_data is None: if _time() - start > timeout: # print("recv timeout") raise RecvTimeout() else: return uart_data