298 lines
8.1 KiB
Python
298 lines
8.1 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: UTF-8 -*-
|
|
|
|
import datetime
|
|
import fileinput
|
|
import re
|
|
import sys
|
|
import serial
|
|
import time
|
|
import xmodem
|
|
import os
|
|
import binascii
|
|
from log import myprint
|
|
from log import log_init
|
|
from bin_to_hex import bin_to_hex_file
|
|
from bin_to_hex import bin_file_decrypt
|
|
from bin_to_hex import clear_tmp
|
|
|
|
|
|
def init_send(s_port:serial.Serial, send_str:str):
|
|
s_info = bytearray()
|
|
send_str=send_str.encode('utf-8')
|
|
while True:
|
|
s_port.write(send_str)
|
|
time.sleep(0.1)
|
|
s_info += s_port.read(1)
|
|
bytes2read = s_port.in_waiting
|
|
tmp = s_port.read(bytes2read)
|
|
s_info += tmp
|
|
if(s_info.find(b"Recieving RAM-IMAGE in xmodem : C")>=0):
|
|
m_ram=True
|
|
else:
|
|
m_ram=False
|
|
if m_ram:
|
|
myprint ("Program enters transmission mode...")
|
|
break
|
|
|
|
|
|
def burn_ram_bin(x_modem:xmodem.XMODEM, r_file):
|
|
global time_stamp_end
|
|
stime = datetime.datetime.now()
|
|
try:
|
|
stream = open(r_file, 'rb')
|
|
except Exception:
|
|
myprint(f"Cannot load file {r_file}")
|
|
sys.exit(-1)
|
|
myprint (f"Transferring {r_file}...")
|
|
|
|
xmodem_send = x_modem.send(stream, callback=download_callback)
|
|
myprint('.')
|
|
etime = datetime.datetime.now()
|
|
time_stamp_end = (etime - stime).seconds
|
|
myprint (f"Transferring {r_file} result: {xmodem_send}, consuming time: {(time_stamp_end- time_stamp_start)} s ")
|
|
|
|
def burn_flash_bin(s_port:serial.Serial, x_modem:xmodem.XMODEM, f_file):
|
|
global time_stamp_end
|
|
s_info = bytearray()
|
|
while True:
|
|
bytes2read = s_port.in_waiting
|
|
tmp = s_port.read(bytes2read)
|
|
if(len(tmp)==0):
|
|
continue
|
|
print_device_str(tmp)
|
|
s_info += tmp
|
|
if(s_info.find(b'CCC')>=0 and bytes2read==1):
|
|
m_flash=True
|
|
else:
|
|
m_flash=False
|
|
if(s_info.find(b"Updating done, PLS reboot the device...")>=0):
|
|
m_done=True
|
|
else:
|
|
m_done=False
|
|
|
|
if m_flash:
|
|
s_info = bytearray()
|
|
stime = datetime.datetime.now()
|
|
try:
|
|
stream = open(f_file, 'rb')
|
|
except Exception:
|
|
myprint(f"Cannot load file {f_file}")
|
|
sys.exit(-1)
|
|
myprint (f"Transferring {f_file}...")
|
|
|
|
xmodem_send = x_modem.send(stream, quiet=True, callback=download_callback,retry=16)
|
|
myprint('.')
|
|
etime = datetime.datetime.now()
|
|
time_stamp_end = (etime - stime).seconds
|
|
myprint (f"Transferring {f_file} result: {xmodem_send}, consuming time: {(time_stamp_end-time_stamp_start)} s ")
|
|
if(xmodem_send is False):
|
|
break
|
|
elif m_done:
|
|
myprint("Update done.")
|
|
break
|
|
else:
|
|
pass
|
|
|
|
|
|
def upload_callback(total_packets, success_count, error_count, packet_size):
|
|
if total_packets % 10 == 0:
|
|
sys.stdout.write('.')
|
|
sys.stdout.flush()
|
|
|
|
|
|
def download_callback(total_packets, success_count, error_count):
|
|
if total_packets % 10 == 0:
|
|
sys.stdout.write('.')
|
|
sys.stdout.flush()
|
|
|
|
# 打印串口收到的字符
|
|
def print_device_str(data:bytearray):
|
|
data:list[bytearray]=data.split(b"\r\n")
|
|
for item in data:
|
|
try:
|
|
d=item.decode('utf-8')
|
|
myprint("DEVICE:",d.strip())
|
|
except Exception as e:
|
|
myprint("DEVICE:",item)
|
|
|
|
# 上传固件
|
|
def upload_bin(x_modem:xmodem.XMODEM, w_file):
|
|
global time_stamp_end
|
|
stime = datetime.datetime.now()
|
|
try:
|
|
stream = open(w_file, 'wb+')
|
|
except Exception:
|
|
myprint(f"Cannot cteate file {w_file}")
|
|
sys.exit(-1)
|
|
myprint (f"Receiving {w_file}..." )
|
|
|
|
xmodem_send = x_modem.recv(stream, callback=upload_callback)
|
|
myprint('.')
|
|
etime = datetime.datetime.now()
|
|
time_stamp_end = (etime - stime).seconds
|
|
myprint (f"Receiving {w_file} result: {xmodem_send}, consuming time: {(time_stamp_end-time_stamp_start)} s ")
|
|
|
|
|
|
|
|
|
|
def getc(size, timeout=1):
|
|
data=ser.read(size)
|
|
return data or None
|
|
|
|
|
|
def putc(data, timeout=1):
|
|
ser.write(data)
|
|
time.sleep(0.03)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_work_dir='work'
|
|
if not os.path.exists(_work_dir):
|
|
os.mkdir(_work_dir)
|
|
|
|
|
|
# 日志文件 固定存放在work目录
|
|
def calc_log_file_name():
|
|
name=os.path.split(iot_flash_file)[-1]
|
|
log_file=name+'.log'
|
|
return os.path.join(_work_dir,log_file)
|
|
|
|
# 上传的flash镜像 固定存放在work目录
|
|
def calc_upload_name():
|
|
name=os.path.split(iot_flash_file)[-1]
|
|
return os.path.join(_work_dir,name)
|
|
|
|
# 根据flash镜像生成的hex文件 固定存放在work目录
|
|
def calc_hex_name():
|
|
name=os.path.split(iot_flash_file)[-1]
|
|
name=name+'.txt'
|
|
return os.path.join(_work_dir,name)
|
|
|
|
# 上传的flash信息 固定存放在work目录
|
|
def calc_flash_info_name():
|
|
name=os.path.split(iot_flash_file)[-1]
|
|
name=name+'.info'
|
|
return os.path.join(_work_dir,name)
|
|
|
|
|
|
|
|
|
|
# 上传固件
|
|
def upload_fun():
|
|
# 显示flash信息
|
|
ser.write(b"f i\n")
|
|
time.sleep(0.5)
|
|
ser_read_data=ser.read(4096)
|
|
print_device_str(ser_read_data)
|
|
with open(calc_flash_info_name(),mode='w+',encoding='utf-8') as f:
|
|
f.writelines(ser_read_data.decode('utf-8').split('\r\n'))
|
|
# 显示image信息
|
|
ser.write(b"f s\n")
|
|
time.sleep(0.5)
|
|
ser_read_data=ser.read(4096)
|
|
print_device_str(ser_read_data)
|
|
with open(calc_flash_info_name(),mode='a+',encoding='utf-8') as f:
|
|
f.writelines(ser_read_data.decode('utf-8').split('\r\n'))
|
|
# 上传整个镜像
|
|
ser.write(b"fw u d all\n")
|
|
time.sleep(0.5)
|
|
myprint(ser.read(4096).decode('utf-8'))
|
|
upload_bin(modem,calc_upload_name())
|
|
|
|
myprint (f"Total transmission time: {(time_stamp_start+time_stamp_end)} s" )
|
|
myprint("Start transform bin to hex.")
|
|
bin_to_hex_file(calc_upload_name(), calc_hex_name())
|
|
myprint("Transform to hex end.")
|
|
|
|
|
|
# 烧录固件
|
|
def burn_fun():
|
|
time.sleep(3)
|
|
ser_read_data=ser.read(4096)
|
|
print_device_str(ser_read_data)
|
|
ser.baudrate=nb_rate
|
|
|
|
burn_flash_bin(ser, modem, iot_flash_file)
|
|
myprint (f"Total transmission time: {(time_stamp_start+time_stamp_end)} s")
|
|
time.sleep(3)
|
|
ser_read_data=ser.read(4096)
|
|
print_device_str(ser_read_data)
|
|
|
|
|
|
|
|
def global_def():
|
|
global init_str
|
|
global serial_com
|
|
global b_rate
|
|
global nb_rate
|
|
global ram_file
|
|
global iot_flash_file
|
|
global ser
|
|
global time_stamp_start
|
|
global time_stamp_end
|
|
global upload
|
|
init_str="WQKL"
|
|
serial_com=sys.argv[1]
|
|
b_rate=115200
|
|
nb_rate=1500000
|
|
|
|
|
|
if(sys.argv[3]=='upload'):
|
|
upload=True
|
|
elif(sys.argv[3]=='download'):
|
|
upload=False
|
|
else:
|
|
if(sys.argv[3].endswith('.bin')):
|
|
if(os.path.exists(sys.argv[3])):
|
|
# 存在这个文件 是下载
|
|
upload=False
|
|
else:
|
|
# 不存在这个文件 是上传
|
|
upload=True
|
|
else:
|
|
print("param err.")
|
|
exit(-1)
|
|
if(upload):
|
|
ram_file=f'bootram_{sys.argv[2]}.bin'
|
|
else:
|
|
ram_file=f'{sys.argv[2]}_ram_build.bin'
|
|
ram_file=os.path.join('bin',ram_file)
|
|
iot_flash_file=sys.argv[-1]
|
|
|
|
|
|
|
|
|
|
# 如果不指定上传还是下载 脚本会根据输入文件是否存在来决定上传还是下载
|
|
# kunlun.py [com] [kl1/kl3] [upload.bin/download.bin]
|
|
# kunlun.py [com] [kl1/kl3] [upload/download] [upload.bin/download.bin]
|
|
if __name__ == '__main__':
|
|
if(len(sys.argv)<4):
|
|
print("param too less")
|
|
exit(-1)
|
|
upload=None
|
|
time_stamp_start, time_stamp_end = 0, 0
|
|
init_str, serial_com, b_rate, nb_rate, ram_file, iot_flash_file, ser = None, None, None, None, None, None, None
|
|
global_def()
|
|
log_init(calc_log_file_name())
|
|
try:
|
|
ser = serial.Serial(port=serial_com, baudrate=b_rate, timeout=0.3)
|
|
except Exception:
|
|
myprint(f"serial com:{serial_com} open failed.")
|
|
sys.exit(-1)
|
|
modem = xmodem.XMODEM(getc, putc, mode='xmodem1k')
|
|
|
|
# 发送启动字符让设备进入xmodem模式
|
|
init_send(ser, init_str)
|
|
burn_ram_bin(modem, ram_file)
|
|
|
|
if(upload):
|
|
upload_fun()
|
|
else:
|
|
iot_flash_file=bin_file_decrypt(iot_flash_file)
|
|
burn_fun()
|
|
clear_tmp()
|