import struct from typing import cast from time import time import biopro.impl.cc2650.event as cc2650 from biopro.devlib.device import * from biopro.devlib.instruction import * from biopro.devlib.library import CompletedDevice from biopro.impl import LowLevelHardwareInterface from biopro.impl.cc2650.cc2650 import CC2650, CC2650Central, GapEvt_LinkTerminated from biopro.impl.cc2650.error_code import ErrorCode from biopro.impl.cc2650.event import RecvTimeout, UnexpectedEvent from biopro.impl.cc2650.gap import GapParameters from biopro.impl.gpio import P3Pin, OutputPin, GPIO from biopro.impl.interface_uart import SerialTimeoutException from biopro.impl.selector import Selector from biopro.text import * from biopro.util.address import EMPTY_ADDRESS, address_str from biopro.util.cli_base import CliOptions, cli_flags from biopro.util.console import YELLOW, CYAN, hex_line, hex_str from biopro.util.lock import Synchronized, synchronized from biopro.util.text import Table from biopro.server.api import ApiRequests from biopro.api.device import DeviceAPI # noinspection PyPep8 ''' Elite Legacy _handle_command_att_find_information Handle UUID Desp value content 0x0001 0x2800 GATT_PRIMARY_SERVICE 0x1800 00 18 0x0002 0x2803 GATT_CHARACTER ------r- 0x0003 0x2A00 02 03 00 00 2A 0x0003 0x2A00 DEVICE_NAME Simple BLE Peripheral 53 69 6D 70 6C 65 20 42 4C 45 20 50 65 72 69 70 68 65 72 61 6C 0x0004 0x2803 GATT_CHARACTER ------r- 0x0005 0x2A01 02 05 00 01 2A 0x0005 0x2A01 APPEARANCE .. 00 00 0x0006 0x2803 GATT_CHARACTER ------r- 0x0007 0x2A04 02 07 00 04 2A 0x0007 0x2A04 PERI_CONN_PARAM P....... 50 00 A0 00 00 00 E8 03 0x0008 0x2800 GATT_PRIMARY_SERVICE 0x1801 01 18 0x0009 0x2800 GATT_PRIMARY_SERVICE 0x180A 0A 18 0x000A 0x2803 GATT_CHARACTER ------r- 0x000B 0x2A23 02 0B 00 23 2A 0x000B 0x2A23 SYSTEM_ID .......h 0C FB 07 00 00 0B C9 68 0x000C 0x2803 GATT_CHARACTER ------r- 0x000D 0x2A24 02 0D 00 24 2A 0x000D 0x2A24 MODEL_NUMBER Model Number 4D 6F 64 65 6C 20 4E 75 6D 62 65 72 0x000E 0x2803 GATT_CHARACTER ------r- 0x000F 0x2A25 02 0F 00 25 2A 0x000F 0x2A25 SERIAL_NUMBER Serial Number 53 65 72 69 61 6C 20 4E 75 6D 62 65 72 0x0010 0x2803 GATT_CHARACTER ------r- 0x0011 0x2A26 02 11 00 26 2A 0x0011 0x2A26 FIRMWARE_REV Firmware Revision 46 69 72 6D 77 61 72 65 20 52 65 76 69 73 69 6F 6E 0x0012 0x2803 GATT_CHARACTER ------r- 0x0013 0x2A27 02 13 00 27 2A 0x0013 0x2A27 HARDWARE_REV Hardware Revision 48 61 72 64 77 61 72 65 20 52 65 76 69 73 69 6F 6E 0x0014 0x2803 GATT_CHARACTER ------r- 0x0015 0x2A28 02 15 00 28 2A 0x0015 0x2A28 SOFTWARE_REV Software Revision 53 6F 66 74 77 61 72 65 20 52 65 76 69 73 69 6F 6E 0x0016 0x2803 GATT_CHARACTER ------r- 0x0017 0x2A29 02 17 00 29 2A 0x0017 0x2A29 MANUFACTURER_NAME Manufacturer Name 4D 61 6E 75 66 61 63 74 75 72 65 72 20 4E 61 6D 65 0x0018 0x2803 GATT_CHARACTER ------r- 0x0019 0x2A2A 02 19 00 2A 2A 0x0019 0x2A2A IEEE_11073_CERT_DATA ..experimental FE 00 65 78 70 65 72 69 6D 65 6E 74 61 6C 0x001A 0x2803 GATT_CHARACTER ------r- 0x001B 0x2A50 02 1B 00 50 2A 0x001B 0x2A50 PNP_ID ....... 01 0D 00 00 00 10 01 0x001C 0x2800 GATT_PRIMARY_SERVICE 0xFFF0 F0 FF 0x001D 0x2803 GATT_CHARACTER ----w-r- 0x001E 0xFFF1 0A 1E 00 F1 FF 0x001E 0xFFF1 Characteristic 1 . 01 0x001F 0x2901 GATT_CHAR_USER_DESC Characteristic 1 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 31 0x0020 0x2803 GATT_CHARACTER ------r- 0x0021 0xFFF2 02 21 00 F2 FF 0x0021 0xFFF2 Characteristic 2 . 02 0x0022 0x2901 GATT_CHAR_USER_DESC Characteristic 2 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 32 0x0023 0x2803 GATT_CHARACTER ----w--- 0x0024 0xFFF3 08 24 00 F3 FF 0x0024 0xFFF3 Characteristic 3 0x0025 0x2901 GATT_CHAR_USER_DESC Characteristic 3 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 33 0x0026 0x2803 GATT_CHARACTER ---n---- 0x0027 0xFFF4 10 27 00 F4 FF 0x0027 0xFFF4 Characteristic 4 0x0028 0x2902 GATT_CLIENT_CHAR_CFG .. 00 00 0x0029 0x2901 GATT_CHAR_USER_DESC Characteristic 4 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 34 0x002A 0x2803 GATT_CHARACTER ------r- 0x002B 0xFFF5 02 2B 00 F5 FF 0x002B 0xFFF5 Characteristic 5 (error) 0x002C 0x2901 GATT_CHAR_USER_DESC Characteristic 5 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 35 Neulive 1.2 _handle_command_att_find_information Handle UUID Desp value content 0x0001 0x2800 GATT_PRIMARY_SERVICE 0x1800 00 18 0x0002 0x2803 GATT_CHARACTER ------r- 0x0003 0x2A00 02 03 00 00 2A 0x0003 0x2A00 DEVICE_NAME Neulive-v0.3 4E 65 75 6C 69 76 65 2D 76 30 2E 33 0x0004 0x2803 GATT_CHARACTER ------r- 0x0005 0x2A01 02 05 00 01 2A 0x0005 0x2A01 APPEARANCE .. 00 00 0x0006 0x2803 GATT_CHARACTER ------r- 0x0007 0x2A04 02 07 00 04 2A 0x0007 0x2A04 PERI_CONN_PARAM P....... 50 00 A0 00 00 00 E8 03 0x0008 0x2800 GATT_PRIMARY_SERVICE 0x1801 01 18 0x0009 0x2800 GATT_PRIMARY_SERVICE 0x180A 0A 18 0x000A 0x2803 GATT_CHARACTER ------r- 0x000B 0x2A23 02 0B 00 23 2A 0x000B 0x2A23 SYSTEM_ID .....-.. 80 9F 9C 00 00 2D 07 98 0x000C 0x2803 GATT_CHARACTER ------r- 0x000D 0x2A24 02 0D 00 24 2A 0x000D 0x2A24 MODEL_NUMBER Model Number 4D 6F 64 65 6C 20 4E 75 6D 62 65 72 0x000E 0x2803 GATT_CHARACTER ------r- 0x000F 0x2A25 02 0F 00 25 2A 0x000F 0x2A25 SERIAL_NUMBER Serial Number 53 65 72 69 61 6C 20 4E 75 6D 62 65 72 0x0010 0x2803 GATT_CHARACTER ------r- 0x0011 0x2A26 02 11 00 26 2A 0x0011 0x2A26 FIRMWARE_REV Firmware Revision 46 69 72 6D 77 61 72 65 20 52 65 76 69 73 69 6F 6E 0x0012 0x2803 GATT_CHARACTER ------r- 0x0013 0x2A27 02 13 00 27 2A 0x0013 0x2A27 HARDWARE_REV Hardware Revision 48 61 72 64 77 61 72 65 20 52 65 76 69 73 69 6F 6E 0x0014 0x2803 GATT_CHARACTER ------r- 0x0015 0x2A28 02 15 00 28 2A 0x0015 0x2A28 SOFTWARE_REV Software Revision 53 6F 66 74 77 61 72 65 20 52 65 76 69 73 69 6F 6E 0x0016 0x2803 GATT_CHARACTER ------r- 0x0017 0x2A29 02 17 00 29 2A 0x0017 0x2A29 MANUFACTURER_NAME Manufacturer Name 4D 61 6E 75 66 61 63 74 75 72 65 72 20 4E 61 6D 65 0x0018 0x2803 GATT_CHARACTER ------r- 0x0019 0x2A2A 02 19 00 2A 2A 0x0019 0x2A2A IEEE_11073_CERT_DATA ..experimental FE 00 65 78 70 65 72 69 6D 65 6E 74 61 6C 0x001A 0x2803 GATT_CHARACTER ------r- 0x001B 0x2A50 02 1B 00 50 2A 0x001B 0x2A50 PNP_ID ....... 01 0D 00 00 00 10 01 0x001C 0x2800 GATT_PRIMARY_SERVICE 0xFFF0 F0 FF 0x001D 0x2803 GATT_CHARACTER ------r- 0x001E 0xFFF1 02 1E 00 F1 FF 0x001E 0xFFF1 Characteristic 1 P. 50 00 0x001F 0x2901 GATT_CHAR_USER_DESC Characteristic 1 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 31 0x0020 0x2803 GATT_CHARACTER ------r- 0x0021 0xFFF2 02 21 00 F2 FF 0x0021 0xFFF2 Characteristic 2 .......... 02 00 00 00 00 00 00 00 00 00 0x0022 0x2901 GATT_CHAR_USER_DESC Characteristic 2 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 32 0x0023 0x2803 GATT_CHARACTER ----w--- 0x0024 0xFFF3 08 24 00 F3 FF 0x0024 0xFFF3 Characteristic 3 0x0025 0x2901 GATT_CHAR_USER_DESC Characteristic 3 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 33 0x0026 0x2803 GATT_CHARACTER ---n---- 0x0027 0xFFF4 10 27 00 F4 FF 0x0027 0xFFF4 Characteristic 4 0x0028 0x2902 GATT_CLIENT_CHAR_CFG .. 00 00 0x0029 0x2901 GATT_CHAR_USER_DESC Characteristic 4 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 34 0x002A 0x2803 GATT_CHARACTER ----w-r- 0x002B 0xFFF5 0A 2B 00 F5 FF 0x002B 0xFFF5 Characteristic 5 ..... 01 02 03 04 05 0x002C 0x2901 GATT_CHAR_USER_DESC Characteristic 5 43 68 61 72 61 63 74 65 72 69 73 74 69 63 20 35 ''' def is_headstage_device(info: cc2650.GapEvt_DeviceInformation) -> Optional[DeviceResponseInfo]: """check device whether a headstage device :param info: :return: ``True`` for this device is a headstage device """ local_name: bytes = None company_code = None version_info = None battery_info = None for data in info.data: if data.ad_type == cc2650.GapAdvertisingDataType.LOCAL_NAME_COMPLETE: local_name = data.ad_data elif data.ad_type == cc2650.GapAdvertisingDataType.MANUFACTURER_SPECIFIC: try: company_code = data.ad_data[:4] version_info = data.ad_data[4:10] battery_info = data.ad_data[10:] except IndexError: pass if local_name is None or company_code is None or version_info is None: return None if company_code == b'BPHS': if len(version_info) != 6: return None if battery_info[:3] == b'BAT': battery_percentage = struct.unpack(' Optional[DeviceResponseInfo]: if local_name is None or company_code is None or version_info is None: return None if company_code == ('B', 'P', 'H', 'S'): if battery_info[:3] == ('B', 'A', 'T'): battery_percentage = battery_info[3] << 8 | battery_info[4] # battery_percentage = struct.unpack(' None: super().__init__(log_name, log_name_color) self._master = master self._interface = interface self._options = options @property def master(self) -> MasterDevice: return self._master @property def interface(self) -> LowLevelHardwareInterface: return self._interface @property def options(self) -> CC2650DeviceOptions: return self._options def shutdown(self, release_resource=True): self._interface.close() self._interface = None @abc.abstractmethod def read_characteristic(self, device: int, handle: int) -> Optional[bytes]: """read data from device. the data format is depend on what instruction you send the caller should take care off the data every times you get.. :param device: device ID :param handle: device characteristic handle :return: data """ pass @abc.abstractmethod def write_characteristic(self, device: int, handle: int, data: bytes): """send instruction or data to device. :param device: device ID :param handle: device characteristic handle :param data: instruction or data, format please see :class:`biopro.devlib.device.DeviceInstruction` """ pass def handle_internal_instruction(self, device: Device, func: str, *para: Any): if func == InternalInstruction.PREDEFINED_NOTIFY: self.set_notify(device, *para) return True return False @abc.abstractmethod def set_notify(self, device: Union[int, Device], enable: bool): pass @classmethod def get_reset_pin(cls) -> OutputPin: if cls.CC2650_RESET_PIN is None: cls.CC2650_RESET_PIN = OutputPin.get_used(P3Pin.RESET, initial=True) return cls.CC2650_RESET_PIN class CC2650Device(Device): def __init__(self, device_id: int, master: CC2650MasterDevice, response_info: Optional[DeviceResponseInfo] = None): device_name = response_info.device_name if device_name is None: raise TypeError('NoneType device name') self._device_id = device_id self._device_name = device_name self._serial_number: DeviceSerialNumber = DeviceSerialNumber.UNKNOWN self._master = master self._link = cc2650.LinkParameter() self._address: ADDRESS = EMPTY_ADDRESS if response_info is not None: self._serial_number = response_info.serial_number self._address = response_info.mac_address if response_info.battery is not None: self._battery = response_info.battery self._start_flag = False self._notify_flag = False # device specific property self._battery: int = 100 self._parent: dict = {} self._recording_file_name: str = 'recording_data' self._coeff: bytes = b'' self._device_version = "" @property def device_id(self) -> int: return self._device_id @property def device_name(self) -> str: return self._device_name @property def link(self) -> cc2650.LinkParameter: """connection parameter""" return self._link @property def serial_number(self) -> DeviceSerialNumber: return self._serial_number @property def mac_address(self) -> ADDRESS: return self._address @property def is_started(self) -> bool: return self._start_flag @is_started.setter def is_started(self, value: bool): if value: self.send_instruction(DeviceInstruction.TYP_VIS, DeviceInstruction.VIS_STI) else: self.send_instruction(DeviceInstruction.TYP_VIS, DeviceInstruction.VIS_INT) @property def notify_enable(self) -> bool: return self._notify_flag def read_feature_mask(self) -> bytes: ret = self._master.read_characteristic(self._device_id, CC2650MasterDevice.FEATURE_HANDLE) if ret is None: # TODO error? raise RuntimeError() return ret def read_command_return_data(self) -> Optional[bytes]: try: return self._master.read_characteristic(self._device_id, CC2650MasterDevice.RETURN_HANDLE) except SendInstructionTimeoutError: self._master.log_warn('device', self.device_id, 'send_command timeout') raise def send_instruction(self, ins_type: int, ins_oper: int, *instruction: int): try: DeviceInstruction.valid_ins_type(ins_type) except ValueError as e: raise UnknownInstructionError(ins_type, ins_oper, instruction) from e data = self._encode_instruction(ins_type, ins_oper, *instruction) # print('send_instruction', data) try: self._master.write_characteristic(self._device_id, CC2650MasterDevice.COMMAND_HANDLE, data) except SendInstructionTimeoutError: self._start_flag = False raise else: if ins_type == DeviceInstruction.TYP_VIS: if ins_oper == DeviceInstruction.VIS_STI or ins_oper == DeviceInstruction.VIS_STARTS or ins_oper == DeviceInstruction.VIS_STARTR: self._start_flag = True elif ins_oper == DeviceInstruction.VIS_INT or ins_oper == DeviceInstruction.VIS_STOPS or ins_oper == DeviceInstruction.VIS_STOPR: self._start_flag = False def _encode_instruction(self, ins_type: int, ins_oper: int, *instruction: int) -> bytes: length = len(instruction) if length == 1 and instruction[0] < 0: return struct.pack('2B1b', (ins_type & 0xF0) | (self.device_id & 0x0F), (ins_oper & 0xF0) | (length & 0x0F), *instruction) return struct.pack('%dB' % (length + 2), (ins_type & 0xF0) | (self.device_id & 0x0F), (ins_oper & 0xF0) | (length & 0x0F), *instruction) def _decode_data(self, ins_oper: int, data: bytes) -> bytes: """CIS data decoder. :param ins_oper: CIS operator :param data: data receive from device :return: decoded data """ # TODO do something instead return directly. # eg: check HEADER return data @property def device_version(self) -> str: if self._device_version == "": self.update_device_version() return self._device_version def update_device_version(self): try: # send code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_VERSION) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) sleep(0.1) # receive data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) data = self._decode_data(DeviceInstruction.CIS_VERSION, data) except SendInstructionTimeoutError: self._master.log_warn('device', self.device_id, 'update_device_version no response') except RuntimeError: self._master.log_warn('device', self.device_id, 'update_device_version no response') else: if data is not None and len(data) > 0: year = struct.unpack(' int: self.update_battery_info() return self._battery @property def calibration(self) -> bytes: return self._coeff @property def parent(self) -> dict: return self._parent @property def recording_file_name(self) -> str: return self._recording_file_name def update_recording_file_name_info(self, _recording_file_name): self._recording_file_name = _recording_file_name def update_parent_info(self, _parent): self._parent = _parent def update_battery_info(self): # self.ensure_feature_support(DeviceFeature.BATTERY) try: # send code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_VOLT) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) sleep(0.1) # receive data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) data = self._decode_data(DeviceInstruction.CIS_VOLT, data) # print('code1', code) # print('data1',data) except SendInstructionTimeoutError: self._master.log_warn('device', self.device_id, 'update_battery_info no response') except RuntimeError: self._master.log_warn('device', self.device_id, 'update_battery_info no response') else: if data is not None and len(data) > 2 : battery = struct.unpack(' Optional[bytes]: if self._coeff is None: return None if len(self._coeff) == 0: self.update_calibration_info(device_type) return self._coeff def update_calibration_info(self, device_type: str): """ get device calibration info """ # FIXME ensure feature will cause error # self.ensure_feature_support(DeviceFeature.CALIBRATION) coeff = [] if device_type == 'TDC4VAF2': # neulive 1.3, 1.6 for i in range(24): # send code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_CALI, i) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) # receive data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) coeff.append(self._decode_data(DeviceInstruction.CIS_CALI, data)) elif device_type == 'TDC4VC': i = 0 request_times = 0 # neulive 2.1 (only support ch1~ch8) # try: # # send # code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_CALI, 0) # self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) # # receive # data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) # # coeff.append(self._decode_data(DeviceInstruction.CIS_CALI, data)) # except SendInstructionTimeoutError: # self._master.log_warn('device', self.device_id, 'update_calibration_info no response') # except RuntimeError: # self._master.log_warn('device', self.device_id, 'update_calibration_info no response') while i < 4: try: # print('i', i) # send code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_CALI, i) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) # receive data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) # print('code3', code) # print('data3', data) coeff.append(self._decode_data(DeviceInstruction.CIS_CALI, data)) except SendInstructionTimeoutError as e: print(e) self._master.log_warn('device', self.device_id, 'update_calibration_info no response', i) raise BaseException except RuntimeError as e: print(e) self._master.log_warn('device', self.device_id, 'update_calibration_info no response - 2', i) request_times += 1 if request_times > 3: self._master.reset(self.device_id) break else: # print('data success') if len(data) > 0: i += 1 else: request_times += 1 if request_times > 3: self._master.reset(self.device_id) break elif device_type == 'NeuliveThreeOne': i = 0 request_times = 0 while i < 8: try: # print('i', i) # send code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_CALI, i) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) # receive data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) coeff.append(self._decode_data(DeviceInstruction.CIS_CALI, data)) except SendInstructionTimeoutError as e: print(e) self._master.log_warn('device', self.device_id, 'update_calibration_info no response', i) raise BaseException except RuntimeError as e: print(e) self._master.log_warn('device', self.device_id, 'update_calibration_info no response - 2', i) request_times += 1 if request_times > 3: self._master.reset(self.device_id) break else: # print('data success') if len(data) > 0: i += 1 else: request_times += 1 if request_times > 3: self._master.reset(self.device_id) break elif device_type == 'EISZeroOne': i = 0 request_times = 0 while i < 7: try: # send code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_CALI, i) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) sleep(0.1) # receive data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) coeff.append(self._decode_data(DeviceInstruction.CIS_CALI, data)) except SendInstructionTimeoutError as e: print(e) self._master.log_warn('device', self.device_id, 'update_calibration_info no response') raise BaseException except RuntimeError as e: print(e) self._master.log_warn('device', self.device_id, 'update_calibration_info no response - 2') request_times += 1 if request_times > 3: self._master.reset(self.device_id) break else: # print('data success') if len(data) > 0: i += 1 else: request_times += 1 if request_times > 3: self._master.reset(self.device_id) break else: # default: neulive 2.1 for i in range(1): # send code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_CALI, i) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) # receive data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE) coeff.append(self._decode_data(DeviceInstruction.CIS_CALI, data)) self._coeff = b''.join(coeff) # self._coeff = None if self._coeff is None: self._master.log_verbose('cali_coeff', '(None)') else: self._master.log_verbose('cali_coeff', hex_line(self._coeff)) def update_radio_frequency(self, radio_frequency) -> bool: try: code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_RF, radio_frequency) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) except SendInstructionTimeoutError: self._master.log_warn('device', self.device_id, 'update radio frequency timeout error') return False except RuntimeError: self._master.log_warn('device', self.device_id, 'update radio frequency runtime error') return False else: return True def update_led_brightness(self, led_brightness) -> bool: try: code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_LED, led_brightness) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) except SendInstructionTimeoutError: self._master.log_warn('device', self.device_id, 'update led brightness timeout error') return False except RuntimeError: self._master.log_warn('device', self.device_id, 'update led brightness runtime error') return False else: return True def update_accelerator_sensitivity(self, accelerator_sensitivity) -> bool: try: code = self._encode_instruction(DeviceInstruction.TYP_CIS, DeviceInstruction.CIS_AS, accelerator_sensitivity) self._master.write_characteristic(self.device_id, CC2650MasterDevice.COMMAND_HANDLE, code) except SendInstructionTimeoutError: self._master.log_warn('device', self.device_id, 'update accelerator sensitivity timeout error') return False except RuntimeError: self._master.log_warn('device', self.device_id, 'update accelerator sensitivity runtime error') return False else: return True def central_rf_set(self, rf_power: int) -> bool: return self._master.central_rf_set(self.device_id, rf_power) def central_version_get(self) -> Optional[list]: return self._master.central_version_get(self._device_id) def __str__(self): return 'device[%d]' % self._device_id def __repr__(self): return 'device[%d]' % self._device_id class CC2650SingleMasterDevice(CC2650MasterDevice, Synchronized): MTU_SIZE = 244 ESTABLISH_LINK_INTERVAL = (8, 30) ESTABLISH_LINK_LATENCY = 0 ESTABLISH_LINK_TIMEOUT = 1000 SCAN_DISCOVER_TIME_MIN = 2000 def __init__(self, master: MasterDevice, interface: LowLevelHardwareInterface, options: CC2650DeviceOptions): super().__init__(master, interface, options, log_name_color=CYAN) Synchronized.__init__(self) print("create single host test 2650") self._disable_data_length_extension = options.flag_disable_data_length_extension self._cc2650 = CC2650(interface) self._info: Optional[cc2650.GapEvt_DeviceInitDone] = None self._link = cc2650.LinkParameter() self._found: List[DeviceResponseInfo] = [] self._handle: Optional[int] = None self._device: Optional[CC2650Device] = None self._scan_max_time: int = 5 self._cond = cc2650.RecvCond({ cc2650.GapEvt_DeviceInitDone: self._device_init_done, cc2650.GapEvt_LinkParamUpdate: self._device_link_update, cc2650.GapEvt_LinkTerminated: self._device_link_terminated, cc2650.AttEvt_MtuUpdated: None, cc2650.AttEvt_Write: None, # write event cc2650.GapEvt_LinkEstablished: None, # establish event cc2650.GapEvt_DeviceDiscovery: None, # scan event cc2650.GapEvt_DeviceInformation: None, # scan event }) def set_log_level(self, level: int): super().set_log_level(level) self._cc2650.set_log_level(level) def get_device(self, device: int) -> Optional[CC2650Device]: return self._device def _device_init_done(self, event: cc2650.GapEvt_DeviceInitDone): self._info = event event.consume() def _device_link_update(self, event: cc2650.GapEvt_LinkParamUpdate): """handle cc2650 GapEvt_LinkParamUpdate, update link parameters. :param event: GapEvt_LinkParamUpdate """ device = self.get_device(event.handle) if device is not None: event.update(device.link) event.consume() def _device_link_terminated(self, event: Union[int, cc2650.GapEvt_LinkTerminated, CC2650Device]): if isinstance(event, int): h = event elif isinstance(event, CC2650Device): h = event.device_id elif isinstance(event, cc2650.GapEvt_LinkEstablished): h = event.handle event.consume() elif hasattr(event, 'handle'): h = int(getattr(event, 'handle')) elif hasattr(event, 'device_id'): h = int(getattr(event, 'device_id')) else: return self.log_verbose('device_link_terminated', h) event.consume() raise cc2650.RecvReturn() @synchronized def reset(self, device: Optional[List[int]] = None, software_reset=True): """reset device. :param device: ignored :param software_reset: """ self.log_info('reset') self.reset_internal() with self: self.reset_hardware() if software_reset: self.reset_software() self.init_hardware() @synchronized def shutdown(self, release_resource=True): """shutdown the CC2650 master device. :param release_resource: close the hardware interface. """ self.log_info('shutdown') self._device = None self._handle = None self._info = None self.reset_hardware() if release_resource: self._interface.close() self._interface = None self._cc2650 = None self._info = None self._link = None self._found = None def reset_internal(self): self._found.clear() self._device = None self._handle = None self._info = None @synchronized def reset_hardware(self): """hardware reset cc2650 (master)""" pin = self.get_reset_pin() self.log_verbose('reset (hardware)') pin.output(False) sleep(0.1) pin.output(True) sleep(0.1) @synchronized def reset_software(self): """software reset cc2650 (master)""" self.log_verbose('reset (software)') try: self._cc2650.hci_reset() self._cc2650.recv_util(cc2650.HciEvt_CommandComplete) except SerialTimeoutException as e: raise RecvTimeout('device CC2650 reset fail') from e @synchronized def init_hardware(self, timeout=1): """initialize cc2650 (master)""" self._cc2650.gap_device_init() self._cc2650.recv_util(..., when=self._cond, timeout=1) # establish link parameter self._cc2650.gap_set_parameter(GapParameters.TGAP_CONN_EST_INT_MIN, self.ESTABLISH_LINK_INTERVAL[0]) self._cc2650.gap_set_parameter(GapParameters.TGAP_CONN_EST_INT_MAX, self.ESTABLISH_LINK_INTERVAL[1]) self._cc2650.gap_set_parameter(GapParameters.TGAP_CONN_EST_LATENCY, self.ESTABLISH_LINK_LATENCY) self._cc2650.gap_set_parameter(GapParameters.TGAP_CONN_EST_SUPERV_TIMEOUT, self.ESTABLISH_LINK_TIMEOUT) # Minimum time to perform scanning self._cc2650.gap_set_parameter(GapParameters.TGAP_GEN_DISC_SCAN, self.SCAN_DISCOVER_TIME_MIN) self._cc2650.gap_set_parameter(GapParameters.TGAP_LIM_DISC_SCAN, self._scan_max_time * 1000) self._cc2650.recv_util(..., when=self._cond, timeout=timeout) @synchronized def scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: if self._scan_max_time != timeout: # self.log_info('set max scan limit time', timeout) self._scan_max_time = timeout self._cc2650.gap_set_parameter(GapParameters.TGAP_LIM_DISC_SCAN, 1000 * timeout) self._cc2650.recv_util(..., when=self._cond, timeout=1) self._found = found = [] def _handle(info: cc2650.GapEvt_DeviceInformation): info.consume() response = is_headstage_device(info) if response is not None: self.log_info('found', address_str(response.mac_address), response.serial_number) found.append(response) callback(response) self._cc2650.gap_device_discovery() cond = self._cond.cover({ cc2650.GapEvt_DeviceInformation: _handle }) try: event = self._cc2650.recv_util(cc2650.GapEvt_DeviceDiscovery, when=cond, timeout=timeout + 1) return event.event_status == cc2650.ErrorCode.SUCCESS except RecvTimeout: return False def found(self) -> List[DeviceResponseInfo]: return self._found @synchronized def connect(self, response: DeviceResponseInfo) -> Optional[CC2650Device]: if self._handle is not None: raise RuntimeError('cannot connect any more') address = response.mac_address address_s = cc2650.address_str(address) self.log_info(DEVICE_CONNECTING, address_s) self._cc2650.gap_establish_link(address) try: event = self._cc2650.recv_util(cc2650.GapEvt_LinkEstablished, when=self._cond, timeout=5) except RecvTimeout as e: raise RuntimeError(DEVICE_CONNECT_FAIL, address_s) from e else: if event.event_status != ErrorCode.SUCCESS: raise RuntimeError(DEVICE_CONNECT_FAIL, address_s) if type(event) != GapEvt_LinkTerminated: self._handle = event.handle self._device = ret = CC2650Device(event.handle, self, response) event.update(ret.link) self.log_info(DEVICE_CONNECTED, address_s) ''' connect_post recv_util fail make self._handle clear ''' try: self._connect_post(ret) except RecvTimeout as e: print(e) self._handle = None return ret else: return None # noinspection PyUnusedLocal @synchronized def _connect_post(self, device: CC2650Device): if not self._disable_data_length_extension: self._cc2650.att_exchange_mtu(self._handle, self.MTU_SIZE) self._cc2650.recv_util(cc2650.AttEvt_ExchangeMTU, when=self._cond, timeout=1) # noinspection PyUnusedLocal @synchronized def _connect_post(self, device: CC2650Device): if not self._disable_data_length_extension: self._cc2650.att_exchange_mtu(self._handle, self.MTU_SIZE) self._cc2650.recv_util(cc2650.AttEvt_ExchangeMTU, when=self._cond, timeout=1) @synchronized def disconnect(self, device: int, force=False) -> bool: self.log_info(DEVICE_DISCONNECTING, device) self._cc2650.gap_terminate_link(self._handle) # consume all disconnect event # no matter Evt_CommandStatus.event_status == SUCCESS or IncorrectMode cond = self._cond.cover({ cc2650.Evt_CommandStatus: None }) try: self._cc2650.recv_util(cc2650.GapEvt_LinkTerminated, when=cond, timeout=1) except RecvTimeout: self.log_warn('disconnect time out') # reset single 2650 if disconnect time out self.reset_internal() self.reset_hardware() self.init_hardware(timeout=1) # self._interface.flush() return False except RuntimeError as e: self.log_warn('suppressed error', str(e)) return False else: return True finally: self.log_info(DEVICE_DISCONNECTED, device) self._handle = None self._device = None @synchronized def read_characteristic(self, device: int, handle: int) -> Optional[bytes]: def _handle(status: cc2650.Evt_CommandStatus): if status.event_status == ErrorCode.SUCCESS: status.consume() elif status.event_status == ErrorCode.Pending: status.consume() raise cc2650.RecvReturn(status) elif status.event_status == ErrorCode.NotConnected: status.consume() raise cc2650.RecvReturn(status) cond = self._cond.cover({ cc2650.Evt_CommandStatus: _handle }) self._cc2650.att_read(self._handle, handle) try: event = self._cc2650.recv_util(cc2650.AttEvt_Read, cond, timeout=3) except RecvTimeout as e: raise SendInstructionTimeoutError(None) from e else: if event.event_status == ErrorCode.NotConnected: raise SendInstructionNotConnectedError(None) elif event.event_status == ErrorCode.Pending: return None else: return event.data @synchronized def write_characteristic(self, device: int, handle: int, data: bytes): # print(hex_line(data)) def _handle(status: cc2650.Evt_CommandStatus): if status.event_status == ErrorCode.SUCCESS: status.consume() elif status.event_status == ErrorCode.Pending: status.consume() raise cc2650.RecvReturn(status) elif status.event_status == ErrorCode.NotConnected: status.consume() raise cc2650.RecvReturn(status) cond = self._cond.cover({ cc2650.Evt_CommandStatus: _handle }) counter = 0 while counter < 3: self._cc2650.att_write(self._handle, handle, data) try: event = self._cc2650.recv_util(cc2650.AttEvt_Write, cond, timeout=3) except RecvTimeout as e: raise SendInstructionTimeoutError(None) from e else: if event.event_status == ErrorCode.NotConnected: raise SendInstructionNotConnectedError(None) elif event.event_status == ErrorCode.Pending: counter += 1 continue else: break else: raise SendInstructionTimeoutError(None, 'pending') def set_notify(self, device: Union[int, Device], enable: bool): if isinstance(device, CompletedDevice): device = device.device.device_id elif isinstance(device, CC2650Device): device = device.device_id assert isinstance(device, int) if enable: self.log_verbose(device, 'notify enable') message = 'enable' value = b'\01\00' else: self.log_verbose(device, 'notify disable') message = 'disable' value = b'\00\00' self._device._notify_flag = enable try: self.write_characteristic(self._handle, self.NOTIFY_HANDLE, value) except SendInstructionTimeoutError: self.log_warn('set notify', message, 'timeout') raise def handle_command(self, device: Optional[int], command: str, options: Dict[str, str]) -> DeviceCommandAction: if device is not None and command == 'att_find_information': self._handle_command_att_find_information() return DeviceCommandAction.NOTHING return self._raise_command_not_found(device, command) @synchronized def _handle_command_att_find_information(self): G = cc2650.GattUUID def _gatt_read_characteristic_value(_handle: int) -> bytes: self._cc2650.gatt_read_characteristic_value(self._handle, _handle) _read = self._cc2650.recv_util(cc2650.AttEvt_Read, when=self._cond, timeout=1) return _read.data self._cc2650.att_find_information(self._handle) try: info_list = self._cc2650.recv_collect(cc2650.AttEvt_FindInformation, when=self._cond, timeout=5) except RecvTimeout: self.log_warn('_handle_command_att_find_information timeout') return # dict {handle -> (permissions, char_uuid)} all_char_handle: Dict[int, Tuple[int, int]] = {} t = Table('Handle', 'UUID', 'Desp', 'value', 'content') for info in info_list: for handle, uuid in info.info: data = b'' value = '' if info.format == 0x01: gatt = G.att_str(cast(int, uuid)) if gatt is None and uuid >= 0xFFF0: gatt = 'Characteristic %d' % (uuid - 0xFFF0) else: gatt = address_str(cast(Tuple[int, ...], uuid)) try: if handle in all_char_handle: permissions, char_uuid = all_char_handle[handle] if (permissions & cc2650.GattValuePermission.READ) > 0: data = _gatt_read_characteristic_value(handle) else: data = _gatt_read_characteristic_value(handle) except UnexpectedEvent: t.append('0x%04X' % handle, '0x%04X' % uuid, gatt, value, '(error)') else: if uuid in (G.DEVICE_NAME, G.GATT_CHAR_USER_DESC): value = hex_str(data) elif uuid == G.GATT_PRIMARY_SERVICE: value = '0x%04X' % ((data[1] << 8) | data[0]) elif uuid == G.GATT_CHARACTER: permissions = data[0] char_handle = (data[2] << 8) | data[1] char_uuid = (data[4] << 8) | data[3] all_char_handle[char_handle] = (permissions, char_uuid) permissions = cc2650.GattValuePermission.permission_str(permissions) char_handle = '0x%04X' % char_handle char_uuid = '0x%04X' % char_uuid value = '%s %s %s' % (permissions, char_handle, char_uuid) elif len(data) > 0: value = hex_str(data) t.append('0x%04X' % handle, '0x%04X' % uuid, gatt, value, hex_line(data)) t.set_format(0, align_right=True, split='') t.set_format(1, align_right=False) t.set_format(2, align_right=False) t.set_format(3, align_right=False) t.set_format(4, align_right=False) t.print() def hardware_test(self) -> Dict[str, Any]: return { 'hardware': 'cc2650:single', 'device': None if self._device is None else self._device.as_json(), 'found': [f.as_json() for f in self._found] } class CC2650MultiMasterDevice(CC2650MasterDevice, Synchronized): def __init__(self, master: MasterDevice, interface: LowLevelHardwareInterface, selector: Selector, options: CC2650DeviceOptions): super().__init__(master, interface, options, log_name_color=YELLOW) Synchronized.__init__(self) self._selector = selector sz = selector.SIZE self._cc2650: List[Optional[CC2650SingleMasterDevice]] = [None for _ in range(sz)] # connected device self._device: List[Optional[CC2650Device]] = [None for _ in range(sz)] self._cc2650_log_level = self.log_level def set_log_level(self, level: int): # do not overwrite self log level, pass to CC2650SingleMasterDevice self._cc2650_log_level = level for d in self._cc2650: if d is not None: d.set_log_level(level) @synchronized def reset(self, device: Optional[List[int]] = None, software_reset=True): self.log_verbose('reset') # reset hardware for i in range(len(self._cc2650)): if device is None or i in device: self._device[i] = None m = self._cc2650[i] if m is None: m = CC2650SingleMasterDevice(self._master, self._interface, self._options) m.set_log_level(self._cc2650_log_level) self._interface.flush() self._selector.select(i) try: self.log_info('reset device', i) with m: m.reset_internal() m.reset_hardware() if software_reset: m.reset_software() self._cc2650[i] = m except RecvTimeout: self.log_warn('reset device', i, 'fail') self._cc2650[i] = None if not software_reset: return # check reset result available_device_id = list(map(lambda it: it[0], filter(lambda it: it[1] is not None, enumerate(self._cc2650)))) if len(available_device_id) == 0: raise RuntimeError("reset fail, no device found") self.log_verbose('available device', ' '.join(map(str, available_device_id))) # initial hardware for i in range(len(self._cc2650)): m = self._cc2650[i] if m is not None: self._interface.flush() self._selector.select(i) try: self.log_info('init device', i) m.init_hardware(timeout=1) except RecvTimeout: self.log_warn('init device', i, 'fail') self._cc2650[i] = None # check init result, again available_device_id = list(map(lambda it: it[0], filter(lambda it: it[1] is not None, enumerate(self._cc2650)))) if len(available_device_id) == 0: raise RuntimeError("init fail, no device success") def _reset(self, device: int) -> bool: master = self._cc2650[device] if master is not None: with self: if self._selector.channel != device: self._interface.flush() self._selector.select(device) # reset try: self.log_info('reset device', device) with master: master.reset_internal() master.reset_hardware() master.reset_software() except RecvTimeout: self.log_warn('reset device', device, 'fail') self._cc2650[device] = None # initial try: self.log_info('init device', device) master.init_hardware(timeout=1) except RecvTimeout: self.log_warn('init device', device, 'fail') self._cc2650[device] = None return self._cc2650[device] is not None @synchronized def shutdown(self, release_resource=True): self.log_verbose('shutdown') for i in range(len(self._cc2650)): m = self._cc2650[i] self._cc2650[i] = None self._device[i] = None if m is not None: self.log_verbose('shutdown', i) self._interface.flush() self._selector.select(i) # shutdown single device, but not close interface because it is shared. m.shutdown(release_resource=False) # call super shutdown to close hardware interface super().shutdown() self._selector.close() def get_device(self, device: int) -> Optional[Device]: return self._device[device] def _get_empty_master(self, skip_channel: Optional[int] = None) -> Optional[int]: for i in range(len(self._cc2650)): if skip_channel is not None: if i == 0: continue else: i = (i + skip_channel) % len(self._cc2650) m = self._cc2650[i] d = self._device[i] if m is not None and d is None: return i return None def _foreach_empty_master(self) -> Iterable[int]: for i in range(len(self._cc2650)): m = self._cc2650[i] d = self._device[i] if m is not None and d is None: yield i def scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: self.log_verbose('scan') all_result = False for device in self._foreach_empty_master(): self.log_verbose('use', device) m = self._cc2650[device] d = self._device[device] if m is not None and d is None: error = None with self: self._interface.flush() self._selector.select(device) try: result = m.scan_callback(callback, timeout=timeout) except RuntimeError as e: error = e else: all_result = all_result or result if result and not all_device: return True if error is not None: self.log_warn('suppressed error : ' + str(error)) self._reset(device) return all_result def found(self) -> List[DeviceResponseInfo]: ret: List[DeviceResponseInfo] = [] def _contain(_found: DeviceResponseInfo): for _response in ret: if _found.match(_response): return True return False for device in self._cc2650: if device is not None: for found in device.found(): if not _contain(found): ret.append(found) return ret def connect(self, response: DeviceResponseInfo) -> Device: self.log_verbose('connect') device = None for device in self._foreach_empty_master(): m = self._cc2650[device] d = self._device[device] if response in m.found() and d is None: break else: if device is None: raise RuntimeError('cannot connect any more device') else: raise RuntimeError('rescan please') # self.log_verbose('use', device) m = self._cc2650[device] d = self._device[device] if d is not None: raise RuntimeError('cc2650 has connected a device') with self: self._interface.flush() self._selector.select(device) d = m.connect(response) # overwrite attribute d._device_id = device d._master = self self._device[device] = d return d def disconnect(self, device: int, force=False) -> bool: self.log_verbose('disconnect') m = self._cc2650[device] d = self._device[device] if m is None or d is None: return False self.log_verbose('disconnect', device) with self: try: return m.disconnect(device, force=force) finally: self._device[device] = None @synchronized def read_characteristic(self, device: int, handle: int) -> Optional[bytes]: self.log_verbose('read_characteristic', device, '0x%02X' % handle) master = self._cc2650[device] if master is not None: self._interface.flush() self._selector.select(device) try: return master.read_characteristic(device, handle) except SendInstructionTimeoutError: self.log_warn('device', device, 'send_instruction timeout') raise @synchronized def write_characteristic(self, device: int, handle: int, data: bytes): self.log_verbose('write_characteristic', device, '0x%02X' % handle) master = self._cc2650[device] if master is not None: self._interface.flush() self._selector.select(device) try: master.write_characteristic(device, handle, data) except SendInstructionTimeoutError: self.log_warn('device', device, 'send_instruction timeout') raise except SendInstructionNotConnectedError: self.log_warn('device', device, 'device is not connected') raise @synchronized def set_notify(self, device: Union[int, Device], enable: bool): if isinstance(device, CompletedDevice): device_id = device.device.device_id elif isinstance(device, CC2650Device): device_id = device.device_id else: device_id = device assert isinstance(device_id, int) master = self._cc2650[device_id] device = self._device[device_id] self.log_verbose('set_notify', device_id, enable) if enable: self.log_verbose(device_id, 'notify enable') message = 'enable' value = b'\01\00' else: self.log_verbose(device_id, 'notify disable') message = 'disable' value = b'\00\00' if master is not None: self._interface.flush() self._selector.select(device_id) device._notify_flag = enable try: master.write_characteristic(device_id, self.NOTIFY_HANDLE, value) except SendInstructionTimeoutError: self.log_warn('set notify', message, 'timeout') raise else: self.log_warn('cannot %s notify for device : %s' % (message, device.__class__.__name__)) def handle_command(self, device: Optional[int], command: str, options: Dict[str, str]) -> DeviceCommandAction: if device is not None and command == 'att_find_information': m = self._cc2650[device] if m is not None: with self: self._interface.flush() self._selector.select(device) return m.handle_command(0, command, options) return self._raise_command_not_found(device, command) def hardware_test(self) -> Dict[str, Any]: return { 'hardware': 'cc2650:multiple', 'available': list(map(lambda it: it[0], filter(lambda it: it[1] is not None, enumerate(self._cc2650)))), 'single': {i: m.hardware_test() for i, m in enumerate(self._cc2650) if m is not None} } class CC2650MasterCentralDevice(MasterDevice, metaclass=abc.ABCMeta): FEATURE_HANDLE = 0x1E '''device characteristic handle for feature support list''' RETURN_HANDLE = 0x21 '''device characteristic handle for data return''' COMMAND_HANDLE = 0x24 '''device characteristic handle for instruction send''' NOTIFY_HANDLE = 0x28 '''device characteristic handle for notification switch''' CC2650_RESET_PIN: OutputPin = None def __init__(self, master: MasterDevice, interface: LowLevelHardwareInterface, options: CC2650DeviceOptions, log_name: Optional[str] = None, log_name_color: int = WHITE) -> None: super().__init__(log_name, log_name_color) self._master = master self._interface = interface self._options = options @property def master(self) -> MasterDevice: return self._master @property def interface(self) -> LowLevelHardwareInterface: return self._interface @property def options(self) -> CC2650DeviceOptions: return self._options def shutdown(self, release_resource=True): self._interface.close() self._interface = None @abc.abstractmethod def read_characteristic(self, device: int, handle: int) -> Optional[bytes]: """read data from device. the data format is depend on what instruction you send the caller should take care off the data every times you get.. :param device: device ID :param handle: device characteristic handle :return: data """ pass @abc.abstractmethod def write_characteristic(self, device: int, handle: int, data: bytes): """send instruction or data to device. :param device: device ID :param handle: device characteristic handle :param data: instruction or data, format please see :class:`biopro.devlib.device.DeviceInstruction` """ pass def handle_internal_instruction(self, device: Device, func: str, *para: Any): if func == InternalInstruction.PREDEFINED_NOTIFY: self.set_notify(device, *para) return True return False @abc.abstractmethod def set_notify(self, device: Union[int, Device], enable: bool): pass @classmethod def get_reset_pin(cls) -> OutputPin: if cls.CC2650_RESET_PIN is None: cls.CC2650_RESET_PIN = OutputPin.get_used(P3Pin.RESET, initial=True) return cls.CC2650_RESET_PIN class CC2650SingleMasterCentralDevice(CC2650MasterDevice, Synchronized): """ instruction format: ins[0]: 0x00: INS_IDLE, 0x01: INS_RESET, 0x02: INS_KEY (reserved for debug), 0x03: INS_SCAN, 0x04: INS_SCAN_RESPONSE, 0x05: INS_CONNECT, 0x06: INS_WRITE, 0x07: INS_READ, 0x08: INS_DISCONNECT ins[0]: 0x04 (get_scan_response) ins[1]: 0x00: number of scanned device, 0x01 ~ 0x08: get a certain device information = device_id (could be 1~8) ins[2]: selected device information 0x00: attr_length, e.g. len(addr)=6, len(company_code)=4 0x01: addr = MAC, 0x02: localName, 0x03: company_code, 0x04: version_info, 0x05: battery_info, 0x06: get all information ins[0]: 0x05 (connect) ins[1]: 0x01 ~ 0x08: device id """ MTU_SIZE = 244 ESTABLISH_LINK_INTERVAL = (8, 16) ESTABLISH_LINK_LATENCY = 0 ESTABLISH_LINK_TIMEOUT = 1000 SCAN_DISCOVER_TIME_MIN = 2000 CC2650_COMMAND_LEN = "3B" CC2650_INSTRUCTION_LEN = "20B" # uart using preamble here, ensure memory board cc2650 sync with controller # The chosen preamble is a 16-bits M-sequence: 0b1010001100101001 = 0xA329 PREAMBLE_16_BIT_M_SEQ1 = 0xA3 PREAMBLE_16_BIT_M_SEQ2 = 0x29 def __init__(self, master: MasterDevice, interface: LowLevelHardwareInterface, options: CC2650DeviceOptions): super().__init__(master, interface, options, log_name_color=CYAN) Synchronized.__init__(self) self._disable_data_length_extension = options.flag_disable_data_length_extension self._cc2650 = CC2650Central(interface) self._info: Optional[cc2650.GapEvt_DeviceInitDone] = None self._link = cc2650.LinkParameter() self._found: List[DeviceResponseInfo] = [] self._found_with_id: List[(DeviceResponseInfo, int)] = [] self._handle: Optional[int] = None self._device: Optional[CC2650Device] = None self._scan_max_time: int = 5 self._notify_enable_status = False def set_log_level(self, level: int): super().set_log_level(level) self._cc2650.set_log_level(level) def get_device(self, device: int) -> Optional[CC2650Device]: return self._device def _device_init_done(self, event: cc2650.GapEvt_DeviceInitDone): self._info = event event.consume() def _device_link_update(self, event: cc2650.GapEvt_LinkParamUpdate): """handle cc2650 GapEvt_LinkParamUpdate, update link parameters. :param event: GapEvt_LinkParamUpdate """ device = self.get_device(event.handle) if device is not None: event.update(device.link) event.consume() def _device_link_terminated(self, event: Union[int, cc2650.GapEvt_LinkTerminated, CC2650Device]): if isinstance(event, int): h = event elif isinstance(event, CC2650Device): h = event.device_id elif isinstance(event, cc2650.GapEvt_LinkEstablished): h = event.handle event.consume() elif hasattr(event, 'handle'): h = int(getattr(event, 'handle')) elif hasattr(event, 'device_id'): h = int(getattr(event, 'device_id')) else: return self.log_verbose('device_link_terminated', h) event.consume() raise cc2650.RecvReturn() @synchronized def reset(self, device: Optional[List[int]] = None, software_reset=True): """reset device. :param device: ignored :param software_reset: """ self.log_info('reset') self.reset_internal() with self: self.reset_hardware() if software_reset: self.reset_software() self.init_hardware() @synchronized def shutdown(self, release_resource=True): """shutdown the CC2650 master device. :param release_resource: close the hardware interface. """ self.log_info('shutdown') self._device = None self._handle = None self._info = None self.reset_hardware() if release_resource: self._interface.close() self._interface = None self._cc2650 = None self._info = None self._link = None self._found = None def reset_internal(self): self._found.clear() self._device = None self._handle = None self._info = None @synchronized def reset_hardware(self): """hardware reset cc2650 (master)""" pin = self.get_reset_pin() self.log_verbose('reset (hardware)') pin.output(False) # sleep(0.001) pin.output(True) @synchronized def reset_software(self): """software reset cc2650 (master)""" self.log_verbose('reset (software)') try: #self._cc2650.send("3B", 1, 0, 0) pass except SerialTimeoutException as e: raise RecvTimeout('device CC2650 reset fail') from e finally: sleep(0.1) @synchronized def init_hardware(self, timeout=1): """initialize cc2650 (master)""" pass def scan_send_ins(self): # send scan command try: print(':: scan_send_ins ::', self.CC2650_COMMAND_LEN, 3, 0, 0) self._cc2650.send(self.CC2650_COMMAND_LEN, 3, 0, 0) except SerialTimeoutException as e: raise RecvTimeout('send scan fail') from e else: # wait scanning # sleep(2) clean_buf = self._cc2650.receive_timeout("20B", timeout=3) print("clean_buf = ", clean_buf) # def cc2650_uart_irq(self): # uart_irq = self.get_uart_irq_pin() # uart_irq.output(False) # sleep(0.001) # uart_irq.output(True) @synchronized def scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: self._found = found = [] self._found_with_id = [] hdr_BPHS = [66, 80, 72, 83] scan_response: Union[Optional[tuple], Any] = None # build scan instruction scan_ins = bytearray() scan_ins.append(3) #scan instruction scan_ins.append(1) #length scan_ins.append(0xF1) # print('send_scan', bytes(scan_ins)) self._cc2650.send("bytes", bytes(scan_ins)) try: scan_response = self._cc2650.recv_uart(timeout) except RecvTimeout: # self.reset_internal() # self.reset_hardware() # self._interface.flush() return False # instruction format: # ins[0]: get_scan_response = 0x04 # ins[1]: number of scanned device=0; a certain device = device_id (could be 1~8) # ins[2]: addr=MAC=1, localName=2, company_code=3, version_info=4, battery_info=5, all_info=6; # attr_length=0, e.g. len(addr)=6, len(company_code)=4 local_mac = None local_cc = None local_ver = None local_bat = None local_name = None local_addr_type = None # get device attribute length attr_length = [6, 4, 6, 5, 20] index = 0 local_mac = get_device_mac_in_address_format(scan_response[index:index + 6]) index = index + 6 # print("local_mac = ", hex(local_mac[0]), hex(local_mac[1]), # hex(local_mac[2]), hex(local_mac[3]), # hex(local_mac[4]), hex(local_mac[5])) local_cc = get_device_company_code(scan_response[index:index + 4]) index = index + 4 # print("local_cc = ", local_cc) local_ver = scan_response[index:index + 6] index = index + 6 # print("local_ver = ", local_ver) local_bat = get_device_battery_info(scan_response[index:index + 5]) index = index + 5 # print("local_bat = ", local_bat) local_name = get_device_name_in_string_format(list(scan_response[index:index + 20])) index = index + 20 # print("local_name = ", local_name) # addr type is don't care in BMD380 print('scan_response:', list(scan_response)) local_addr_type = int(scan_response[index:index + 1][0]) # local_addr_type = 0xFF index = index + 1 # print("local_addr_type = ", local_addr_type) response = is_headstage_device_central_version(local_mac, local_addr_type, local_name, local_cc, local_ver, local_bat) if response is not None: self.log_info('found', address_str(response.mac_address), response.serial_number) self._interface.flush_input() # apppend into db devicesList = DeviceAPI.getByMac(address_str(response.mac_address)) if devicesList is not None: if len(devicesList) == 0: DeviceAPI.create(response.device_name, local_ver, address_str(response.mac_address)) found.append(response) # print('scan_done_found', found) self._found_with_id.append((response, 0 + 1)) callback(response) return True def decode_uart_preamble(self, raw_uart: tuple, expect_ret_len: int = 0) -> Optional[list]: # print("decode_uart_preamble: raw_uart = ", raw_uart) if raw_uart is None: return None raw_uart_len = len(raw_uart) if expect_ret_len > 0: for index in range(raw_uart_len): if raw_uart[index] is self.PREAMBLE_16_BIT_M_SEQ1 \ and raw_uart[index + 1] is self.PREAMBLE_16_BIT_M_SEQ2: # print("ret = ", list(raw_uart[index + 2:])) return list(raw_uart[index + 2:index + 2 + expect_ret_len]) else: for index in range(7): if raw_uart[index] is self.PREAMBLE_16_BIT_M_SEQ1 \ and raw_uart[index + 1] is self.PREAMBLE_16_BIT_M_SEQ2: # print("ret = ", list(raw_uart[index + 2:])) return list(raw_uart[index + 2:raw_uart_len]) return None def found(self) -> List[DeviceResponseInfo]: return self._found def check_mem_survive(self) -> Optional[CC2650Device]: ack = [] ins = bytearray() ins.append(10) ins.append(1) #length ins.append(0xF1) # print('ins', list(ins)) try: self._cc2650.send("bytes", bytes(ins)) except SerialTimeoutException as e: raise RecvTimeout('device CC2650 check_mem_survive timeout') from e else: try: ack = self._cc2650.recv_uart(0.001) except RecvTimeout: self.log_info("no memory board") # else: # print('ack=', ack) if ack == [3]: return True return False @synchronized def connect(self, response: DeviceResponseInfo, direct_connect: bool = False) -> Optional[CC2650Device]: if self._handle is not None: raise RuntimeError('cannot connect any more') address = response.mac_address addr_type = response.addr_type address_s = cc2650.address_str(address) self.log_info(DEVICE_CONNECTING, address_s) connect_ins = bytearray() connect_ins.append(5) connect_ins.append(8) #length connect_ins.append(addr_type) connect_ins.append(address[0]) connect_ins.append(address[1]) connect_ins.append(address[2]) connect_ins.append(address[3]) connect_ins.append(address[4]) connect_ins.append(address[5]) connect_ins.append(0xF1) connected = False # send connect command if direct_connect is True: # send device mac and addrType try: self._cc2650.send("bytes", bytes(connect_ins)) except SerialTimeoutException as e: raise RecvTimeout('device CC2650 connect fail') from e return None # connection establish done? for retry_recv_ack in range(10): try: con_done = self._cc2650.recv_uart(timeout = 0.5) # print("con_done = ", con_done) except RecvTimeout: self.log_info("recv connection timeout, retry... ") continue # is the ack valid? if con_done is None: continue elif con_done[0] is 46 and \ con_done[1] is 80 and \ con_done[2] is 48 and \ con_done[3] is 4: connected = True break else: for dev in self._found: if dev.mac_address == address: # send device mac and addrType try: # print('send_connect',bytes(connect_ins)) self._cc2650.send("bytes", bytes(connect_ins)) except SerialTimeoutException as e: raise RecvTimeout('device CC2650 connect fail') from e else: sleep(2) # connection establish done? for retry_recv_ack in range(5): self._cc2650.send("bytes", bytes((0, 0, 0, 0))) try: con_done = self._cc2650.recv_uart(timeout = 0.1) except RecvTimeout: self.log_info("recv connection timeout, retry... ") continue # is the ack valid? if con_done is None: continue elif con_done[0] is 46 and \ con_done[1] is 80 and \ con_done[2] is 48 and \ con_done[3] is 4: connected = True # print('con_done=', con_done) break else: continue # if select device is invalid or connect failed self._interface.flush_input() if connected is False: if direct_connect is True: self.reset_internal() self.reset_hardware() self._interface.flush() return None # CC2650Device(device_id, master, scan_response) is a slave device # device_id is don't care, because it will be overwrite later dont_care = 0 self._device = ret = CC2650Device(device_id=dont_care, master=self, response_info=response) self.log_info(DEVICE_CONNECTED, address_s) sleep(0.5) print('ret',ret) return ret @synchronized def disconnect(self, device: int, force=False) -> bool: self.log_info(DEVICE_DISCONNECTING, device) ins = bytearray() ins.append(8) ins.append(1) #length ins.append(0xF1) try: # print('send_disconnect',bytes(ins)) self._cc2650.send("bytes", bytes(ins)) except RecvTimeout: self.log_warn('disconnect time out') return False except RuntimeError as e: self.log_warn('suppressed error', str(e)) return False else: sleep(0.01) return True finally: self.log_info(DEVICE_DISCONNECTED, device) # reset single 2650 after disconnected self.reset_internal() self.reset_hardware() self._interface.flush() @synchronized def read_characteristic(self, device: int, handle: int) -> Optional[bytes]: # print("read_characteristic, expect_data_length = ", expect_data_length) ret = None ins = bytearray() ins.append(7) ins.append(2) #length ins.append(handle) ins.append(0xF1) for _ in range(2): try: # print('send_read',bytes(ins)) self._cc2650.send("bytes", bytes(ins)) except SerialTimeoutException: raise RecvTimeout('device CC2650 send read_characteristic fail') try: ret = self._cc2650.recv_uart(timeout = 2) except Exception as e2: raise RecvTimeout() else: # print("======== read ret = ", ret) self._interface.flush_input() # try: # ret = self._cc2650.recv_uart(timeout = 1) # except Exception as e2: # raise RecvTimeout() # else: # # print("read_characteristic ret = ", ret) # pass if ret is None: return None return bytes(ret) @synchronized def write_characteristic(self, device: int, handle: int, data: bytes): # print("device", device) # print("handle", handle) # print("data", data.hex()) # self.log_info('reset') self.log_verbose('write_characteristic', device, handle) self.log_verbose('[CC2650]', 'att_write', str.upper(data.hex())) ack = None ret = None data_array = bytearray() data_array.append(6) data_array.append(len(data)+2) #length data_array.append(handle) data_array.extend(data) data_array.append(0xF1) try: self._cc2650.send("bytes", bytes(data_array)) # print('send_write',bytes(data_array)) except SerialTimeoutException as e: raise RecvTimeout('device CC2650 send instruction fail') from e # read error code try: ret = self._cc2650.recv_uart() # print("_______ write ack = ", ret) except Exception as e2: pass else: pass self._interface.flush_input() self._interface.flush_output() def set_notify(self, device: Union[int, Device], enable: bool): if isinstance(device, CompletedDevice): device = device.device.device_id elif isinstance(device, CC2650Device): device = device.device_id assert isinstance(device, int) if enable: self.log_verbose(device, 'notify enable') message = 'enable' value = b'\01\00' else: self.log_verbose(device, 'notify disable') message = 'disable' value = b'\00\00' try: self.write_characteristic(self._handle, self.NOTIFY_HANDLE, value) except SendInstructionTimeoutError: self.log_warn('set notify', message, 'timeout') raise def central_rf_set(self, rf_power: int) -> bool: rf_set_ins = 0x0A ins = bytearray() ins.append(rf_set_ins) ins.append(rf_power) # len(data_array) should be 21 while len(ins) < 21: ins.append(0) try: self._cc2650.send("bytes", bytes(ins)) except SerialTimeoutException as e: raise RecvTimeout('device CC2650 send instruction fail') from e return False else: return True def central_version_get(self) -> Optional[list]: version_get_ins = 0x09 ins = bytearray() ins.append(version_get_ins) # len(data_array) should be 21 while len(ins) < 21: ins.append(0) try: self._cc2650.send("bytes", bytes(ins)) except SerialTimeoutException as e: raise RecvTimeout('device CC2650 send instruction fail') from e return None try: ret = self._cc2650.recv_uart() except Exception as e2: return None else: # print("central version = ", ret) return ret def handle_command(self, device: Optional[int], command: str, options: Dict[str, str]) -> DeviceCommandAction: if device is not None and command == 'att_find_information': self._handle_command_att_find_information() return DeviceCommandAction.NOTHING return self._raise_command_not_found(device, command) class CC2650MultiMasterCentralDevice(CC2650MasterDevice, Synchronized): def __init__(self, master: MasterDevice, interface: LowLevelHardwareInterface, mem_selector: Selector, selector: Selector, options: CC2650DeviceOptions): super().__init__(master, interface, options, log_name_color=YELLOW) Synchronized.__init__(self) self._selector = selector # self._mem_selector = mem_selector sz = selector.SIZE self._cc2650: List[Optional[CC2650SingleMasterCentralDevice]] = [None for _ in range(sz)] # connected device self._device: List[Optional[CC2650Device]] = [None for _ in range(sz)] self._cc2650_log_level = self.log_level def read_characteristic(self, device: int, handle: int) -> Optional[bytes]: self.log_verbose('read_characteristic', device, '0x%02X' % handle) master = self._cc2650[device] if master is not None: self._interface.flush() self._selector.select(device) # self._mem_selector.select(device) try: return master.read_characteristic(device, handle) except SendInstructionTimeoutError: self.log_warn('device', device, 'send_instruction timeout') raise def write_characteristic(self, device: int, handle: int, data: bytes): master = self._cc2650[device] if master is not None: self._interface.flush() self._selector.select(device) # self._mem_selector.select(device) # print("multiMaster selector = ", device) # print("\n") sleep(0.0001) master.write_characteristic(device, handle, data) def set_notify(self, device: Union[int, Device], enable: bool): if isinstance(device, CompletedDevice): device_id = device.device.device_id elif isinstance(device, CC2650Device): device_id = device.device_id else: device_id = device assert isinstance(device_id, int) master = self._cc2650[device_id] device = self._device[device_id] self.log_verbose('set_notify', device_id, enable) if enable: self.log_verbose(device_id, 'notify enable') message = 'enable' value = b'\01\00' else: self.log_verbose(device_id, 'notify disable') message = 'disable' value = b'\00\00' if master is not None: self._interface.flush() self._selector.select(device_id) # self._mem_selector.select(device_id) # print("multiMaster selector = ", device_id) # print("\n") # device._notify_flag = enable try: master.write_characteristic(device_id, self.NOTIFY_HANDLE, value) except SendInstructionTimeoutError: self.log_warn('set notify', message, 'timeout') raise finally: device._notify_flag = enable else: self.log_warn('cannot %s notify for device : %s' % (message, device.__class__.__name__)) def reset(self, device: Optional[List[int]] = None, software_reset=True): # self.log_verbose('reset') # reset hardware # print("reset device = ", device) for i in range(len(self._cc2650)): if device is None or i in device: self._device[i] = None m = self._cc2650[i] if m is None: m = CC2650SingleMasterCentralDevice(self._master, self._interface, self._options) self._cc2650[i] = m self._interface.flush() self._selector.select(i) try: self.log_info('reset device', i) with m: m.reset_internal() m.reset_hardware() m.reset_software() except RecvTimeout: self.log_warn('reset device', i, 'fail') self._cc2650[i] = None if not software_reset: return def _foreach_empty_master(self) -> Iterable[int]: for i in range(len(self._device)): d = self._device[i] if d is None: yield i def scan_callback(self, callback: Callable[[DeviceResponseInfo], None], timeout=5, all_device=False) -> bool: self.log_verbose('scan') all_result = False for device in range(len(self._cc2650)): self.log_verbose('use', device) m = self._cc2650[device] d = self._device[device] if d is None: error = None with self: self._interface.flush() self._selector.select(device) try: if m.check_mem_survive() == False: continue result = m.scan_callback(callback, timeout=0.001) except RuntimeError as e: error = e else: if result == False: result = True all_result = all_result or result continue all_result = all_result or result if result and not all_device: return True if error is not None: self.log_warn('suppressed error : ' + str(error)) return all_result def found(self) -> List[DeviceResponseInfo]: ret: List[DeviceResponseInfo] = [] def _contain(_found: DeviceResponseInfo): for _response in ret: if _found.match(_response): return True return False for device in self._cc2650: if device is not None: for found in device.found(): if not _contain(found): ret.append(found) return ret def connect(self, response: DeviceResponseInfo, direct_connect: bool = False) -> Device: self.log_verbose('connect') master = None for master in self._foreach_empty_master(): m = self._cc2650[master] d = self._device[master] if direct_connect is True and d is None: break elif response in m.found() and d is None: break else: if master is None: raise RuntimeError('cannot connect any more device') else: raise RuntimeError('rescan please') # self.log_verbose('use', master) m = self._cc2650[master] d = self._device[master] if d is not None: raise RuntimeError('cc2650 has connected a device') with self: self._interface.flush() self._selector.select(master) sleep(0.0001) d = m.connect(response, direct_connect) if d is None: raise RuntimeError('connect failed, retry please') # overwrite attribute d._device_id = master d._master = self self._device[master] = d return d def disconnect(self, device: int, force=False) -> bool: self.log_verbose('disconnect') m = self._cc2650[device] d = self._device[device] if m is None or d is None: return False self.log_verbose('disconnect', device) with self: try: return m.disconnect(device, force=force) finally: self._device[device] = None def get_device(self, device: int) -> Optional[Device]: pass def central_rf_set(self, device: int, rf_power: int) -> bool: m = self._cc2650[device] d = self._device[device] if m is None or d is None: return False self.log_verbose('central_rf_set', device, ": rf power", rf_power) with self: self._interface.flush() self._selector.select(device) try: result = m.central_rf_set(rf_power) except RuntimeError as e: error = e return False else: return result def central_version_get(self, device: int) -> Optional[list]: m = self._cc2650[device] d = self._device[device] if m is None or d is None: return False self.log_verbose('central_version_get', device) with self: self._interface.flush() self._selector.select(device) try: result = m.central_version_get() except RuntimeError as e: error = e return None else: return result