kunlun_ramtool/kunlun.py

#!/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
import threading
import argparse
import base64
from bin.log import myprint
from bin.log import log_init
from bin.log import mywrite
from bin.bin_to_hex import bin_to_hex_file
from bin.bin_to_hex import bin_file_decrypt
from bin.bin_to_hex import clear_tmp
from bin.bin_to_hex import load_flash_info
from bin.bin_to_hex import check_bin_type
from bin.base import bin_path
from bin.factory_mode import ftm_handle
from bin.crc import CRC16
from bin.get_log import print_log
from password import bootram_pssword


def init_send(s_port:serial.Serial, send_str:str):
    s_info = bytearray()
    send_str=send_str.encode('utf-8')
    key_words={
        b"WQKL":b"Recieving RAM-IMAGE in xmodem : C",
        b"HZPG":b"waiting receive image: C"
    }
    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(key_words[send_str])>=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) or (s_info.find(b'Recieving FLASH-IMAGE in xmodem : C')>=0):
            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=data.replace(b"\r\n",b"\n")
    index=0
    while index<len(data):
        if(bytes([data[index]]).isascii()):
            index+=1
        else:
            break
    try:
        d=data[:index].decode('utf-8')
        mywrite(d)
        if(index<len(data)):
            mywrite(f"{data[index:]}")
    except Exception as e:
        mywrite(f"{data}\n")



_recv_thread=False
# 读取input
def read_input(s_port:serial.Serial):
    global _recv_thread
    def read():
        while True:
            try:
                d=input()
            except KeyboardInterrupt:
                break
            if(d=='exit'):
                break
            try:
                s_port.write(d.encode('utf-8')+b'\n')
            except Exception:
                break
    read()
    _recv_thread=False

def recv_ser_end():
    global _recv_thread
    _recv_thread=False
# 循环接收串口数据
def recv_ser_data(s_port:serial.Serial):
    global _recv_thread
    def recv(s_port:serial.Serial):
        global _recv_thread
        while _recv_thread:
            time.sleep(0.1)
            bytes2read = s_port.in_waiting
            if(bytes2read>0):
                tmp = s_port.read(bytes2read)
                try:
                    txt=tmp.decode('utf-8')
                    mywrite(txt)
                except Exception as e:
                    pass
    _recv_thread=True
    t=threading.Thread(target=recv,args=(s_port,))
    t.start()

# 上传固件
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 recv_ser_data_ftm(s_port:serial.Serial,timeout:float=1):
    recv_data=bytearray()
    while timeout>0:
        time.sleep(0.1)
        timeout-=0.1
        bytes2read = s_port.in_waiting
        if(bytes2read>0):
            tmp = s_port.read(bytes2read)
            recv_data+=tmp
    ftm_handle(recv_data)


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(log_file)[-1]
    file=name+'.log'
    return os.path.join(_work_dir,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)



# 计算bootram密码
def calc_bootram_pssword(info_file:str):
    with open(info_file,mode='r',encoding='utf-8') as f:
        lines=f.readlines()
    for line in lines:
        if(line.startswith('Chip Id')):
            chip_id=line.split(':')[-1].strip()
    password=bootram_pssword(chip_id)
    return f"{password:08x}"


# 上传固件
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'))
    # 输入密码
    password=calc_bootram_pssword(calc_flash_info_name())
    ser.write(f"PSS {password}\n".encode("utf-8"))
    time.sleep(0.5)
    ser_read_data=ser.read(4096)
    print_device_str(ser_read_data)
    # 显示image信息
    ser.write(b"f s\n")
    time.sleep(1)
    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(f"fw u d {get_upload_cfg(upload_key)}\n".encode("utf-8"))
    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(),layout_index)
    myprint("Transform to hex end.")



# 发送配置信息
def send_cfg_fun():
    global nb_rate
    rate_table={
        115200:0,
        460800:1,
        1500000:2
    }
    baud_cfg=f"u_baud={rate_table.get(nb_rate,0)}"
    cfg_str=base64.b64encode(baud_cfg.encode(encoding='utf-8')).decode(encoding='utf-8')
    text=f"ram_config:{cfg_str}:gifnoc_mar"
    crc=CRC16(text.encode('utf-8'))
    cfg_text=f"{text}#crc={crc}"
    # print(cfg_text)
    ser.write(f"{cfg_text}\n".encode("utf-8"))



