Compare commits

...

1 Commits

Author SHA1 Message Date
Roy e152e56031 improve BLE formate 2023-04-26 16:51:40 +08:00
3 changed files with 25 additions and 26 deletions
+7 -13
View File
@@ -470,6 +470,7 @@ class CC2650Device(Device):
# receive
data = self._master.read_characteristic(self.device_id, CC2650MasterDevice.RETURN_HANDLE)
data = self._decode_data(DeviceInstruction.CIS_VERSION, data)
print('data:',list(data))
except SendInstructionTimeoutError:
self._master.log_warn('device', self.device_id, 'update_device_version no response')
@@ -483,13 +484,9 @@ class CC2650Device(Device):
day = struct.unpack('<B', data[4:5])[0]
hour = struct.unpack('<B', data[5:6])[0]
minute = struct.unpack('<B', data[6:7])[0]
mac1 = struct.unpack('<B', data[7:8])[0]
mac2 = struct.unpack('<B', data[8:9])[0]
mac1 = "%02X" % mac1
mac2 = "%02X" % mac2
self._device_version = str(year) + '/' + str(month) + '/' + str(day) + " " + str(hour) + ":" + str(
minute) + " | " + str(mac1) + ":" + str(mac2)
minute)
@property
def battery(self) -> int:
@@ -2134,7 +2131,6 @@ class CC2650SingleMasterCentralDevice(CC2650MasterDevice, Synchronized):
addr_type = response.addr_type
address_s = cc2650.address_str(address)
connected = False
connect_response = None
ins = bytearray()
@@ -2162,6 +2158,7 @@ class CC2650SingleMasterCentralDevice(CC2650MasterDevice, Synchronized):
else:
try:
connect_response = self._cc2650.recv_uart(2)
print('!!connect_response:', list(connect_response))
except RecvTimeout:
self.log_verbose('[CC2650]', 'connect response timeout')
if retry < 5:
@@ -2169,27 +2166,24 @@ class CC2650SingleMasterCentralDevice(CC2650MasterDevice, Synchronized):
continue
else:
break
if connect_response is None:
return False
pack_len = connect_response[0]
connect_ack = connect_response[1:pack_len+1]
if connect_ack[0] == 46 and connect_ack[1] == 80 and \
connect_ack[2] == 48 and connect_ack[3] == 4:
connected = True
if connect_ack == [3]:
connect_response_hex = ''.join(format(i, '02X') for i in connect_response)
self.log_verbose('[CC2650]', 'connect success', '0x'+connect_response_hex)
if connected == True:
# 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_verbose('[CC2650]', DEVICE_CONNECTED, address_s)
return ret
self.log_verbose('[CC2650]', 'connect fail')
return False
+16 -11
View File
@@ -372,6 +372,7 @@ class InternalInstruction(SingleInstruction):
else:
data = [v for v in data]
print('cdr data:', data)
parser.parse_instruction(context, data)
@@ -776,6 +777,12 @@ class InstructionDataContent(InstructionContent):
def parse_instruction(self, context: Scope, buffer: List[int], offset: int = 0, shift: int = 0) -> Tuple[int, int]:
print('self._width', self._width)
print('self._width.size', self._width.size)
for _ in range (self._width.size - len(buffer)):
buffer.append(0)
if self._width.is_array:
raise NotImplementedError()
@@ -789,21 +796,20 @@ class InstructionDataContent(InstructionContent):
value = 0
if self._width.size == 1:
value = buffer[offset]
for i in range(self._width.size):
if self._width.little_endian:
value |= (buffer[offset] << (1 * i))
else:
value = (value << 8) | buffer[offset]
offset += 1
else:
for i in range(self._width.size):
if self._width.little_endian:
value |= (buffer[offset] << (1 * i))
else:
value = (value << 8) | buffer[offset]
offset += 1
if len(self._value) > 0:
context[self._value] = value
print('value',value)
print('hex(value)',hex(value))
return offset, 0
else:
@@ -821,7 +827,6 @@ class InstructionDataContent(InstructionContent):
if self._width.bytes_unit:
if shift != 0:
offset += 1
return offset + self._width.size, 0
else:
+2 -2
View File
@@ -3708,8 +3708,8 @@ class CC2650Central(LoggerFlag):
def _recv_byte(self) -> Optional[int]:
ret = self._recv_bytes(1)
# if ret is not None:
# print('packet = {0}'.format(hex(ret[0]).upper()))
if ret is not None:
print('packet = {0}'.format(hex(ret[0]).upper()))
return ret[0] if ret is not None else None
def _recv_bytes(self, size: int = 1) -> Union[None, bytes]: