Roy_01
177 lines
5.0 KiB
Python
177 lines
5.0 KiB
Python
import serial
|
|
import time
|
|
|
|
|
|
def survive_cmd(ser):
|
|
send = [0x0A, 0x01, 0xF1]
|
|
ser.reset_input_buffer()
|
|
ser.write(send)
|
|
read = ser.read(4)
|
|
print("survive_cmd")
|
|
print("UART_BT_Tx:\t" + " ".join("%02X" % b for b in send))
|
|
print("UART_BT_Rx:\t" + " ".join("%02X" % b for b in read) + "\n")
|
|
|
|
|
|
def scan_cmd(ser, is_raw_data=True):
|
|
peer_addr = []
|
|
send = [0x03, 0x01, 0xF1]
|
|
ser.reset_input_buffer()
|
|
ser.write(send)
|
|
print("scan_cmd")
|
|
print("UART_BT_Tx:\t" + " ".join("%02X" % b for b in send))
|
|
|
|
for i in range(10):
|
|
hci_packet_event = ser.read(2)
|
|
if int.from_bytes(hci_packet_event, "little") != 0x0004:
|
|
continue
|
|
len = ser.read(1)
|
|
payload = ser.read(int.from_bytes(len, "little"))
|
|
|
|
peer_addr = payload[0:6]
|
|
|
|
if is_raw_data:
|
|
print(
|
|
"UART_BT_Rx:\t"
|
|
+ " ".join("%02X" % b for b in hci_packet_event)
|
|
+ " "
|
|
+ " ".join("%02X" % b for b in len)
|
|
+ " "
|
|
+ " ".join("%02X" % b for b in payload)
|
|
)
|
|
else:
|
|
print("-" * 100)
|
|
print(" peer address: " + ":".join("%02X" % b for b in payload[0:6]))
|
|
print("product module: " + str(payload[6:10], "utf-8"))
|
|
print(" hw version: " + " ".join("%02X" % b for b in payload[10:14]))
|
|
print(" build time: " + " ".join("%02X" % b for b in payload[14:16]))
|
|
print(" Parameter1: " + str(payload[16:19], "utf-8"))
|
|
print(
|
|
" battery volt: "
|
|
+ str(int.from_bytes(payload[19:21], "big") / 1000.0)
|
|
+ "V"
|
|
)
|
|
print(" device name: " + str(payload[21:], "utf-8"))
|
|
print("\n")
|
|
return peer_addr
|
|
|
|
|
|
def connect(ser, peer_addr=[]):
|
|
DISCONNECT_RSPONSE = [0x04, 0x00, 0x01, 0xFF]
|
|
CONNECTED_RSPONSE = [0x04, 0x00, 0x01, 0x03]
|
|
|
|
send = [0x05, 0x08, 0x00]
|
|
send += peer_addr
|
|
send += [0xF1]
|
|
ser.reset_input_buffer()
|
|
ser.write(send)
|
|
read = ser.read(4)
|
|
print("connect " + ":".join(f"{b:02X}" for b in peer_addr))
|
|
print("UART_BT_Tx:\t" + " ".join("%02X" % b for b in send))
|
|
print("UART_BT_Rx:\t" + " ".join("%02X" % b for b in read) + "\n")
|
|
|
|
if list(read) == DISCONNECT_RSPONSE:
|
|
return False
|
|
elif list(read) == CONNECTED_RSPONSE:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
|
|
def disconnect(ser):
|
|
send = [0x08, 0x01, 0xF1]
|
|
ser.reset_input_buffer()
|
|
ser.write(send)
|
|
read = ser.read(4)
|
|
print("disconnect")
|
|
print("UART_BT_Tx:\t" + " ".join("%02X" % b for b in send))
|
|
print("UART_BT_Rx:\t" + " ".join("%02X" % b for b in read) + "\n")
|
|
|
|
|
|
def write_char(ser: serial.Serial, handle: int, write_data: tuple):
|
|
send = [0x06, len(write_data) + 2, handle]
|
|
send += write_data
|
|
send += [0xF1]
|
|
ser.reset_input_buffer()
|
|
ser.write(send)
|
|
read = ser.read(4)
|
|
print("write_char")
|
|
print("UART_BT_Tx:\t" + " ".join("%02X" % b for b in send))
|
|
print("UART_BT_Rx\t" + " ".join("%02X" % b for b in read) + "\n")
|
|
|
|
|
|
def read_char(ser: serial.Serial, handle: int):
|
|
send = [0x07, 0x02, handle, 0xF1]
|
|
ser.reset_input_buffer()
|
|
ser.write(send)
|
|
read = ser.read(0xFFFF)
|
|
print("read_char")
|
|
print("UART_BT_Tx:\t" + " ".join("%02X" % b for b in send))
|
|
print("UART_BT_Rx\t" + " ".join("%02X" % b for b in read) + "\n")
|
|
|
|
|
|
def main():
|
|
ser = serial.Serial("COM14", 115200, 8, inter_byte_timeout=0.01)
|
|
|
|
survive_cmd(ser)
|
|
|
|
peer_addr = scan_cmd(ser, True)
|
|
|
|
# want_connected_addr = bytes.fromhex("C0 38 14 75 AD A9")
|
|
want_connected_addr = bytes.fromhex("C8 DC F1 3B 07 74")
|
|
peer_addr = want_connected_addr
|
|
connect_success = connect(ser, peer_addr)
|
|
time.sleep(1)
|
|
|
|
if connect_success == False:
|
|
print("connect fail, exit python code")
|
|
ser.close()
|
|
return
|
|
|
|
# read version
|
|
write_char(ser, 0x0024, [0x74, 0x40])
|
|
read_char(ser, 0x0021)
|
|
time.sleep(0.1)
|
|
|
|
# -----------------------------------------------------
|
|
# run auto scan mode
|
|
# -----------------------------------------------------
|
|
PEL_RUN_AUTO_SCAN_MODE = False
|
|
if PEL_RUN_AUTO_SCAN_MODE:
|
|
|
|
# event_notify:enable
|
|
write_char(ser, 0x0028, [0x01, 0x00])
|
|
time.sleep(0.1)
|
|
|
|
# PELv3.0 start auto_scan_mode
|
|
write_char(
|
|
ser,
|
|
0x0024,
|
|
[0x35, 0x00, 0x02, 0x01, 0x00, 0x02, 0x00, 0x01, 0x00, 0x05, 0x00, 0x0A],
|
|
)
|
|
|
|
print("---------- notify data ----------")
|
|
time.sleep(0.5)
|
|
print("--------------------------------")
|
|
|
|
# PELv3.0 stop auto_scan_mode
|
|
write_char(ser, 0x0024, [0x30, 0x00, 0x02, 0x00])
|
|
time.sleep(0.1)
|
|
|
|
# event_notify:disable
|
|
write_char(ser, 0x0028, [0x00, 0x00])
|
|
time.sleep(0.1)
|
|
|
|
# auto scan mode completed
|
|
# -----------------------------------------------------
|
|
|
|
disconnect(ser)
|
|
time.sleep(1)
|
|
|
|
ser.close()
|
|
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|