# 烧录固件
def burn_fun():
    # time.sleep(1)
    send_cfg_fun()
    # time.sleep(0.02)
    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 get_upload_cfg(key:str):
    if(key=='all'):
        return key
    info_list=load_flash_info(calc_flash_info_name())
    # print(info_list)
    for item in info_list:
        index=item["ImgType"].find(key.upper())
        if(index>=0):
            return f"{item['Offset']} {item['Size']}"
    myprint(f'Can not found key:{key}')
    return "all"


# 列出所有 .bin 文件
def list_bin_file():
    path=os.path.curdir
    items:list[str]=[]
    l=os.listdir(path)
    for item in l:
        if(os.path.isfile(item)):
            items.append(item)
    l=os.listdir(bin_path())
    for item in l:
        tmp=os.path.join(bin_path(),item)
        if(os.path.isfile(tmp)):
            items.append(item)
    ret_list:list[str]=[]
    for item in items:
        if(item.endswith(".bin")):
            ret_list.append(item)
    return ret_list

# 获取ram文件的路径
def ram_file_redirect(file:str):
    if os.path.exists(file):
        return file
    else:
        p=os.path.join(bin_path(),file)
        if os.path.exists(p):
            return p
    myprint(f"文件不存在 {file}")
    sys.exit(-1)

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 function_type
    global upload_key
    global parser
    global user_log
    global log_timeout
    global layout_index
    global log_file
    global skip_ram
    global extract_log
    args = parser.parse_args()

    init_str="WQKL"
    serial_com=args.port
    b_rate=args.b_rate
    nb_rate=args.nb_rate
    user_log=args.log
    log_timeout=args.timeout
    layout_index=args.layout_index
    skip_ram=args.skip_ram
    extract_log=args.extract_log


    # 上传或者下载flash镜像
    if(args.flash_file is not None):
        iot_flash_file=args.flash_file
        log_file=iot_flash_file
        if(args.ram_file is not None):
            ram_file=ram_file_redirect(args.ram_file)
        if(os.path.exists(iot_flash_file)):
            function_type='download'
            iot_flash_file=bin_file_decrypt(iot_flash_file)
            if args.kunlun_version is None:
                args.kunlun_version=check_bin_type(iot_flash_file)
            if ram_file is None:
                ram_file=f'kl{args.kunlun_version}_ram_build.bin'
                ram_file=ram_file_redirect(ram_file)
        else:
            function_type='upload'
            if ram_file is None:
                ram_file=f'bootram_kl{args.kunlun_version}.bin'
                ram_file=ram_file_redirect(ram_file)
        if(args.kunlun_version is None):
            print("请指定参数 -k")
            sys.exit(-1)
        elif(args.kunlun_version == "4"):
            init_str="HZPG"
        if(args.upload_key is not None):
            upload_key = args.upload_key
        else:
            upload_key='all'
    elif(args.ram_file is not None):
        function_type='ram'
        ram_file=ram_file_redirect(args.ram_file)
        log_file=args.ram_file
        if(args.kunlun_version is None):
            print("请指定参数 -k")
            sys.exit(-1)
        elif(args.kunlun_version == "4"):
            init_str="HZPG"
    elif(args.bin_convert is not None):
        function_type='convert'
        iot_flash_file=args.bin_convert
        log_file=iot_flash_file
    elif(args.console is not None):
        log_file=args.console
        function_type='console'
    elif(args.ftm):
        log_file=time.strftime("%Y%m%d-%H%M%S")
        function_type='ftm'
    elif(args.list):
        log_file=time.strftime("%Y%m%d-%H%M%S")
        function_type='list'




def parser_init():
    global parser
    parser = argparse.ArgumentParser(description='kunlun相关工具脚本')
    # 添加参数
    parser.add_argument('-p','--port',action='store', help='输入使用的串口')
    parser.add_argument('-r', '--ram_file', action='store', help='输入要下载的ram文件名称，此项和 -f 选项不能同时指定')
    parser.add_argument('-f', '--flash_file', action='store', help='输入要下载或上传的flash文件名称，如果文件存在则下载；不存在则上传')
    parser.add_argument('-c', '--console', action='store', help='指定控制台保存的文件名，不指定则不进入控制台')
    parser.add_argument('-u','--upload_key',action='store',help='上传时保存分区的关键字，不指定则上传所有')
    parser.add_argument('-k','--kunlun_version',action='store',help='上传下载时的kunlun版本 3或者1')
    parser.add_argument('-b','--bin_convert',action='store',help='转换bin文件')
    parser.add_argument('-l','--log',action='store_true',default=False, help='下载ram或flash程序之后是否进入log界面，默认否')
    parser.add_argument('-t','--timeout',action='store',type=float,default=5,help='启用log时接收log的时间')
    parser.add_argument('-i','--layout_index',action='store',type=int,help='解析接收到的flash文件或转换bin文件时使用的layout，不指定则不解析')
    parser.add_argument('--b_rate',action='store',type=int,default=115200,help='下载ram程序 上传flash数据 控制台 接收log等 使用的串口波特率')
    parser.add_argument('--nb_rate',action='store',type=int,default=1500000,help='下载flash程序使用的串口波特率，支持115200,460800,1500000')
    parser.add_argument('--ftm',action='store_true',default=False,help='进入工厂模式')
    parser.add_argument('--list',action='store_true',default=False,help='列出所有.bin文件')
    parser.add_argument('--skip_ram',action='store_true',default=False,help='下载flash文件时是否跳过下载ram.bin的步骤，默认否')
    parser.add_argument('--extract_log',action='store_true',default=False,help='是否提取bin中的log信息，默认否')





def print_help():
    help=f'''
    输入:
        {sys.argv[0]} -h
    查看帮助
    '''
    print(help)

if __name__ == '__main__':
    function_type=None # 功能类型 "download" 下载，"upload" 上传，"ram" 下载ram，"convert"" 转换bin
    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
    upload_key, parser, layout_index, log_file, extract_log=None, None, None, None, False
    user_log=False
    log_timeout=0
    skip_ram=False
    parser_init()
    global_def()
    if(function_type is None):
        print("param err.")
        print_help()
        sys.exit(-1)
    log_init(calc_log_file_name())
    if(function_type=='convert'):
        bin_to_hex_file(iot_flash_file,calc_hex_name(),layout_index)
        if(extract_log):
            print_log(iot_flash_file)
        sys.exit(0)
    if(function_type=="list"):
        l=list_bin_file()
        myprint("当前目录和bin目录的.bin文件如下")
        for item in l:
            myprint(item)
        sys.exit(0)
    if(serial_com is None):
        myprint(f"请指定串口 {serial_com}")
        sys.exit(-1)
    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)

    if(function_type=='console'):
        recv_ser_data(ser)
        read_input(ser)
        sys.exit(0)
    if(function_type=="ftm"):
        myprint("此功能尚未实现")
        sys.exit(0)

    modem = xmodem.XMODEM(getc, putc, mode='xmodem1k')
    # 发送启动字符让设备进入xmodem模式
    if not skip_ram:
        init_send(ser, init_str)
        burn_ram_bin(modem, ram_file)

    if(function_type=='upload'):
        upload_fun()
        if(extract_log):
            print_log(calc_upload_name())
    elif(function_type=='download'):
        burn_fun()
        clear_tmp()
        if(user_log):
            ser.baudrate=b_rate
            recv_ser_data(ser)
            time.sleep(log_timeout)
            recv_ser_end()
    elif(function_type=='ram'):
        if(user_log):
            ser.baudrate=b_rate
            recv_ser_data(ser)
            time.sleep(log_timeout)
            recv_ser_end()
        else:
            ser.baudrate=b_rate
            recv_ser_data(ser)
            read_input(ser)

# if __name__ == "__main__":
#     nb_rate=115200
#     send_cfg_fun()